Function Call 的原理及实践

[
  {
    "type": "function",
    "function": {
      "name": "get_current_weather",
      "description": "获取给定位置的当前天气",
      "parameters": {
        "type": "object",
        "properties": {
          "location": {
            "type": "string",
            "description": "城市或区，例如北京、海淀"
          }
        },
        "required": ["location"]
      }
    }
  }
]

[
  {
    "role": "system",
    "content": "你是一个天气播报小助手，你需要根据用户提供的地址来回答当地的天气情况，如果用户提供的问题具有不确定性，不要自己编造内容，提示用户明确输入"
  },
  {
    "role": "user",
    "content": "今天北京的天气如何"
  }
]

Completion(
    model='glm-4',
    created=1765805092,
    choices=[CompletionChoice(
        index=0,
        finish_reason='tool_calls',          # ← 关键：模型因「要调工具」而结束，不是因为答完了
        message=CompletionMessage(
            content=None,                    # ← 没有文本内容，模型这一轮压根没打算回答
            role='assistant',
            reasoning_content=None,
            tool_calls=[                     # ← 这就是「委托单」
                CompletionMessageToolCall(
                    id='call_-8074878091637713008',   # 委托单编号，回执要靠它配对
                    function=Function(
                        arguments='{"location": "北京"}',   # 参数：字符串形式的 JSON
                        name='get_current_weather'         # 函数名：要调哪个函数
                    ),
                    type='function',
                    index=0
                )
            ]
        )
    )],
    request_id='2025121521245166e1ccfc510a4af9',
    id='2025121521245166e1ccfc510a4af9',
    usage=CompletionUsage(prompt_tokens=155, completion_tokens=11, total_tokens=166)
)

[
  { "role": "system", "content": "你是一个天气播报小助手……" },

  { "role": "user", "content": "今天北京的天气如何" },

  {
    "role": "assistant",
    "content": null,
    "tool_calls": [
      {
        "id": "call_-8074878091637713008",
        "type": "function",
        "function": {
          "name": "get_current_weather",
          "arguments": "{\"location\": \"北京\"}"
        },
        "index": 0
      }
    ]
  },

  {
    "role": "tool",
    "tool_call_id": "call_-8074878091637713008",
    "name": "get_current_weather",
    "content": "{\"location\": \"北京\", \"high_temperature\": \"高温 33℃\", \"low_temperature\": \"低温 17℃\", \"week\": \"星期一\", \"type\": \"晴\"}"
  }
]

# -*- coding: utf-8 -*-
"""
Function Call 最小骨架（单函数版）
==========================================
天气案例去掉「天气」之后，剩下的就是这 60 行。
只有一个工具、固定两轮对话，是理解 Function Call 的最短路径。
需要多个工具或函数间依赖，请用 fc_skeleton.py（带注册表 + while 多轮）。
"""

import json
import os

# TODO(拓展点 1)：换 client / 换模型
from zhipuai import ZhipuAI

client = ZhipuAI(api_key=os.environ["ZHIPUAI_API_KEY"])
MODEL = "glm-4"
# openai 包写法（只差 import 和 base_url，下面代码一字不改）：
# from openai import OpenAI
# client = OpenAI(api_key=os.environ["OPENAI_API_KEY"],
#                 base_url="https://open.bigmodel.cn/api/paas/v4/")


# TODO(拓展点 2)：换成你的真实函数（模型不执行它，执行的是你）
def my_function(param1):
    """示例：查天气、查订单、算运费……换成你的业务逻辑。"""
    return {"param1": param1, "result": "这里换成真实结果"}


# TODO(拓展点 3)：函数描述——name 必须和上面的函数名一致
tools = [{
    "type": "function",
    "function": {
        "name": "my_function",
        "description": "这个函数是干什么的、什么时候该调用它",
        "parameters": {
            "type": "object",
            "properties": {
                "param1": {"type": "string", "description": "参数含义，写清格式和示例"},
            },
            "required": ["param1"],
        },
    },
}]


def run(user_input):
    messages = [{"role": "user", "content": user_input}]

    # 第一次调用：模型决定要不要用工具
    first = client.chat.completions.create(
        model=MODEL, messages=messages, tools=tools, tool_choice="auto")
    msg = first.choices[0].message

    if not getattr(msg, "tool_calls", None):
        return msg.content                        # 模型觉得不用工具，直接回答

    tool_call = msg.tool_calls[0]
    args = json.loads(tool_call.function.arguments)   # arguments 是字符串 JSON
    result = my_function(**args)                      # 本地执行

    messages.append(msg.model_dump())             # 回填 assistant（带 tool_calls）
    messages.append({                             # 回填 tool（函数回执）
        "role": "tool",
        "tool_call_id": tool_call.id,
        "name": tool_call.function.name,
        "content": json.dumps(result, ensure_ascii=False),   # 必须是字符串
    })

    # 第二次调用：模型拿着真实结果，翻译成人话
    second = client.chat.completions.create(
        model=MODEL, messages=messages, tools=tools, tool_choice="auto")
    return second.choices[0].message.content


if __name__ == "__main__":
    # TODO(拓展点 4)：换成你的提问
    print(run("这里换成用户的问题"))

"""
单一函数应用（天气）—— 工具层：函数定义 + 函数描述 tools + 调用结果解析
文件路径：./ChatGLM3_FunctionCall/weather/tools.py
"""
import json                      # 解析模型返回的 arguments（字符串形式的 JSON）
import requests                  # 请求第三方天气接口

# ============ 1. 函数描述 tools：写给「模型」看的说明书 ============
# 注意：tools 不是代码逻辑，它只是一份 JSON Schema 清单，
#      模型读了它才知道「我手上有哪些函数可以委托、每个函数要填哪些参数」。
tools = [
    {
        "type": "function",                       # 工具类型，目前固定为 function
        "function": {
            # 函数名：必须与下方真实的 Python 函数同名，否则派发时会 KeyError
            "name": "get_current_weather",
            # 函数描述：模型「选不选这个函数」完全依据这句话，要写清楚用途
            "description": "获取给定位置的当前天气",
            "parameters": {                       # 参数定义，遵循 JSON Schema
                "type": "object",                 # 参数整体是一个对象，固定写法
                "properties": {                   # 对象里的各个字段
                    "location": {                 # 参数名，要与函数形参一致
                        "type": "string",         # 参数类型
                        # 参数描述：模型据此从用户原话里抽取出「北京」这个值
                        "description": "城市或区，例如北京、海淀",
                    },
                },
                "required": ["location"],         # 必填参数；缺失时模型会反问用户而不是瞎编
            },
        }
    }
]


# ============ 2. 真正干活的本地函数：查询天气 ============
def get_current_weather(location):
    """得到给定地址的当前天气信息（真正执行方是本函数，不是模型）"""
    # 2.1 读取「城市 -> 邮政编码」映射表，天气接口用编码而不是中文城市名
    with open('./cityCode_use.json', 'r', encoding='utf-8') as file:
        data = json.load(file)                    # json.load 直接把文件流解析成 Python 列表

    city_code = ""                                # 城市编码，默认空
    weather_info = {}                             # 查询结果，默认空字典

    # 2.2 遍历映射表，找到与 location 同名的城市，取出它的编码
    for loc in data:
        if location == loc["市名"]:
            city_code = loc["编码"]

    # 2.3 拿到编码才发起请求；没匹配到就直接返回空结果，避免拼出错误 URL
    if city_code:
        weather_url = "http://t.weather.itboy.net/api/weather/city/" + city_code
        response = requests.get(weather_url)      # 发起 HTTP GET 请求
        result1 = eval(response.text)             # 示例写法：把返回文本转成字典
        # 更稳的写法是 result1 = response.json()，eval 会执行任意表达式，生产环境不要用

        forecast = result1["data"]["forecast"][0]  # forecast[0] 就是「今天」的预报
        # 2.4 只挑模型需要的字段返回，喂给模型的内容越干净，最终回答越准
        weather_info = {
            "location": location,
            "high_temperature": forecast["high"],
            "low_temperature": forecast["low"],
            "week": forecast["week"],
            "type": forecast["type"],
        }

    # 2.5 必须返回字符串：messages 里 role 为 tool 的 content 只接受字符串
    #     ensure_ascii=False 保证中文不被转成 \uXXXX 乱码
    return json.dumps(weather_info, ensure_ascii=False)


# ============ 3. 解析模型回复：判断要不要调函数，并把函数调起来 ============
def parse_response(response):
    """根据模型回复决定是否调用工具函数；若调用，返回函数执行结果"""
    response_message = response.choices[0].message

    # 3.1 关键判断：模型回复里有 tool_calls，说明它填了「委托单」
    #     没有 tool_calls，说明模型认为直接用文本回答就够了
    if response_message.tool_calls:
        # 3.2 函数名 -> 函数对象 的派发表；多函数场景往这里继续加即可
        available_functions = {
            "get_current_weather": get_current_weather,
        }
        # 3.3 从模型回复中取出函数名（模型只告诉你「叫什么」）
        function_name = response_message.tool_calls[0].function.name
        fuction_to_call = available_functions[function_name]

        # 3.4 arguments 是「字符串形式的 JSON」，必须 json.loads 才能当参数用
        function_args = json.loads(response_message.tool_calls[0].function.arguments)

        # 3.5 真正执行函数 —— 这一行是整个 Function Call 里唯一「执行」的地方
        function_response = fuction_to_call(
            location=function_args.get("location"),
        )
        return function_response


if __name__ == '__main__':
    # 单独测试本地函数是否可用（不经过模型，先保证函数本身跑得通）
    function_name = get_current_weather
    response = function_name(location="北京")
    print(response)

[
  {"市名": "北京", "编码": "101010100"},
  {"市名": "海淀", "编码": "101010200"},
  {"市名": "朝阳", "编码": "101010300"},
  {"市名": "上海", "编码": "101020100"},
  {"市名": "天津", "编码": "101030100"},
  {"市名": "郑州", "编码": "101180101"},
  {"市名": "深圳", "编码": "101280601"},
  {"市名": "广州", "编码": "101280101"},
  {"市名": "杭州", "编码": "101210101"},
  {"市名": "成都", "编码": "101270101"}
]

"""
单一函数应用（天气）—— 主逻辑：两次调用模型，中间夹一次本地函数执行
文件路径：./ChatGLM3_FunctionCall/weather/weather_zhipu.py

运行前准备：
    pip install zhipuai python-dotenv requests
    在同目录 .env 中写入： zhipu_api=你的APIKEY
"""
import os                                    # 读取环境变量，密钥不写死在代码里
from dotenv import load_dotenv, find_dotenv  # 从 .env 文件加载环境变量
from tools import *                          # 导入 tools 列表、get_current_weather、parse_response
from zhipuai import ZhipuAI                  # 智谱官方 SDK

_ = load_dotenv(find_dotenv())               # find_dotenv 自动向上查找 .env，load 进 os.environ
zhupu_ak = os.environ['zhipu_api']           # 从环境变量取 API-KEY（严禁硬编码进源码）
client = ZhipuAI(api_key=zhupu_ak)           # 创建客户端
ChatGLM = "glm-4"                            # 选用支持 Function Call 的模型


def chat_completion_request(messages, tools=None, tool_choice=None, model=ChatGLM):
    """
    统一的模型调用入口（两次调用复用同一个函数，只是传入的 messages 不同）

    参数:
        messages:    对话消息列表，Function Call 的全部状态都靠它承载
        tools:       函数描述清单；不传则模型无从得知有哪些函数可用
        tool_choice: "auto" 表示由模型自行决定用文本回答还是调用函数
        model:       模型名称
    返回:
        成功返回响应对象，失败返回异常对象
    """
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=tools,
            tool_choice=tool_choice,
        )
        return response
    except Exception as e:
        print("Unable to generate ChatCompletion response")
        print(f"Exception: {e}")
        return e


def main():
    # ---------- 第 1 步：构造初始 messages（此时 2 条：system + user）----------
    messages = []
    messages.append({
        "role": "system",
        "content": "你是一个天气播报小助手，你需要根据用户提供的地址来回答当地的天气情况，"
                   "如果用户提供的问题具有不确定性，不要自己编造内容，提示用户明确输入"
    })
    messages.append({"role": "user", "content": "今天北京的天气如何"})

    # ---------- 第 2 步：第一次调用模型（带上 tools）----------
    # 模型此时不会给出天气，而是返回 tool_calls：函数名 + 参数
    response = chat_completion_request(messages, tools=tools, tool_choice="auto")

    # ---------- 第 3 步：本地真正执行函数，拿到真实数据 ----------
    function_response = parse_response(response)

    # ---------- 第 4 步：把模型那条 assistant 消息原样回填进 messages ----------
    # 必须回填！否则下一条 role 为 tool 的消息会找不到它所对应的 tool_calls
    assistant_message = response.choices[0].message
    messages.append(assistant_message.model_dump())   # model_dump 把对象转成字典

    # 取出函数名与本次工具调用的 id（id 用来把「回执」和「委托单」配对）
    function_name = response.choices[0].message.tool_calls[0].function.name
    function_id = response.choices[0].message.tool_calls[0].id

    # ---------- 第 5 步：把函数执行结果作为 role 为 tool 的消息加入 messages ----------
    messages.append({
        "role": "tool",                 # 固定角色名，表示这是工具执行结果
        "tool_call_id": function_id,    # 必须与上一条 assistant 中的 tool_calls[0].id 一致
        "name": function_name,          # 函数名
        "content": function_response,   # 函数返回值，字符串
    })
    # 此刻 messages 共 4 条：system / user / assistant(tool_calls) / tool

    # ---------- 第 6 步：第二次调用模型，得到最终自然语言答案 ----------
    last_response = chat_completion_request(messages, tools=tools, tool_choice="auto")
    print(f'last_response--》{last_response.choices[0].message}')


if __name__ == '__main__':
    main()

# ① 天气案例（单一函数）最终输出
last_response--》CompletionMessage(
    content='根据您的查询，我获取到了北京市当前的天气情况。今天是星期一，北京的天气情况是晴天，
             最高气温为33℃，最低气温为17℃。请注意天气变化，做好防晒和保暖措施。',
    role='assistant',
    tool_calls=None)          # ← tool_calls 变成 None，说明这次是真答案，循环结束

# ② 航班案例（多个函数）最终输出
last_response--》CompletionMessage(
    content='2024年4月2日，郑州到北京的航班号为1123，票价为668元。',
    role='assistant',
    tool_calls=None)

# ③ 数据库案例最终输出
last_response--》content='根据您的查询，我已经为您找到了工资最高的员工。这位员工的姓名是KING，他的工资是5000元。'
                 role='assistant' tool_calls=None

"""
多函数应用（航班）—— 函数层：定义多个可被委托的函数 + 统一的调用分发
文件路径：./ChatGLM3_FunctionCall/airplane/muti_utils.py
说明：可直接运行的完整脚本。
"""
import json


# ============ 函数 1：根据「出发地 + 目的地 + 日期」查航班号 ============
def get_plane_number(date, start, end):
    """
    用写死的字典模拟航班库；真实项目里把这里换成航司 API 或数据库查询即可，
    对模型而言接口形态不变 —— 它只负责填参数，不关心你内部怎么查。
    """
    plane_number = {
        "北京": {
            "深圳": "126",
            "广州": "356",
        },
        "郑州": {
            "北京": "1123",
            "天津": "3661",
        }
    }
    # 返回结构里带上 date，方便模型在下一轮把日期继续传给票价函数
    return {"date": date, "number": plane_number[start][end]}


# ============ 函数 2：根据「航班号 + 日期」查票价 ============
def get_ticket_price(date: str, number: str):
    """用固定票价模拟；重点在于它依赖上一个函数的输出（航班号）"""
    print(date)
    print(number)
    return {"ticket_price": "668"}


# ============ 统一分发：按模型给出的函数名，派发到对应的本地函数 ============
def parse_function_call(model_response):
    """
    与单函数版最大的不同：这里要用 if 判断函数名，
    因为模型可能选 get_plane_number，也可能选 get_ticket_price。
    """
    function_result = ''
    if model_response.choices[0].message.tool_calls:
        tool_call = model_response.choices[0].message.tool_calls[0]
        args = tool_call.function.arguments        # 字符串形式的 JSON

        function_result = {}
        if tool_call.function.name == "get_plane_number":
            # ** 解包：把 {"start":"郑州","end":"北京","date":"2024-04-02"} 展开成关键字参数
            function_result = get_plane_number(**json.loads(args))
        if tool_call.function.name == "get_ticket_price":
            function_result = get_ticket_price(**json.loads(args))

    return function_result

"""
多函数应用（航班）—— 工具描述层：一次性把两个函数都描述给模型
文件路径：./ChatGLM3_FunctionCall/airplane/airplane_function_tools.py

关键点：tools 是一个数组，里面放几个元素，模型手上就有几张「可委托事项」。
模型每一轮只会挑一个去填委托单，挑哪个取决于 description 写得准不准。
"""

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_plane_number",
            # 描述里点明了三个输入和一个输出，模型才知道「先查航班号」这一步归它
            "description": "根据始发地、目的地和日期，查询对应日期的航班号",
            "parameters": {
                "type": "object",
                "properties": {
                    "start": {
                        "description": "出发地",
                        "type": "string"
                    },
                    "end": {
                        "description": "目的地",
                        "type": "string"
                    },
                    "date": {
                        "description": "日期",
                        "type": "string",
                    }
                },
                # 三个参数都必填：用户说不全时，模型会追问而不是编造
                "required": ["start", "end", "date"]
            },
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_ticket_price",
            # 这个函数需要「航班号」，而航班号只能由上一个函数产出 ——
            # 两个函数因此形成依赖链，模型会自动分成两轮来调
            "description": "查询某航班在某日的价格",
            "parameters": {
                "type": "object",
                "properties": {
                    "number": {
                        "description": "航班号",
                        "type": "string"
                    },
                    "date": {
                        "description": "日期",
                        "type": "string",
                    }
                },
                "required": ["number", "date"]
            },
        }
    },
]

"""
多函数应用（航班）—— 主逻辑：三次调用模型，中间夹两次本地函数执行
文件路径：./ChatGLM3_FunctionCall/airplane/muti_function_zhipu.py
说明：可直接运行的完整脚本。

一句话看懂本例的串联：
    用户只问「票价」，但票价函数要航班号 ——
    模型第 1 轮先委托 get_plane_number 拿到航班号 1123，
    第 2 轮才委托 get_ticket_price 拿到票价 668，
    第 3 轮把两段回执合成一句人话。

运行前准备：
    pip install zhipuai python-dotenv
    在同目录 .env 中写入： zhipu_api=你的APIKEY
"""
import os
import json
from zhipuai import ZhipuAI
from dotenv import load_dotenv, find_dotenv
from muti_utils import *                      # 三个自定义函数
from airplane_function_tools import *         # tools 描述清单

_ = load_dotenv(find_dotenv())
zhupu_ak = os.environ['zhipu_api']            # 从环境变量读取 API-KEY
client = ZhipuAI(api_key=zhupu_ak)
ChatGLM = "glm-4"


def chat_completion_request(messages, tools=None, tool_choice=None, model=ChatGLM):
    """模型调用入口，与单函数版完全一致 —— 多函数的复杂度不在调用，而在 messages 的累积"""
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=tools,
            tool_choice=tool_choice,
        )
        return response
    except Exception as e:
        print("Unable to generate ChatCompletion response")
        print(f"Exception: {e}")
        return e


def main():
    # ---------- 准备 messages ----------
    messages = []
    messages.append({
        "role": "system",
        "content": "现在你是一个航班查询助手，将根据用户问题提供答案，"
                   "但是不要假设或猜测传入函数的参数值。如果用户的描述不明确，请要求用户提供必要信息"
    })
    messages.append({
        "role": "user",
        "content": "帮我查询2024年4月2日，郑州到北京的航班的票价"
    })

    # ========== 第一轮：查航班号 ==========
    # 1. 第一次调用模型 —— 模型判断出要先拿航班号
    first_response = chat_completion_request(messages, tools=tools, tool_choice="auto")
    assistant_message1 = first_response.choices[0].message
    print(f'assistant_message1-->{assistant_message1}')

    # 2. 把模型这条 assistant 回复（内含 tool_calls）原样加入 messages
    messages.append(first_response.choices[0].message.model_dump())

    # 3. 按模型给的函数名和参数，在本地真正执行 get_plane_number
    first_function = parse_function_call(model_response=first_response)

    # 4. 把第一次的函数结果作为 role 为 tool 的消息加入 messages
    tool_call = first_response.choices[0].message.tool_calls[0]
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,                       # 与第一张委托单配对
        "content": str(json.dumps(first_function)),         # content 必须是字符串
    })

    # ========== 第二轮：用航班号查票价 ==========
    # 5. 第二次调用模型 —— 模型看到航班号 1123，于是委托第二个函数
    second_response = chat_completion_request(messages, tools=tools, tool_choice="auto")

    # 6. 第二条 assistant 回复同样要回填
    messages.append(second_response.choices[0].message.model_dump())

    # 7. 本地执行 get_ticket_price
    second_function = parse_function_call(model_response=second_response)
    tool2_call = second_response.choices[0].message.tool_calls[0]

    # 8. 第二次的函数结果继续追加为 role 为 tool 的消息
    messages.append({
        "role": "tool",
        "tool_call_id": tool2_call.id,                      # 与第二张委托单配对
        "content": str(json.dumps(second_function)),
    })
    # 此刻 messages 共 6 条：
    # system / user / assistant(委托1) / tool(回执1) / assistant(委托2) / tool(回执2)

    # ========== 第三轮：合成最终答案 ==========
    # 9. 最后一次调用，模型不再返回 tool_calls，而是给出自然语言结果
    last_response = chat_completion_request(messages, tools=tools, tool_choice="auto")
    print(f'last_response--》{last_response.choices[0].message}')


if __name__ == '__main__':
    main()

"""
数据库查询应用 —— 工具层：表结构描述 + 执行 SQL 的函数 + tools 描述 + 解析分发
文件路径：./ChatGLM3_FunctionCall/sql/sql_function_tools.py
说明：可直接运行的完整脚本。

本例的精髓：函数只有一个（ask_database），但它的参数是一条 SQL 语句。
把「建表语句」塞进参数的 description 里，模型就知道有哪些表、哪些字段，
从而自己写出正确的 SQL —— 模型负责写 SQL，pymysql 负责执行 SQL。
"""
import os
import json
import pymysql

# ============ 1. 把数据库表结构写成字符串（单表版本）============
database_schema_string = """
CREATE TABLE `emp` (
  `empno` int DEFAULT NULL, --员工编号, 默认为空
  `ename` varchar(50) DEFAULT NULL, --员工姓名, 默认为空
  `job` varchar(50) DEFAULT NULL,--员工工作, 默认为空
  `mgr` int DEFAULT NULL,--员工领导, 默认为空
  `hiredate` date DEFAULT NULL,--员工入职日期, 默认为空
  `sal` int DEFAULT NULL,--员工的月薪, 默认为空
  `comm` int DEFAULT NULL,--员工年终奖, 默认为空
  `deptno` int DEFAULT NULL,--员工部分编号, 默认为空
)"""

# ============ 2. 多表版本：表之间的关系也一并交代清楚 ============
# 注意每个字段后面的中文注释 —— 它们同样是给模型看的，
# 模型靠这些注释把「工资」对应到 sal、把「部门」对应到 deptno。
database_schema_string1 = """
CREATE TABLE `emp` (
  `empno` int DEFAULT NULL, --员工编号, 默认为空
  `ename` varchar(50) DEFAULT NULL, --员工姓名, 默认为空
  `job` varchar(50) DEFAULT NULL,--员工工作, 默认为空
  `mgr` int DEFAULT NULL,--员工领导, 默认为空
  `hiredate` date DEFAULT NULL,--员工入职日期, 默认为空
  `sal` int DEFAULT NULL,--员工的月薪, 默认为空
  `comm` int DEFAULT NULL,--员工年终奖, 默认为空
  `deptno` int DEFAULT NULL,--员工部分编号, 默认为空
);
CREATE TABLE `DEPT` (
  `DEPTNO` int NOT NULL, -- 部门编码, 默认为空
  `DNAME` varchar(14) DEFAULT NULL,--部门名称, 默认为空
  `LOC` varchar(13) DEFAULT NULL,--地点, 默认为空
  PRIMARY KEY (`DEPTNO`)
);
"""


# ============ 3. 真正执行 SQL 的本地函数 ============
def ask_database(query):
    """连接数据库，执行模型生成的 SQL 并返回结果"""
    print("进入函数内部")
    # 3.1 连接 MySQL（账号口令一律走环境变量，不要硬编码在源码里）
    conn = pymysql.connect(
        host=os.environ.get("MYSQL_HOST", "localhost"),
        port=int(os.environ.get("MYSQL_PORT", 3306)),
        user=os.environ.get("MYSQL_USER", "root"),
        password=os.environ["MYSQL_PASSWORD"],          # 从环境变量读取，缺失即报错
        database=os.environ.get("MYSQL_DATABASE", "it_heima"),
        charset='utf8mb4',
    )
    cursor = conn.cursor()      # 3.2 创建游标
    print('开始测试')
    cursor.execute(query)       # 3.3 执行模型写好的 SQL
    result = cursor.fetchall()  # 3.4 取回全部查询结果
    cursor.close()              # 3.5 关闭游标
    conn.close()                # 3.6 关闭连接
    return result


# ============ 4. tools：把表结构塞进参数描述里 ============
tools = [
    {
        "type": "function",
        "function": {
            "name": "ask_database",
            "description": "使用此函数回答业务问题，要求输出是一个SQL查询语句",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        # 这段 f-string 就是本例的核心：
                        # 把建表语句拼进 description，模型才能写出字段正确的 SQL
                        "description": f"SQL查询提取信息以回答用户的问题。"
                                       f"SQL应该使用以下数据库模式编写:{database_schema_string1}"
                                       f"查询应该以纯文本返回，而不是JSON。"
                                       f"查询应该只包含MySQL支持的语法。",
                    }
                },
                "required": ["query"],
            },
        }
    }
]


# ============ 5. 解析模型回复并执行函数 ============
def parse_response(response):
    response_message = response.choices[0].message
    if response_message.tool_calls:
        available_functions = {
            "ask_database": ask_database
        }
        function_name = response_message.tool_calls[0].function.name
        fuction_to_call = available_functions[function_name]
        function_args = json.loads(response_message.tool_calls[0].function.arguments)
        # 这里的 query 参数，内容就是模型自己写出来的那条 SQL 语句
        function_response = fuction_to_call(
            query=function_args.get("query"),
        )
        return function_response

"""
数据库查询应用 —— 主逻辑：两次调用模型，中间夹一次 SQL 执行
文件路径：./ChatGLM3_FunctionCall/sql/sql_zhipu.py
说明：可直接运行的完整脚本。

运行前准备：
    pip install zhipuai python-dotenv pymysql
    在同目录 .env 中写入：
        zhipu_api=你的APIKEY
        MYSQL_PASSWORD=你的数据库密码
"""
import os
import json
from zhipuai import ZhipuAI
from dotenv import load_dotenv, find_dotenv
from sql_function_tools import *      # tools、ask_database、parse_response

_ = load_dotenv(find_dotenv())
zhupu_ak = os.environ['zhipu_api']    # 从环境变量读取 API-KEY
client = ZhipuAI(api_key=zhupu_ak)
ChatGLM = "glm-4"


def chat_completion_request(messages, tools=None, tool_choice=None, model=ChatGLM):
    """模型调用入口：三个案例共用同一套写法"""
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=tools,
            tool_choice=tool_choice,
        )
        return response
    except Exception as e:
        print("Unable to generate ChatCompletion response")
        print(f"Exception: {e}")
        return e


def main():
    # ---------- 第 1 步：准备 messages ----------
    messages = []
    messages.append({
        "role": "system",
        "content": "通过针对业务数据库生成 SQL 查询来回答用户的问题"
    })
    messages.append({
        "role": "user",
        "content": "查询一下最高工资的员工姓名及对应的工资"
    })

    # ---------- 第 2 步：第一次调用模型 ----------
    # 模型不会去查库，它返回的 arguments 里是一条它写好的 SQL，例如：
    # SELECT ename, sal FROM emp ORDER BY sal DESC LIMIT 1;
    response = chat_completion_request(messages, tools=tools, tool_choice="auto")

    # ---------- 第 3 步：本地用 pymysql 执行这条 SQL ----------
    function_response = parse_response(response)

    # ---------- 第 4 步：回填 assistant 消息 ----------
    assistant_message = response.choices[0].message
    print(f'assistant_message-->{assistant_message}')
    messages.append(assistant_message.model_dump())

    function_name = response.choices[0].message.tool_calls[0].function.name
    function_id = response.choices[0].message.tool_calls[0].id

    # ---------- 第 5 步：把查询结果作为 role 为 tool 的消息加入 ----------
    # 注意：fetchall() 返回的是元组，必须转成字符串才能放进 content
    messages.append({
        "role": "tool",
        "tool_call_id": function_id,
        "name": function_name,
        "content": str(function_response),
    })

    # ---------- 第 6 步：第二次调用模型，把查询结果翻译成人话 ----------
    last_response = chat_completion_request(messages, tools=tools, tool_choice="auto")
    print(f'last_response--》{last_response.choices[0].message}')


if __name__ == '__main__':
    main()

-- 用户原话：查询一下最高工资的员工姓名及对应的工资
-- 模型据此写出的 SQL（放在 tool_calls[0].function.arguments 的 query 字段里）
SELECT ename, sal
FROM emp
ORDER BY sal DESC
LIMIT 1;

-- pymysql 执行后 cursor.fetchall() 返回：(('KING', 5000),)
-- 这份结果转成字符串，作为 role 为 tool 的消息回填进 messages

# -*- coding: utf-8 -*-
"""
Function Call 可复用骨架模板（通用版）
==========================================
把这个文件整个复制到你自己的项目里，只改 TODO(拓展点) 标注的地方就能跑。
三个案例（天气 / 航班 / 数据库）都是这个骨架的特例。

设计要点：
  1. available_functions 是「函数注册表」——加新工具只动这一个字典和 tools 列表；
  2. run_conversation 用 while 循环自动多轮，直到模型不再返回 tool_calls；
     （天气案例是 2 轮，航班案例是 3 轮，这里不写死轮数）
  3. 与 OpenAI 风格 SDK 完全兼容：client.chat.completions.create(tools=..., tool_choice="auto")。
"""

import json
import os

# ============================================================
# 一、client 初始化：两种写法二选一，只差 import 和 base_url
# ============================================================
# TODO(拓展点 1)：换模型 / 换厂商，只改这一段

# 写法 A —— 智谱官方 SDK（本讲示例用的就是这个）
from zhipuai import ZhipuAI

client = ZhipuAI(api_key=os.environ["ZHIPUAI_API_KEY"])
MODEL = "glm-4"

# 写法 B —— openai 包（同一套代码可对接 OpenAI / DeepSeek / Qwen / 本地 vLLM）
# from openai import OpenAI
# client = OpenAI(
#     api_key=os.environ["OPENAI_API_KEY"],
#     base_url="https://open.bigmodel.cn/api/paas/v4/",   # 换成你的厂商地址即可
# )
# MODEL = "glm-4"
#
# 常见 base_url 参考：
#   OpenAI      https://api.openai.com/v1
#   智谱         https://open.bigmodel.cn/api/paas/v4/
#   DeepSeek    https://api.deepseek.com
#   本地 vLLM    http://127.0.0.1:8000/v1
# 两种写法下面的业务代码一个字都不用改——tool_calls 的字段结构是一样的。


# ============================================================
# 二、你的真实函数：模型永远不执行它们，执行的是这里
# ============================================================
# TODO(拓展点 2)：把下面两个示例函数换成你自己的业务函数
#   规则：参数名要和 tools 描述里的 properties 完全一致；
#         返回值建议是 str / dict，方便统一 json.dumps。

def my_function_a(param1, param2="默认值"):
    """示例函数 A：换成你的——查数据库、调接口、读文件、发消息都行。"""
    # ... 你的业务逻辑 ...
    return {"param1": param1, "param2": param2, "result": "这里换成真实结果"}


def my_function_b(param1):
    """示例函数 B：用来演示「多函数」和「函数间依赖」。"""
    return {"input": param1, "result": "第二个函数的结果"}


# ============================================================
# 三、函数注册表：加新工具时，这里加一行
# ============================================================
# TODO(拓展点 3)：注册你的函数（键必须与 tools 里的 name 一致）
available_functions = {
    "my_function_a": my_function_a,
    "my_function_b": my_function_b,
}


# ============================================================
# 四、tools 描述：告诉模型有什么可用、参数怎么填
# ============================================================
# TODO(拓展点 4)：照着抄一份就是一个新工具
#   description 写清楚「什么时候该用它」，模型靠它决定用不用；
#   required 里的参数是必填，模型缺了会反问用户。
tools = [
    {
        "type": "function",
        "function": {
            "name": "my_function_a",                      # 必须与 available_functions 的键一致
            "description": "这个函数是干什么的、什么情况下调用它",
            "parameters": {
                "type": "object",
                "properties": {
                    "param1": {
                        "type": "string",
                        "description": "参数含义，写清楚格式和示例，例如：城市名，如 北京",
                    },
                    "param2": {
                        "type": "string",
                        "description": "可选参数的含义",
                    },
                },
                "required": ["param1"],                   # 必填参数
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "my_function_b",
            "description": "第二个函数的用途说明",
            "parameters": {
                "type": "object",
                "properties": {
                    "param1": {"type": "string", "description": "参数含义"},
                },
                "required": ["param1"],
            },
        },
    },
]


# ============================================================
# 五、解析分发：把模型给的「函数名 + 参数」真正执行掉
# ============================================================
def parse_response(tool_call):
    """执行一次 tool_call，返回字符串形式的结果。

    注意两个坑：
      1. arguments 是【字符串形式的 JSON】，必须 json.loads 才能取值；
      2. 回填给模型的 content 必须是【字符串】，元组/字典要先 dumps 或 str。
    """
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments or "{}")

    func = available_functions.get(name)
    if func is None:
        return f"错误：未注册的函数 {name}"

    try:
        result = func(**args)
    except Exception as exc:                      # 函数报错也要如实回给模型，别让它瞎猜
        return f"函数 {name} 执行失败：{exc}"

    if isinstance(result, str):
        return result
    return json.dumps(result, ensure_ascii=False)


# ============================================================
# 六、主循环：自动多轮，直到模型不再要求调用工具
# ============================================================
def run_conversation(user_input, system_prompt=None, max_rounds=5):
    """前面的实现是固定两轮/三轮，这里做成 while 循环，函数依赖再深也能跑完。"""
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": user_input})

    rounds = 0
    while True:                                   # 自动多轮，直到模型不再要求调用工具
        rounds += 1
        if rounds > max_rounds:                   # max_rounds 是保险丝，防止死循环
            return "已达到最大轮数上限，请检查函数是否陷入循环调用。"

        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=tools,                          # 每一轮都要继续传 tools
            tool_choice="auto",                   # auto = 由模型自己决定用不用工具
        )
        msg = response.choices[0].message

        # 没有 tool_calls，说明模型已经能直接回答，收工
        if not getattr(msg, "tool_calls", None):
            return msg.content

        # 1) 先把带 tool_calls 的 assistant 消息原样回填（漏了这步，下一条 tool 消息会成孤儿）
        messages.append(msg.model_dump())

        # 2) 逐个执行，每个 tool_call 回填一条 role="tool" 的消息
        for tool_call in msg.tool_calls:
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,     # 靠它和上面的调用配对，不能省
                "name": tool_call.function.name,
                "content": parse_response(tool_call),
            })
        # 3) 回到循环顶部，带着函数结果再问一次模型


# ============================================================
# 七、跑起来
# ============================================================
if __name__ == "__main__":
    # TODO(拓展点 5)：换成你自己的提问和 system 提示词
    answer = run_conversation(
        user_input="这里换成用户的问题",
        system_prompt="你是一个助手，需要时请调用提供的工具，不要编造数据。",
    )
    print(answer)

# -*- coding: utf-8 -*-
"""
Function Call 数据库问答骨架（参数即 SQL 版）
==========================================
SQL 案例去掉 emp/dept 表之后的通用版本。
特点：只有一个工具 ask_database，它的唯一参数就是【一整条 SQL】——
      SQL 由模型写，执行仍然由你的代码做。

⚠️ 安全红线：模型写的 SQL 绝不能直接执行。
   本模板内置只读校验（只允许 SELECT、禁止多语句、强制 LIMIT），
   生产环境还必须配一个【只读数据库账号】兜底。
"""

import json
import os
import re

import pymysql

# TODO(拓展点 1)：换 client / 换模型
from zhipuai import ZhipuAI

client = ZhipuAI(api_key=os.environ["ZHIPUAI_API_KEY"])
MODEL = "glm-4"
# openai 包写法：
# from openai import OpenAI
# client = OpenAI(api_key=os.environ["OPENAI_API_KEY"],
#                 base_url="https://open.bigmodel.cn/api/paas/v4/")


# ============================================================
# 一、表结构描述：模型全靠它写 SQL
# ============================================================
# TODO(拓展点 2)：把这里换成你自己的建表语句
#   两个要点：① 字段一定要带中文注释，模型靠注释把「工资」对应到 sal；
#            ② 多表时必须写明表间关系，否则模型写不出正确的 JOIN。
database_schema_string = """
CREATE TABLE your_table (
    id      INT PRIMARY KEY,        -- 主键
    name    VARCHAR(50),            -- 名称，例如 员工姓名
    amount  DECIMAL(10,2),          -- 金额，例如 月薪
    dept_id INT                     -- 部门编号，关联 dept.id
);

CREATE TABLE dept (
    id    INT PRIMARY KEY,          -- 部门编号
    dname VARCHAR(50)               -- 部门名称
);
-- 表间关系：your_table.dept_id = dept.id
"""


# ============================================================
# 二、执行 SQL 的函数（带安全校验）
# ============================================================
# TODO(拓展点 3)：换成你的数据库连接信息（一律走环境变量，别写死口令）
DB_CONF = dict(
    host=os.environ.get("DB_HOST", "127.0.0.1"),
    port=int(os.environ.get("DB_PORT", 3306)),
    user=os.environ["DB_USER"],            # 建议：只读账号
    password=os.environ["DB_PASSWORD"],
    database=os.environ.get("DB_NAME", "your_db"),
    charset="utf8mb4",
)

MAX_ROWS = 200


def _guard(sql):
    """把模型写的 SQL 关进笼子里。缺了这一步，一句 DELETE 就能删库。"""
    s = sql.strip().rstrip(";").strip()
    if ";" in s:
        raise ValueError("拒绝执行：不允许多语句")
    if not re.match(r"(?is)^\s*(select|with)\b", s):
        raise ValueError("拒绝执行：只允许 SELECT 查询")
    if re.search(r"(?i)\b(insert|update|delete|drop|alter|truncate|grant|create)\b", s):
        raise ValueError("拒绝执行：检测到写操作关键字")
    if not re.search(r"(?i)\blimit\b", s):
        s += f" LIMIT {MAX_ROWS}"             # 没写 LIMIT 就替它加上，防止拉全表
    return s


def ask_database(query):
    """执行模型生成的查询，返回字符串形式的结果。"""
    safe_sql = _guard(query)
    conn = pymysql.connect(**DB_CONF)
    try:
        with conn.cursor() as cursor:
            cursor.execute(safe_sql)
            rows = cursor.fetchall()
    finally:
        conn.close()                          # 异常路径也要关，别泄漏连接
    return str(rows)


available_functions = {"ask_database": ask_database}


# ============================================================
# 三、tools 描述：把表结构塞进参数说明里
# ============================================================
tools = [{
    "type": "function",
    "function": {
        "name": "ask_database",
        "description": "用 SQL 查询数据库来回答用户关于业务数据的问题。",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": (
                        "一条合法的 MySQL SELECT 语句，用于回答用户问题。"
                        "必须依据下面的表结构书写，只返回 SQL 本身，不要加解释：\n"
                        f"{database_schema_string}"
                    ),
                },
            },
            "required": ["query"],
        },
    },
}]


# ============================================================
# 四、主流程：两次调用
# ============================================================
def run(user_input):
    messages = [
        {"role": "system", "content": "你是数据分析助手，只依据数据库查询结果回答，不要编造数字。"},
        {"role": "user", "content": user_input},
    ]

    first = client.chat.completions.create(
        model=MODEL, messages=messages, tools=tools, tool_choice="auto")
    msg = first.choices[0].message

    if not getattr(msg, "tool_calls", None):
        return msg.content

    messages.append(msg.model_dump())
    for tool_call in msg.tool_calls:
        args = json.loads(tool_call.function.arguments)
        try:
            content = available_functions[tool_call.function.name](**args)
        except Exception as exc:
            content = f"查询失败：{exc}"       # 把拒绝原因如实告诉模型，它会换一种问法
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "name": tool_call.function.name,
            "content": content,
        })

    second = client.chat.completions.create(
        model=MODEL, messages=messages, tools=tools, tool_choice="auto")
    return second.choices[0].message.content


if __name__ == "__main__":
    # TODO(拓展点 4)：换成你的提问
    print(run("这里换成用户的问题，例如：金额最高的是谁"))