ai

1. 基础概念 #

Tool Calls 让 LLM 不再「只说不做」——模型输出结构化的函数调用请求，由你的代码执行后把结果写回对话。本节先建立三个角色与一轮调用的基本图景，后续章节再展开格式与顺序约束。

1.1 什么是 Tool Calls？ #

Tool Calls 是 OpenAI API 的一种功能，允许 AI 模型调用外部函数/工具来获取信息或执行操作。

用户提问 → AI决定调用工具 → 执行工具 → 返回结果 → AI生成最终回答

1.2 关键角色 #

一轮 Tool Calls 对话里通常只出现三种 role（不含 system）：

角色	说明	示例
user	用户消息	"北京的天气怎么样？"
assistant	AI响应	可以包含文本或tool_calls
tool	工具执行结果	`{"temperature": 25}`

2. 消息格式 #

OpenAI 兼容 API 用三种 role 承载 Tool Calls 对话：assistant 携带 tool_calls 声明意图，tool 回传执行结果。下面分别看两种消息的 JSON 结构，以及一条完整的多轮序列。

2.1 Assistant 消息（包含 tool_calls） #

当模型决定调用工具时，assistant 消息的 content 常为 null，真正的意图写在 tool_calls 数组里：

{
    "role": "assistant",
    "content": None,  # 或可选的文本内容
    "tool_calls": [
        {
            "id": "call_abc123",  # 唯一标识符
            "type": "function",
            "function": {
                "name": "get_weather",
                "arguments": '{"city": "Beijing", "unit": "celsius"}'
            }
        },
        {
            "id": "call_def456",
            "type": "function",
            "function": {
                "name": "get_time",
                "arguments": '{"timezone": "Asia/Shanghai"}'
            }
        }
    ]
}

关键字段说明：

id：每个工具调用的唯一ID，必须在后续的tool消息中引用
function.arguments：必须是有效的JSON字符串
一个assistant消息可以包含多个tool_calls

2.2 Tool 消息（响应工具调用） #

tool 消息是 assistant → tool 配对的另一半：每个 tool_call_id 必须精确对应上文某条 tool_calls[].id，content 建议用 JSON 字符串承载结构化结果。

# 单个工具响应
{
    "role": "tool",
    "tool_call_id": "call_abc123",  # 必须匹配对应的tool_call ID
    "content": '{"temperature": 25, "condition": "sunny"}'
}

# 多个工具响应（顺序必须匹配）
[
    {
        "role": "tool",
        "tool_call_id": "call_abc123",
        "content": '{"temperature": 25}'
    },
    {
        "role": "tool",
        "tool_call_id": "call_def456",
        "content": '{"time": "14:30"}'
    }
]

重要规则：

tool_call_id 必须与 assistant 消息中的 id 完全匹配
content 可以是任何字符串，但建议使用JSON格式
每个 tool_calls 必须有且仅有一个对应的 tool 消息

2.3 完整的消息序列示例 #

把 §2.1 与 §2.2 串起来：一次用户提问触发两个并行工具，工具结果回传后 assistant 再给出自然语言总结。注意 messages 数组里的顺序——这就是 §3 要 enforce 的规则。

messages = [
    # 1. 用户提问
    {
        "role": "user",
        "content": "北京现在天气怎么样？现在几点了？"
    },

    # 2. AI决定调用工具
    {
        "role": "assistant",
        "content": None,
        "tool_calls": [
            {
                "id": "call_1",
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "arguments": '{"city": "Beijing"}'
                }
            },
            {
                "id": "call_2",
                "type": "function",
                "function": {
                    "name": "get_time",
                    "arguments": '{"timezone": "Asia/Shanghai"}'
                }
            }
        ]
    },

    # 3. 工具执行结果（必须按顺序）
    {
        "role": "tool",
        "tool_call_id": "call_1",
        "content": '{"temp": 25, "condition": "sunny"}'
    },
    {
        "role": "tool",
        "tool_call_id": "call_2",
        "content": '{"time": "14:30"}'
    },

    # 4. AI生成最终回答
    {
        "role": "assistant",
        "content": "北京现在天气晴朗，气温25°C，当前时间是14:30。"
    }
]

3. 消息顺序规则 #

API 对 messages 的顺序有硬约束：一旦出现带 tool_calls 的 assistant，后面必须连续跟上等量的 tool 消息，中间不能插 user 或其他角色。违反时典型报错是 insufficient tool messages following tool_calls message（s13 后台通知注入踩过这个坑）。

3.1 核心规则（必须遵守） #

合法消息链在逻辑上始终是下面这条线——中间不能跳过或插入其他 role：

1. user (提问)
2. assistant (tool_calls)
3. tool (结果1)
4. tool (结果2)
5. ... (更多tool)
6. assistant (最终回答)

3.2 规则详解 #

三条规则分别约束 tool 消息顺序、数量配对、中间不得插入其他 role。Agent 实现里应在 append assistant 后立刻 loop 执行全部 tool_calls，再考虑注入通知或继续下一轮。

3.2.1 规则1：顺序严格匹配 #

tool 消息在数组中的出现顺序，应与同一条 assistant 里 tool_calls 的下标顺序一致（DeepSeek / OpenAI 均按此校验）：

# 正确
assistant: tool_calls: [call_A, call_B, call_C]
tool: call_A
tool: call_B
tool: call_C

# 错误 - 顺序不对
assistant: tool_calls: [call_A, call_B, call_C]
tool: call_B
tool: call_A
tool: call_C

3.2.2 规则2：完整的配对 #

tool_calls 里有几个 id，后面就必须有几条 tool 消息——少一条即 400：

# 正确 - 3个工具调用，3个响应
assistant: tool_calls: [call_1, call_2, call_3]
tool: call_1
tool: call_2
tool: call_3

# 错误 - 缺少响应
assistant: tool_calls: [call_1, call_2]
tool: call_1
# 缺少 call_2 的响应

3.2.3 规则3：不能插入其他消息 #

从第一条 tool 到最后一条 tool 之间，只能连续出现 role: tool，不能夹 user、带 tool_calls 的 assistant，或后台通知文本：

# 错误 - 在tool消息之间插入了其他内容
assistant: tool_calls: [call_1, call_2]
tool: call_1
user: "请继续"  # 不允许！
tool: call_2

4. 完整工作流程 #

从用户输入到最终回答，最小 Agent 固定两次 API 调用（有 tool 时）：第一次拿 tool_calls，本地执行后第二次拿汇总回复。复杂 Agent 则在 while tool_calls 里循环（见 §7.1）。

4.1 流程图 #

下面时序图概括 §4.2 代码的两段式调用；若模型连续多轮发起 tool_calls，则步骤 4–6 在循环中重复（见 §7.1）。

sequenceDiagram participant User as 用户 participant API as API participant AI as AI助手 participant Tool as 工具 User->>API: 1. 用户发送消息 API->>AI: 2. API处理请求 AI->>API: 3. AI响应 alt 有 tool_calls API->>Tool: 4. 执行所有工具（可多次） Tool-->>API: 工具结果 API->>AI: 5. 构造 tool 消息 (顺序匹配)\n6. 发送完整对话给API AI->>API: 7. AI生成最终回答 else 无 tool_calls API-->>User: 直接返回给用户 end

4.2 代码实现 #

下面是对 §2.3 流程的可运行实现（test.py）：DeepSeek + 天气/时间双工具，演示「第一次 create → 执行 tool → 第二次 create」。

# 指定 Python 解释器路径
#!/usr/bin/env python3
# 文件文档字符串，说明功能
"""Tool Calls 最小示例：DeepSeek + 天气/时间工具。"""

# 导入 json 模块，用于处理 JSON 数据
import json
# 导入 os 模块，用于操作环境变量和文件
import os
# 导入 sys 模块，用于操作系统交互
import sys
# 导入 datetime、timedelta、timezone 类，用于处理时间和时区
from datetime import datetime, timedelta, timezone

# 从 dotenv 模块导入 load_dotenv，用于加载环境变量
from dotenv import load_dotenv
# 从 openai 模块导入 OpenAI 类
from openai import OpenAI

# 加载 .env 文件中的环境变量（覆盖已有变量）
load_dotenv(override=True)

# 创建 OpenAI 客户端对象，使用环境变量中的 API_KEY 和 BASE_URL
client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.getenv("OPENAI_BASE_URL"),
)
# 获取模型 ID
MODEL = os.environ["MODEL_ID"]


# 定义用于获取天气的函数，参数为城市名，返回字典
def get_weather(city: str) -> dict:
    # 模拟天气 API。
    """模拟天气 API。"""
    # 定义静态天气数据
    data = {
        "Beijing": {"temp": 25, "condition": "sunny", "unit": "celsius"},
        "Shanghai": {"temp": 30, "condition": "rainy", "unit": "celsius"},
        "北京": {"temp": 25, "condition": "sunny", "unit": "celsius"},
        "上海": {"temp": 30, "condition": "rainy", "unit": "celsius"},
    }
    # 根据城市名返回对应天气信息，若无则返回默认 unknown
    return data.get(city, {"temp": "unknown", "condition": "unknown", "unit": "celsius"})


# 定义用于获取时间的函数，参数为时区名，返回字典
def get_time(tz: str) -> dict:
    # 返回指定时区当前时间（固定 UTC 偏移，无需 tzdata）。
    """返回指定时区当前时间（固定 UTC 偏移，无需 tzdata）。"""
    # 定义部分时区的时差
    offsets = {
        "Asia/Shanghai": 8,
        "Asia/Hong_Kong": 8,
        "Asia/Beijing": 8,
        "UTC": 0,
    }
    # 获取该时区对应的小时偏移，默认为8
    hours = offsets.get(tz, 8)
    # 获取本地时间，并设置为指定时区
    now = datetime.now(timezone(timedelta(hours=hours)))
    # 返回当前时区、时间、日期信息
    return {
        "timezone": tz,
        "time": now.strftime("%H:%M:%S"),
        "date": now.strftime("%Y-%m-%d"),
    }


# 定义函数，用于构建 tools 需要的 function 调用描述
def _fn_tool(name: str, description: str, properties: dict, required: list[str]) -> dict:
    # 返回符合 OpenAI tool schema 的结构
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }


# 构建 get_weather 工具的 schema
get_weather_tool = _fn_tool(
    "get_weather",
    "查询指定城市的天气。",
    {"city": {"type": "string", "description": "城市名，如 Beijing 或 北京"}},
    ["city"],
)

# 构建 get_time 工具的 schema
get_time_tool = _fn_tool(
    "get_time",
    "查询指定时区的当前时间。",
    {"timezone": {"type": "string", "description": "IANA 时区，如 Asia/Shanghai"}},
    ["timezone"],
)

# 定义全部可用工具
TOOLS = [get_weather_tool, get_time_tool]

# 定义工具回调函数字典，将工具名映射到 Lambda 处理方法
TOOL_HANDLERS = {
    "get_weather": lambda args: get_weather(args["city"]),
    "get_time": lambda args: get_time(args["timezone"]),
}


# 定义主流程函数，处理用户输入并调用 tool
def process_with_tools(user_query: str) -> str:
    # 构造初始 Messages，将用户输入内容加入
    messages = [{"role": "user", "content": user_query}]

    # 调用 OpenAI 接口获得初步 assistant 消息，可包含 tool_calls
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
    )

    # 取出第一条 assistant 响应
    assistant_message = response.choices[0].message

    # 如果没有建议调用工具则直接返回内容
    if not assistant_message.tool_calls:
        return assistant_message.content or ""

    # 否则，将 assistant 的 tool 调用信息追加入 messages
    messages.append({
        "role": "assistant",
        "content": assistant_message.content,
        "tool_calls": [
            {
                "id": tc.id,
                "type": tc.type,
                "function": {
                    "name": tc.function.name,
                    "arguments": tc.function.arguments,
                },
            }
            for tc in assistant_message.tool_calls
        ],
    })

    # 对于 assistant 请求调用的每个工具，逐一执行
    for tool_call in assistant_message.tool_calls:
        # 获取工具名
        name = tool_call.function.name
        # 解析参数（JSON 格式字符串）
        args = json.loads(tool_call.function.arguments or "{}")
        # 根据名字获取对应的处理回调
        handler = TOOL_HANDLERS.get(name)
        # 执行回调获取结果，若找不到工具则报错
        result = handler(args) if handler else {"error": f"Unknown tool: {name}"}
        # 打印调试信息
        print(f"  [tool] {name}({args}) -> {result}")

        # 把工具返回的结果加入 messages 供下一步模型使用
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result, ensure_ascii=False),
        })

    # 工具结果返回后，再次发起请求让 assistant 总结输出
    final_response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=TOOLS,
    )

    # 获取最终响应
    final = final_response.choices[0].message
    # 如果还要调用工具，提示需要多运行一轮
    if final.tool_calls:
        return final.content or "（模型请求继续调用工具，请再运行一轮）"
    # 否则直接返回 assistant 文本内容
    return final.content or ""


# 如果作为主程序执行
if __name__ == "__main__":
    # 如果是 Windows 平台，需要将标准输出编码设为 utf-8
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8")
    # 打印模型和用法说明
    print(f"Tool Calls 示例（模型: {MODEL}）")
    print("输入问题，回车发送。输入 q 退出。\n")

    # 进入主循环，持续获取用户输入
    while True:
        try:
            # 获取用户输入，去掉首尾空白字符
            query = input(">> ").strip()
        except (EOFError, KeyboardInterrupt):
            # 捕捉 Ctrl+C、Ctrl+D 退出
            break
        # 如果用户输入 q 或 exit 或空则退出
        if query.lower() in ("q", "exit", ""):
            break
        # 处理用户输入并输出结果
        print(process_with_tools(query))
        # 打印空行分隔
        print()

5. 常见错误及解决方案 #

生产里绝大多数 Tool Calls 报错不是模型「不会用工具」，而是 messages 结构不合法。本节按报错信息对照五种高频错误；调试时可配合 §8 的 debug_messages / validate_message_chain。

5.1 错误1：缺少 tool 消息 #

错误信息：

BadRequestError: An assistant message with 'tool_calls' must be followed by 
tool messages responding to each 'tool_call_id'

原因：有 tool_calls 但没有对应的 tool 消息

解决方案：

# 错误
messages = [
    {"role": "user", "content": "查询天气"},
    {"role": "assistant", "tool_calls": [call_1]}
]
# 缺少 tool 消息

# 正确
messages = [
    {"role": "user", "content": "查询天气"},
    {"role": "assistant", "tool_calls": [call_1]},
    {"role": "tool", "tool_call_id": call_1.id, "content": "25°C"}
]

5.2 错误2：tool_call_id 不匹配 #

错误信息：

BadRequestError: Invalid 'tool_call_id' in tool message

解决方案：

# 错误 - ID不匹配
assistant: tool_calls: [{"id": "call_abc"}]
tool: {"tool_call_id": "call_xyz"}  # 错误的ID

# 正确
assistant: tool_calls: [{"id": "call_abc"}]
tool: {"tool_call_id": "call_abc"}  # 匹配

5.3 错误3：消息顺序错误 #

错误信息：

BadRequestError: Expected tool messages after assistant message with tool_calls

解决方案：

# 错误 - 顺序颠倒
messages = [
    {"role": "user", "content": "查询天气"},
    {"role": "tool", "tool_call_id": "call_1", "content": "25°C"},
    {"role": "assistant", "tool_calls": [call_1]}
]

# 正确
messages = [
    {"role": "user", "content": "查询天气"},
    {"role": "assistant", "tool_calls": [call_1]},
    {"role": "tool", "tool_call_id": "call_1", "content": "25°C"}
]

5.4 错误4：工具结果格式错误 #

错误信息：

BadRequestError: Invalid JSON in function arguments

解决方案：

# 错误 - 不是有效的JSON
arguments = "{city: Beijing}"  # 缺少引号

# 正确
arguments = '{"city": "Beijing"}'  # 有效JSON

# 使用json.dumps自动处理
arguments = json.dumps({"city": "Beijing"})

5.5 错误5：混合使用旧版函数调用 #

错误信息：

BadRequestError: Can't use both function_call and tool_calls

解决方案：

# 错误 - 同时使用旧版和新版
{
    "functions": [...],      # 旧版
    "tools": [...]           # 新版
}

# 正确 - 只使用一种方式
{
    "tools": [...]  # 推荐使用新版
}

6. 实战示例 #

以下三个可运行脚本由简到繁，均使用 DeepSeek（.env 配置）。每个示例后附推荐测试 prompt。

6.1 实战1：单个工具调用 #

对应文件： t2.py
要点： 仅一个 get_weather 工具，固定两轮 API——有 tool_calls 则执行后必再调一次 LLM。

# 指定 python3 作为解释器
#!/usr/bin/env python3
# 文件头部文档字符串，说明为单工具调用示例，结合 DeepSeek 与 get_weather
"""Tool Calls 单工具示例（对应 ToolCalls.md §6.1）：DeepSeek + get_weather。"""

# 导入 json 标准库，用于数据的序列化与反序列化
import json
# 导入 os 标准库，用于操作环境变量等
import os
# 导入 sys 标准库，用于系统级操作
import sys

# 从 dotenv 包导入 load_dotenv 用于加载 .env 文件中的环境变量
from dotenv import load_dotenv
# 从 openai 包导入 OpenAI 客户端类
from openai import OpenAI

# 加载 .env 文件中的环境变量（如果有重复则覆盖）
load_dotenv(override=True)

# 获取环境变量 SSL_CERT_FILE 的值
_ssl_cert = os.getenv("SSL_CERT_FILE")
# 如果设置了 SSL_CERT_FILE 且对应文件不存在，则从环境变量中移除
if _ssl_cert and not os.path.isfile(_ssl_cert):
    os.environ.pop("SSL_CERT_FILE", None)

# 创建 OpenAI 客户端实例，使用指定的 API key 和 base url
client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.getenv("OPENAI_BASE_URL"),
)
# 从环境变量中获取模型名称
MODEL = os.environ["MODEL_ID"]

# 定义一个获取天气信息的函数，参数为城市名，返回天气字典
def get_weather(city: str) -> dict:
    # 文档字符串：说明为模拟天气 API
    """模拟天气 API。"""
    # 预设的天气数据，支持中英文城市名
    weather_data = {
        "Beijing": {"temp": 25, "condition": "sunny"},
        "Shanghai": {"temp": 30, "condition": "rainy"},
        "北京": {"temp": 25, "condition": "sunny"},
        "上海": {"temp": 30, "condition": "rainy"},
    }
    # 返回对应城市的天气数据，若不存在则返回未知
    return weather_data.get(city, {"temp": "unknown", "condition": "unknown"})

# 定义天气工具的描述信息，用于 OpenAI 工具调用
weather_tool = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get weather information for a city",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name",
                }
            },
            "required": ["city"],
        },
    },
}

# 定义主业务逻辑函数，输入用户查询字符串，输出字符串
def run(user_query: str) -> str:
    # 初始化消息列表，将用户输入加入对话消息中
    messages = [{"role": "user", "content": user_query}]

    # 调用 openai 聊天模型，传入消息、工具定义，指定工具选择为 auto
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=[weather_tool],
        tool_choice="auto",
    )

    # 获取助手返回的消息对象
    assistant_msg = response.choices[0].message

    # 如果助手没有工具调用，直接返回助手回复内容
    if not assistant_msg.tool_calls:
        return assistant_msg.content or ""

    # 否则，将助手消息（包括工具调用内容）加入 messages 列表
    messages.append({
        "role": "assistant",
        "content": assistant_msg.content,
        "tool_calls": [
            {
                "id": tc.id,
                "type": tc.type,
                "function": {
                    "name": tc.function.name,
                    "arguments": tc.function.arguments,
                },
            }
            for tc in assistant_msg.tool_calls
        ],
    })

    # 遍历每个工具调用，解析参数并调用本地 weather api
    for tool_call in assistant_msg.tool_calls:
        # 解析工具调用参数（JSON 格式字符串转为字典）
        args = json.loads(tool_call.function.arguments or "{}")
        # 获取天气信息
        result = get_weather(args["city"])
        # 打印工具调用过程到控制台
        print(f"  [tool] get_weather({args}) -> {result}")

        # 将工具调用返回结果追加到消息流，供后续模型使用
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result, ensure_ascii=False),
        })

    # 再次调用 openai 聊天模型，让其综合工具调用结果得出最终答复
    final_response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=[weather_tool],
    )

    # 返回最终响应内容，如果不存在则返回空字符串
    return final_response.choices[0].message.content or ""

# 判断当前文件是否为主程序入口
if __name__ == "__main__":
    # 若系统为 windows，则重定义 stdout，设置编码为 utf-8
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8")

    # 打印程序头部说明，包含所用模型名
    print(f"单工具示例（模型: {MODEL}）")
    # 打印交互使用说明
    print("输入问题，回车发送。直接回车使用默认问题。输入 q 退出。\n")

    # 进入主循环，不断读取用户输入
    while True:
        try:
            # 读取一行用户输入，并去除首尾空白
            query = input(">> ").strip()
        except (EOFError, KeyboardInterrupt):
            # 用户退出（Ctrl+D 或 Ctrl+C）则跳出循环
            break
        # 若用户输入 q（不区分大小写），则退出主循环
        if query.lower() == "q":
            break
        # 若输入为空，则使用预设默认问题
        if not query:
            query = "北京的天气怎么样？"
        # 调用主 run 函数处理输入，并打印结果
        print(run(query))
        # 再打印一个空行，便于阅读
        print()

北京的天气怎么样？

What's the weather in Shanghai and Beijing? Which city is hotter?

深圳今天天气如何？

6.2 实战2：多个工具调用（并行） #

对应文件： t3.py
要点： 同一轮 assistant 可发起多个 tool_calls；本地按顺序 append 多条 tool 消息后，再一次 LLM 汇总（与 §7.2 线程并行执行对比：此处串行执行、并行决策）。

#!/usr/bin/env python3
"""Tool Calls 多工具并行示例（对应 ToolCalls.md §6.2）：DeepSeek + weather/time/calculate。"""

import json
import os
import sys
from datetime import datetime, timedelta, timezone

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv(override=True)

_ssl_cert = os.getenv("SSL_CERT_FILE")
if _ssl_cert and not os.path.isfile(_ssl_cert):
    os.environ.pop("SSL_CERT_FILE", None)

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.getenv("OPENAI_BASE_URL"),
)
MODEL = os.environ["MODEL_ID"]


def get_weather(city: str) -> dict:
    """模拟天气 API。"""
    data = {
        "Beijing": {"temp": 25, "condition": "sunny"},
        "Shanghai": {"temp": 30, "condition": "rainy"},
        "北京": {"temp": 25, "condition": "sunny"},
        "上海": {"temp": 30, "condition": "rainy"},
    }
    return data.get(city, {"temp": "unknown", "condition": "unknown"})


def get_time(tz: str) -> dict:
    """返回指定时区当前时间。"""
    offsets = {
        "Asia/Shanghai": 8,
        "Asia/Hong_Kong": 8,
        "Asia/Beijing": 8,
        "UTC": 0,
    }
    hours = offsets.get(tz, 8)
    now = datetime.now(timezone(timedelta(hours=hours)))
    return {
        "timezone": tz,
        "time": now.strftime("%H:%M:%S"),
        "date": now.strftime("%Y-%m-%d"),
    }


def calculate(expression: str) -> dict:
    """简单四则运算（仅允许数字与 +-*/()）。"""
    allowed = set("0123456789+-*/(). ")
    if not expression or not all(c in allowed for c in expression):
        return {"error": "表达式含非法字符"}
    try:
        result = eval(expression, {"__builtins__": {}}, {})  # noqa: S307
        return {"expression": expression, "result": result}
    except Exception as e:
        return {"error": str(e)}


def _fn_tool(name: str, description: str, properties: dict, required: list[str]) -> dict:
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }


weather_tool = _fn_tool(
    "get_weather",
    "查询指定城市的天气。",
    {"city": {"type": "string", "description": "城市名"}},
    ["city"],
)

time_tool = _fn_tool(
    "get_time",
    "查询指定时区的当前时间。",
    {"timezone": {"type": "string", "description": "IANA 时区，如 Asia/Shanghai"}},
    ["timezone"],
)

calculator_tool = _fn_tool(
    "calculate",
    "计算数学表达式。",
    {"expression": {"type": "string", "description": "如 25 * 4 + 10"}},
    ["expression"],
)

TOOLS = [weather_tool, time_tool, calculator_tool]

TOOL_HANDLERS = {
    "get_weather": lambda args: get_weather(args["city"]),
    "get_time": lambda args: get_time(args["timezone"]),
    "calculate": lambda args: calculate(args["expression"]),
}


def handle_multiple_tools(user_query: str) -> str:
    messages = [{"role": "user", "content": user_query}]

    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
    )

    assistant_msg = response.choices[0].message

    if not assistant_msg.tool_calls:
        return assistant_msg.content or ""

    messages.append({
        "role": "assistant",
        "content": assistant_msg.content,
        "tool_calls": [
            {
                "id": tc.id,
                "type": tc.type,
                "function": {
                    "name": tc.function.name,
                    "arguments": tc.function.arguments,
                },
            }
            for tc in assistant_msg.tool_calls
        ],
    })

    tool_results = []
    for tool_call in assistant_msg.tool_calls:
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments or "{}")
        handler = TOOL_HANDLERS.get(name)
        result = handler(args) if handler else {"error": f"Unknown tool: {name}"}
        print(f"  [tool] {name}({args}) -> {result}")

        tool_results.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result, ensure_ascii=False),
        })

    messages.extend(tool_results)

    final_response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=TOOLS,
    )

    final = final_response.choices[0].message
    if final.tool_calls:
        return final.content or "（模型请求继续调用工具，请再运行一轮）"
    return final.content or ""


if __name__ == "__main__":
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8")

    print(f"多工具并行示例（模型: {MODEL}）")
    print("输入问题，回车发送。直接回车使用默认问题。输入 q 退出。\n")

    default = "北京现在天气怎么样？现在几点了？顺便算一下 25 * 4 + 10"

    while True:
        try:
            query = input(">> ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if query.lower() == "q":
            break
        if not query:
            query = default
        print(handle_multiple_tools(query))
        print()

北京现在天气怎么样？现在几点了？顺便算一下 25 * 4 + 10

对比一下北京和上海的天气，哪个更热？

查一下深圳的天气，并计算 1000 / 25

6.3 实战3：处理后台任务 #

对应文件： t4.py（对齐 s13 思路）
要点： 慢操作派发到后台线程，立即返回占位 tool 消息；<task_notification> 必须在下一轮 LLM 调用前注入，不能插在 assistant 与 tool 之间。

# 指定使用 Python 3 的解释器
#!/usr/bin/env python3
# 文件头注释，说明本文件为 Tool Calls 的后台任务示例
"""Tool Calls 后台任务示例（对应 ToolCalls.md §6.3 / s13）：DeepSeek + 线程后台 + 通知注入。"""

# 导入处理 JSON 的模块
import json
# 导入操作系统相关的模块
import os
# 导入系统相关模块
import sys
# 导入线程相关的模块
import threading
# 导入时间相关的模块
import time

# 导入 dotenv 用于读取 .env 文件
from dotenv import load_dotenv
# 导入 OpenAI 接口库
from openai import OpenAI

# 加载 .env 文件中的环境变量，override=True 表示覆盖已存在的环境变量
load_dotenv(override=True)

# 获取 SSL 证书文件路径
_ssl_cert = os.getenv("SSL_CERT_FILE")
# 如果配置了 SSL 证书但对应文件不存在，则移除环境变量
if _ssl_cert and not os.path.isfile(_ssl_cert):
    os.environ.pop("SSL_CERT_FILE", None)

# 创建 OpenAI API 客户端，api_key 和 base_url 来自环境变量
client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.getenv("OPENAI_BASE_URL"),
)
# 获取模型名称
MODEL = os.environ["MODEL_ID"]

# DEMO_SLEEP 表示模拟耗时任务的秒数（默认 5）
DEMO_SLEEP = int(os.getenv("DEMO_SLEEP", "5"))

# 定义工具的元数据结构生成函数
def _fn_tool(name: str, description: str, properties: dict, required: list[str]) -> dict:
    # 返回符合 OpenAI 函数调用协议的工具描述字典
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }

# 定义一个模拟耗时任务的工具描述
long_running_tool = _fn_tool(
    "run_slow_task",
    "执行耗时任务。设置 run_in_background=true 时在后台运行，立即返回占位符。",
    {
        "task_name": {"type": "string", "description": "任务名称"},
        "seconds": {"type": "integer", "description": f"模拟耗时秒数，默认 {DEMO_SLEEP}"},
        "run_in_background": {"type": "boolean", "description": "是否在后台执行"},
    },
    ["task_name"],
)

# 定义一个计算数学表达式的工具描述
calculate_tool = _fn_tool(
    "calculate",
    "计算数学表达式。",
    {"expression": {"type": "string", "description": "如 123 + 456"}},
    ["expression"],
)

# 工具列表
TOOLS = [long_running_tool, calculate_tool]

# 定义模拟长时间任务的函数（阻塞线程）
def run_slow_task(task_name: str, seconds: int | None = None) -> dict:
    """模拟长时间任务（阻塞当前线程）。"""
    # 若提供 seconds 参数则用之，否则用默认值
    delay = seconds if seconds is not None else DEMO_SLEEP
    # 睡眠 delay 秒，模拟耗时
    time.sleep(delay)
    # 返回任务完成的信息
    return {
        "status": "completed",
        "task": task_name,
        "seconds": delay,
        "result": f"任务 {task_name} 已完成（耗时 {delay}s）",
    }

# 定义计算数学表达式的函数
def calculate(expression: str) -> dict:
    # 允许的字符集合
    allowed = set("0123456789+-*/(). ")
    # 判断表达式是否存在非法字符
    if not expression or not all(c in allowed for c in expression):
        return {"error": "表达式含非法字符"}
    try:
        # 使用 eval 计算表达式，禁用内置函数
        result = eval(expression, {"__builtins__": {}}, {})  # noqa: S307
        # 返回表达式和结果
        return {"expression": expression, "result": result}
    except Exception as e:
        # 捕获异常，返回错误信息
        return {"error": str(e)}

# 工具名称到处理函数的映射
TOOL_HANDLERS = {
    "run_slow_task": lambda args: run_slow_task(
        args["task_name"],
        args.get("seconds"),
    ),
    "calculate": lambda args: calculate(args["expression"]),
}

# 后台任务自增计数器
_bg_counter = 0
# 存储后台任务信息的字典
background_tasks: dict[str, dict] = {}
# 存储后台任务结果的字典
background_results: dict[str, dict] = {}
# 用于并发访问后台任务的锁
background_lock = threading.Lock()

# 判断工具是否应当在后台运行
def should_run_background(tool_name: str, tool_input: dict) -> bool:
    # 若参数设置 run_in_background，则后台执行
    if tool_input.get("run_in_background"):
        return True
    # 若是 run_slow_task 且 seconds 超过等于 3，则后台执行
    return tool_name == "run_slow_task" and tool_input.get("seconds", DEMO_SLEEP) >= 3

# 根据工具名称执行相应工具逻辑
def execute_tool(name: str, args: dict) -> dict:
    # 获取对应处理函数
    handler = TOOL_HANDLERS.get(name)
    # 若处理函数存在，调用
    if handler:
        return handler(args)
    # 否则返回未知工具错误
    return {"error": f"Unknown tool: {name}"}

# 启动一个后台任务
def start_background_task(tool_call_id: str, name: str, args: dict) -> str:
    # 声明使用全局计数器
    global _bg_counter
    # 自增计数，生成后台任务 ID
    _bg_counter += 1
    bg_id = f"bg_{_bg_counter:04d}"
    # 获取任务标签（用于显示）
    label = args.get("task_name") or args.get("command") or name

    # 定义线程工作函数
    def worker():
        # 执行具体工具
        result = execute_tool(name, args)
        # 上锁写入任务状态和结果
        with background_lock:
            background_tasks[bg_id]["status"] = "completed"
            background_results[bg_id] = result

    # 上锁添加任务到后台任务列表，设置状态
    with background_lock:
        background_tasks[bg_id] = {
            "tool_call_id": tool_call_id,
            "command": label,
            "status": "running",
        }
    # 启动后台线程，执行 worker
    threading.Thread(target=worker, daemon=True).start()
    # 打印后台任务派发信息
    print(f"  [后台] 已派发 {bg_id}: {label}")
    # 返回后台任务 ID
    return bg_id

# 收集所有已经完成的后台任务结果，并生成通知
def collect_background_results() -> list[str]:
    # 上锁，获取所有已完成的后台任务 ID
    with background_lock:
        ready_ids = [bid for bid, t in background_tasks.items() if t["status"] == "completed"]

    # 用于存放所有通知
    notifications = []
    # 遍历所有已完成的后台任务 ID
    for bg_id in ready_ids:
        # 上锁读取并删除任务及结果
        with background_lock:
            task = background_tasks.pop(bg_id)
            output = background_results.pop(bg_id, {})
        # 将输出内容转换为 JSON 字符串
        summary = json.dumps(output, ensure_ascii=False)
        # 如内容太长则截断
        if len(summary) > 200:
            summary = summary[:200] + "..."
        # 创建任务完成通知字符串
        notifications.append(
            f"<task_notification>\n"
            f"  <task_id>{bg_id}</task_id>\n"
            f"  <status>completed</status>\n"
            f"  <command>{task['command']}</command>\n"
            f"  <summary>{summary}</summary>\n"
            f"</task_notification>"
        )
        # 打印后台任务完成日志
        print(f"  [后台完成] {bg_id}: {task['command']}")
    # 返回所有通知
    return notifications

# 支持后台任务的主处理流程，处理用户输入并多轮与模型交互
def process_with_background_tasks(user_query: str) -> str:
    # 初始化消息列表，加入用户输入
    messages = [{"role": "user", "content": user_query}]

    # 无限循环，直到模型给出最终回复
    while True:
        # 收集后台任务完成的通知
        notifications = collect_background_results()
        # 如果有通知，注入进消息列表
        if notifications:
            messages.append({
                "role": "user",
                "content": "\n\n".join(notifications),
            })
            # 打印注入后台通知数
            print(f"  [注入] {len(notifications)} 条后台通知")

        # 向 OpenAI 模型发起聊天补全请求
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )

        # 取出助手的消息
        assistant_msg = response.choices[0].message

        # 将助手回复（包括 tools）加入消息列表
        messages.append({
            "role": "assistant",
            "content": assistant_msg.content,
            **({"tool_calls": [
                {
                    "id": tc.id,
                    "type": tc.type,
                    "function": {
                        "name": tc.function.name,
                        "arguments": tc.function.arguments,
                    },
                }
                for tc in assistant_msg.tool_calls
            ]} if assistant_msg.tool_calls else {}),
        })

        # 如果没有工具调用，则表明会话结束，直接返回内容
        if not assistant_msg.tool_calls:
            return assistant_msg.content or ""

        # 逐个处理工具调用
        for tool_call in assistant_msg.tool_calls:
            # 获取工具名称
            name = tool_call.function.name
            # 解析工具调用的参数
            args = json.loads(tool_call.function.arguments or "{}")
            # 打印当前调用的工具
            print(f"> {name}")

            # 判定是否需要后台调度
            if should_run_background(name, args):
                # 启动后台任务
                bg_id = start_background_task(tool_call.id, name, args)
                # 构造后台任务启动的返回内容
                output = {
                    "background_task_id": bg_id,
                    "status": "running",
                    "message": "任务已在后台启动，完成后将通过 task_notification 通知。",
                }
            else:
                # 前台直接执行，并打印结果
                output = execute_tool(name, args)
                print(f"  [sync] {output}")

            # 把工具执行结果写入消息流
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(output, ensure_ascii=False),
            })

        # 等待本轮后台任务全部完成，避免多轮嵌套产生竞态
        while True:
            # 上锁判断是否仍有运行中的后台任务
            with background_lock:
                running = any(t["status"] == "running" for t in background_tasks.values())
            # 若都已完成则退出等待循环
            if not running:
                break
            # 否则短暂睡眠后再检查
            time.sleep(0.2)

# 主程序入口
if __name__ == "__main__":
    # 如果在 Windows 平台，确保标准输出为 UTF-8 编码
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8")

    # 打印提示
    print(f"后台任务示例（模型: {MODEL}，模拟耗时: {DEMO_SLEEP}s）")
    print("输入问题，回车发送。直接回车使用默认问题。输入 q 退出。\n")

    # 定义默认问题，用于快捷演示
    default = (
        f"用 run_in_background 在后台执行 run_slow_task，任务名 pip-sim，耗时 {DEMO_SLEEP} 秒；"
        "同时用 calculate 计算 123 + 456"
    )

    # 进入主循环，持续获取用户输入
    while True:
        try:
            # 获取用户输入，并去掉首尾空白
            query = input(">> ").strip()
        except (EOFError, KeyboardInterrupt):
            # 响应 Ctrl+C/EOF，退出主循环
            break
        # 输入 q 则退出
        if query.lower() == "q":
            break
        # 若输入为空则使用默认问题
        if not query:
            query = default
        # 处理并返回模型结果
        print(process_with_background_tasks(query))
        # 打印空行分割多次对话
        print()

用 run_in_background 在后台执行 run_slow_task，任务名 pip-sim，耗时 5 秒；同时用 calculate 计算 123 + 456

在后台运行一个名为 data-export 的慢任务，耗时 8 秒，run_in_background 设为 true。启动后先告诉我任务 ID，不要干等。

先同步执行 run_slow_task，任务名 quick-check，耗时 2 秒（不要设 run_in_background）；再计算 (100 + 50) * 3

7. 高级技巧 #

§6 的示例都是「一轮 tool_calls → 汇总」。真实 Agent 还需要：多轮链式调用、并行加速、失败重试、流式收集、发送前校验。以下脚本均可独立运行，共用 tool_demo_common.py。

7.1 工具调用链（Chain of Tools） #

对应文件： 工具调用链（Chain of Tools）.py
与 t3 的区别： while 循环最多 N 轮——上一轮 tool 结果进入 history 后，模型可再次发起 tool_calls（例如先查天气、再算表达式、再查时间）。

# 指定使用 Python 3 解释器
#!/usr/bin/env python3
# 模块文档：Tool Calls 第 7.1 节工具调用链（多轮循环）
"""Tool Calls 工具调用链示例（对应 ToolCalls.md §7.1）：DeepSeek + 多轮循环。"""

# 导入 json 模块
import json
# 导入 os 模块
import os
# 导入 sys 模块
import sys
# 导入 datetime 相关类
from datetime import datetime, timedelta, timezone

# 导入 dotenv 加载环境变量
from dotenv import load_dotenv
# 导入 OpenAI 客户端
from openai import OpenAI

# 加载 .env 文件
load_dotenv(override=True)

# 读取 SSL 证书路径环境变量
_ssl_cert = os.getenv("SSL_CERT_FILE")
# 若证书路径无效则从环境中移除
if _ssl_cert and not os.path.isfile(_ssl_cert):
    os.environ.pop("SSL_CERT_FILE", None)

# 创建 OpenAI 客户端（DeepSeek 兼容接口）
client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.getenv("OPENAI_BASE_URL"),
)
# 从环境变量读取模型 ID
MODEL = os.environ["MODEL_ID"]
# 工具调用链最大循环轮数
MAX_ITERATIONS = 5


# 定义模拟天气查询函数
def get_weather(city: str) -> dict:
    # 函数说明
    """模拟天气 API。"""
    # 预设城市天气数据
    data = {
        "Beijing": {"temp": 25, "condition": "sunny"},
        "Shanghai": {"temp": 30, "condition": "rainy"},
        "北京": {"temp": 25, "condition": "sunny"},
        "上海": {"temp": 30, "condition": "rainy"},
    }
    # 返回对应城市数据，未知城市返回 unknown
    return data.get(city, {"temp": "unknown", "condition": "unknown"})


# 定义获取指定时区时间的函数
def get_time(tz: str) -> dict:
    # 函数说明
    """返回指定时区当前时间。"""
    # 常用时区相对 UTC 的小时偏移
    offsets = {
        "Asia/Shanghai": 8,
        "Asia/Hong_Kong": 8,
        "Asia/Beijing": 8,
        "UTC": 0,
    }
    # 取偏移，默认东八区
    hours = offsets.get(tz, 8)
    # 构造带时区的当前时间
    now = datetime.now(timezone(timedelta(hours=hours)))
    # 返回时区、时分秒、日期
    return {
        "timezone": tz,
        "time": now.strftime("%H:%M:%S"),
        "date": now.strftime("%Y-%m-%d"),
    }


# 定义简单四则运算函数
def calculate(expression: str) -> dict:
    # 函数说明
    """简单四则运算。"""
    # 允许的字符集合
    allowed = set("0123456789+-*/(). ")
    # 表达式为空或含非法字符则报错
    if not expression or not all(c in allowed for c in expression):
        return {"error": "表达式含非法字符"}
    # 尝试 eval 计算
    try:
        result = eval(expression, {"__builtins__": {}}, {})  # noqa: S307
        return {"expression": expression, "result": result}
    # 捕获计算异常
    except Exception as e:
        return {"error": str(e)}


# 定义 OpenAI tools schema 构造辅助函数
def _fn_tool(name: str, description: str, properties: dict, required: list[str]) -> dict:
    # 返回标准 function tool 描述 dict
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }


# 注册三个可用工具 schema
available_tools = [
    _fn_tool(
        "get_weather",
        "查询指定城市的天气。",
        {"city": {"type": "string", "description": "城市名"}},
        ["city"],
    ),
    _fn_tool(
        "get_time",
        "查询指定时区的当前时间。",
        {"timezone": {"type": "string", "description": "IANA 时区，如 Asia/Shanghai"}},
        ["timezone"],
    ),
    _fn_tool(
        "calculate",
        "计算数学表达式。",
        {"expression": {"type": "string", "description": "如 25 * 4 + 10"}},
        ["expression"],
    ),
]

# 工具名到实现函数的映射
TOOL_HANDLERS = {
    "get_weather": lambda args: get_weather(args["city"]),
    "get_time": lambda args: get_time(args["timezone"]),
    "calculate": lambda args: calculate(args["expression"]),
}


# 执行单个 tool_call
def execute_tool(tool_call) -> dict:
    # 函数说明
    """执行单次 tool_call，返回 dict 结果。"""
    # 读取工具名
    name = tool_call.function.name
    # 解析 JSON 参数字符串
    args = json.loads(tool_call.function.arguments or "{}")
    # 查找处理函数
    handler = TOOL_HANDLERS.get(name)
    # 若存在则调用
    if handler:
        return handler(args)
    # 未知工具返回错误 dict
    return {"error": f"Unknown tool: {name}"}


# 将 SDK Message 转为可序列化的 assistant dict
def _assistant_message_dict(message) -> dict:
    # 导出 message 字段，忽略 None
    data = message.model_dump(exclude_none=True)
    # 显式设置 role
    data["role"] = "assistant"
    return data


# 工具调用链主函数：多轮直到无 tool_calls 或达上限
def chain_tool_calls(user_query: str) -> str:
    # 初始化 messages
    messages = [{"role": "user", "content": user_query}]

    # 最多循环 MAX_ITERATIONS 轮
    for i in range(1, MAX_ITERATIONS + 1):
        # 打印当前轮次
        print(f"\n--- 第 {i} 轮 ---")
        # 调用 LLM
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=available_tools,
            tool_choice="auto",
        )

        # 取出 assistant 消息
        assistant_msg = response.choices[0].message

        # 无 tool_calls 则返回最终文本
        if not assistant_msg.tool_calls:
            return assistant_msg.content or ""

        # 追加 assistant 消息到 history
        messages.append(_assistant_message_dict(assistant_msg))

        # 执行本轮所有 tool_calls
        for tool_call in assistant_msg.tool_calls:
            result = execute_tool(tool_call)
            print(f"  [tool] {tool_call.function.name} -> {result}")
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })

    # 超过最大轮数则返回提示
    return "Max iterations reached"


# 脚本直接运行
if __name__ == "__main__":
    # Windows 下配置 UTF-8 输出
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8")

    # 打印说明
    print(f"工具调用链示例（模型: {MODEL}，最多 {MAX_ITERATIONS} 轮）")
    print("输入问题，回车发送。直接回车使用默认问题。输入 q 退出。\n")

    # 默认问题：需多轮链式调用（天气 → 计算 → 时间）
    default = (
        "先查北京的天气；如果温度高于 20，就计算 25 * 4 + 10，"
        "否则计算 10 + 10；最后告诉我现在北京时间"
    )

    # 交互主循环
    while True:
        try:
            query = input(">> ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if query.lower() == "q":
            break
        if not query:
            query = default
        print(chain_tool_calls(query))
        print()

7.2 并行工具执行优化 #

对应文件： 并行工具执行优化.py
要点： 同一轮多个 tool_calls 彼此无依赖时，用 ThreadPoolExecutor 并行执行，缩短 wall-clock；append 到 messages 的顺序仍须与 tool_calls 数组一致。

# 指定使用 Python 3 解释器
#!/usr/bin/env python3
# 模块文档：Tool Calls 第 7.2 节并行工具执行
"""Tool Calls §7.2：并行工具执行优化。"""

# 导入 json 模块
import json
# 导入线程池，用于并行执行多个工具
from concurrent.futures import ThreadPoolExecutor

# 从共用模块导入客户端、工具列表等
from tool_demo_common import (
    MODEL,
    assistant_dict,
    available_tools,
    client,
    execute_tool,
    setup_stdout,
)


# 定义并行执行多个 tool_call 的函数
def execute_tools_parallel(tool_calls) -> list[dict]:
    # 创建最多 5 个工作线程的线程池
    with ThreadPoolExecutor(max_workers=5) as executor:
        # 为每个 tool_call 提交异步任务，保存 (id, future) 对
        futures = [(tc.id, executor.submit(execute_tool, tc)) for tc in tool_calls]
        # 存放最终 tool 消息列表
        results = []
        # 按提交顺序等待每个 future 完成
        for tool_call_id, future in futures:
            # 阻塞获取工具执行结果
            result = future.result()
            # 构造符合 API 的 tool 消息
            results.append({
                "role": "tool",
                "tool_call_id": tool_call_id,
                "content": json.dumps(result, ensure_ascii=False),
            })
        # 返回全部 tool 消息
        return results


# 定义主流程：用户提问 → 并行工具 → 最终回答
def run(user_query: str) -> str:
    # 初始化消息列表，加入用户消息
    messages = [{"role": "user", "content": user_query}]

    # 第一次调用 LLM，获取 tool_calls 决策
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=available_tools,
        tool_choice="auto",
    )
    # 取出 assistant 消息
    assistant_msg = response.choices[0].message

    # 若无 tool_calls，直接返回文本
    if not assistant_msg.tool_calls:
        return assistant_msg.content or ""

    # 将 assistant 消息（含 tool_calls）追加到 history
    messages.append(assistant_dict(assistant_msg))
    # 打印并行执行的工具数量
    print(f"  并行执行 {len(assistant_msg.tool_calls)} 个工具…")
    # 线程池并行执行所有工具
    tool_results = execute_tools_parallel(assistant_msg.tool_calls)

    # 逐条打印 tool 结果摘要
    for tr in tool_results:
        print(f"  [tool] {tr['tool_call_id']}: {tr['content'][:80]}")

    # 将全部 tool 消息追加到 history
    messages.extend(tool_results)

    # 第二次调用 LLM，生成最终汇总回答
    final = client.chat.completions.create(
        model=MODEL, messages=messages, tools=available_tools,
    )
    # 返回最终 assistant 文本
    return final.choices[0].message.content or ""


# 脚本直接运行时进入交互 REPL
if __name__ == "__main__":
    # 配置 stdout UTF-8
    setup_stdout()
    # 打印程序说明
    print(f"并行工具执行（模型: {MODEL}）")
    print("输入问题，回车发送。直接回车使用默认问题。输入 q 退出。\n")

    # 默认测试问题（三工具并行）
    default = "北京现在天气怎么样？现在几点了？顺便算一下 25 * 4 + 10"

    # 主循环
    while True:
        # 读取用户输入
        try:
            query = input(">> ").strip()
        # Ctrl+C / Ctrl+D 退出
        except (EOFError, KeyboardInterrupt):
            break
        # 输入 q 退出
        if query.lower() == "q":
            break
        # 空输入使用默认问题
        if not query:
            query = default
        # 执行并打印结果
        print(run(query))
        # 空行分隔
        print()

7.3 错误处理和重试 #

对应文件： 错误处理和重试.py
要点： 工具执行可能因网络/磁盘等瞬态错误失败；对 execute_tool 包一层指数退避重试，耗尽次数后仍返回 tool 消息（含 error JSON），避免破坏 tool 配对。

# 指定使用 Python 3 解释器
#!/usr/bin/env python3
# 模块文档：Tool Calls 第 7.3 节错误处理和重试
"""Tool Calls §7.3：错误处理和重试。"""

# 导入 json 模块
import json
# 导入 time 模块，用于重试退避 sleep
import time

# 从共用模块导入模型名、工具执行函数、stdout 配置
from tool_demo_common import MODEL, execute_tool, setup_stdout

# 记录每个 tool_call_id 已失败次数的全局字典
_fail_counts: dict[str, int] = {}


# 定义带重试的工具执行函数
def execute_tool_with_retry(
    tool_call,
    executor=None,
    max_retries: int = 3,
) -> dict:
    # 若未传入 executor 则使用默认 execute_tool
    run = executor or execute_tool
    # 最多尝试 max_retries 次
    for attempt in range(max_retries):
        # 尝试执行工具
        try:
            result = run(tool_call)
            # 成功则构造标准 tool 消息并返回
            return {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False),
            }
        # 捕获任意执行异常
        except Exception as e:
            # 若已是最后一次尝试
            if attempt == max_retries - 1:
                # 返回包含错误信息的 tool 消息
                return {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps({
                        "error": str(e),
                        "message": "Tool execution failed after retries",
                    }, ensure_ascii=False),
                }
            # 计算指数退避延迟（1s, 2s, 4s…）
            delay = 2 ** attempt
            # 打印重试日志
            print(f"  [retry] 第 {attempt + 1} 次失败，{delay}s 后重试: {e}")
            # 等待后进入下一轮重试
            time.sleep(delay)


# 定义模拟不稳定的工具执行：前 2 次失败，第 3 次成功
def flaky_execute_tool(tool_call) -> dict:
    # 函数说明
    """模拟前 2 次失败、第 3 次成功。"""
    # 以 tool_call.id 作为计数键
    key = tool_call.id
    # 累加该 ID 的调用次数
    _fail_counts[key] = _fail_counts.get(key, 0) + 1
    # 前两次调用故意抛错
    if _fail_counts[key] < 3:
        raise RuntimeError(f"模拟瞬态错误（第 {_fail_counts[key]} 次）")
    # 第三次起调用真实 execute_tool
    return execute_tool(tool_call)


# 定义演示入口
def run_demo():
    # 打印标题
    print(f"错误处理和重试（模型: {MODEL}）\n")

    # 定义简易 tool_call 占位类
    class _TC:
        pass

    # 创建模拟 tool_call 实例
    tc = _TC()
    # 设置 tool_call_id
    tc.id = "call_demo"
    # 构造 function 对象（name + arguments）
    tc.function = type("F", (), {
        "name": "get_weather",
        "arguments": '{"city": "北京"}',
    })()

    # 使用 flaky 执行器调用带重试的执行函数
    msg = execute_tool_with_retry(tc, executor=flaky_execute_tool, max_retries=3)
    # 打印最终 tool 消息 content
    print(f"  最终结果: {msg['content']}")


# 脚本直接运行时执行
if __name__ == "__main__":
    # 配置 stdout 编码
    setup_stdout()
    # 运行演示
    run_demo()

7.4 流式处理工具调用 #

对应文件： 流式处理工具调用.py
要点： stream=True 时 tool_calls 分片到达，需按 index 合并 name / arguments 片段后再执行；文本 delta 与 tool delta 可能交错出现。

# 指定使用 Python 3 解释器
#!/usr/bin/env python3
# 模块文档：Tool Calls 第 7.4 节流式处理工具调用
"""Tool Calls §7.4：流式处理工具调用。"""

# 导入 json 模块
import json

# 从共用模块导入客户端与工具
from tool_demo_common import (
    MODEL,
    available_tools,
    client,
    execute_tool,
    setup_stdout,
)


# 定义函数：合并流式响应中分片的 tool_calls
def _merge_stream_tool_calls(chunks: list) -> list[dict]:
    # 函数说明
    """合并流式 delta 中的 tool_calls 片段。"""
    # 按 index 累积每个 tool_call 的片段
    acc: dict[int, dict] = {}
    # 遍历每个 delta 中的 tool_call 片段
    for tc in chunks:
        # 获取片段所属的 tool_call 序号
        idx = tc.index
        # 若该序号尚未初始化
        if idx not in acc:
            # 创建空的 tool_call 结构
            acc[idx] = {"id": "", "type": "function", "function": {"name": "", "arguments": ""}}
        # 若本片段带有 id 则更新
        if tc.id:
            acc[idx]["id"] = tc.id
        # 若本片段带有 function 字段
        if tc.function:
            # 累加 function.name 片段
            if tc.function.name:
                acc[idx]["function"]["name"] += tc.function.name
            # 累加 function.arguments 片段
            if tc.function.arguments:
                acc[idx]["function"]["arguments"] += tc.function.arguments
    # 按 index 排序后返回完整 tool_calls 列表
    return [acc[i] for i in sorted(acc)]


# 定义主流程：流式 API 收集 tool_calls 并执行
def stream_with_tools(user_query: str) -> str:
    # 初始化 messages
    messages = [{"role": "user", "content": user_query}]

    # 开启流式 completions 请求
    stream = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=available_tools,
        tool_choice="auto",
        stream=True,
    )

    # 收集 tool_call 分片
    delta_tool_chunks = []
    # 收集 assistant 文本分片
    text_parts = []

    # 逐 chunk 消费流
    for chunk in stream:
        # 跳过无 choices 的空 chunk
        if not chunk.choices:
            continue
        # 取 delta 对象
        delta = chunk.choices[0].delta
        # 若有文本增量则追加
        if delta.content:
            text_parts.append(delta.content)
        # 若有 tool_calls 增量则追加
        if delta.tool_calls:
            delta_tool_chunks.extend(delta.tool_calls)

    # 若收集到流式文本则打印预览
    if text_parts:
        print("  [stream text]", "".join(text_parts)[:80])

    # 合并分片为完整 tool_calls
    merged = _merge_stream_tool_calls(delta_tool_chunks)
    # 若无 tool_calls 则直接返回文本
    if not merged:
        return "".join(text_parts) or ""

    # 打印合并后的 tool_call 数量
    print(f"  [stream] 收集到 {len(merged)} 个 tool_call")

    # 构造 assistant 消息并追加（含 tool_calls）
    messages.append({
        "role": "assistant",
        "content": "".join(text_parts) or None,
        "tool_calls": merged,
    })

    # 逐个执行合并后的 tool_call
    for tc in merged:
        # 定义简易对象模拟 SDK 的 tool_call 结构
        class _TC:
            pass
        # 创建实例
        obj = _TC()
        # 设置 id
        obj.id = tc["id"]
        # 动态构造 function 属性
        obj.function = type("F", (), {"name": tc["function"]["name"], "arguments": tc["function"]["arguments"]})()
        # 执行工具
        result = execute_tool(obj)
        # 打印执行结果
        print(f"  [tool] {tc['function']['name']} -> {result}")
        # 追加 tool 消息
        messages.append({
            "role": "tool",
            "tool_call_id": tc["id"],
            "content": json.dumps(result, ensure_ascii=False),
        })

    # 非流式第二次调用，获取最终回答
    final = client.chat.completions.create(
        model=MODEL, messages=messages, tools=available_tools,
    )
    # 返回最终文本
    return final.choices[0].message.content or ""


# 脚本直接运行
if __name__ == "__main__":
    setup_stdout()
    print(f"流式工具调用（模型: {MODEL}）")
    print("输入问题，回车发送。直接回车使用默认问题。输入 q 退出。\n")

    default = "北京现在天气怎么样？现在几点了？"

    while True:
        try:
            query = input(">> ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if query.lower() == "q":
            break
        if not query:
            query = default
        print(stream_with_tools(query))
        print()

7.5 工具响应格式验证 #

对应文件： 工具响应格式验证.py
要点： 在 append 到 messages 前校验 role / tool_call_id / content；content 非字符串时自动 json.dumps，避免 API 400。

# 指定使用 Python 3 解释器
#!/usr/bin/env python3
# 模块文档：Tool Calls 第 7.5 节工具响应格式验证
"""Tool Calls §7.5：工具响应格式验证。"""

# 导入 json 模块，用于序列化 content
import json

# 从共用模块导入模型名与 stdout 配置函数
from tool_demo_common import MODEL, setup_stdout


# 定义工具响应验证函数
def validate_tool_response(response: dict, tool_call_id: str) -> dict:
    # 函数说明：验证 tool 消息是否符合 OpenAI API 要求
    """验证工具响应是否符合 API 要求。"""
    # 定义 tool 消息必须包含的字段列表
    required_fields = ["role", "tool_call_id", "content"]

    # 遍历每个必需字段
    for field in required_fields:
        # 若响应中缺少该字段
        if field not in response:
            # 抛出 ValueError 说明缺失字段名
            raise ValueError(f"Missing required field: {field}")

    # 若 role 不是 tool
    if response["role"] != "tool":
        # 抛出 ValueError 说明非法 role
        raise ValueError(f"Invalid role: {response['role']}")

    # 若 tool_call_id 与期望 ID 不一致
    if response["tool_call_id"] != tool_call_id:
        # 抛出 ID 不匹配错误
        raise ValueError("Tool call ID mismatch")

    # 若 content 不是字符串类型
    if not isinstance(response["content"], str):
        # 将 content 自动转为 JSON 字符串
        response["content"] = json.dumps(response["content"], ensure_ascii=False)

    # 返回校验（可能已修正）后的响应字典
    return response


# 定义单次测试用例运行函数
def _check(title: str, fn):
    # 尝试执行传入的测试函数
    try:
        fn()
        # 未抛异常则打印通过
        print(f"[{title}] ✓ 通过")
    # 捕获预期的 ValueError
    except ValueError as e:
        # 打印正确拒绝及原因
        print(f"[{title}] ✓ 正确拒绝: {e}")


# 定义演示入口：运行多组验证用例
def run_demo():
    # 打印标题与当前模型名
    print(f"工具响应格式验证（模型: {MODEL}）\n")

    # 测试用例 1：合法响应
    _check("合法响应", lambda: validate_tool_response({
        "role": "tool", "tool_call_id": "call_abc", "content": '{"temp": 25}',
    }, "call_abc"))

    # 测试用例 2：缺少 content 字段
    _check("缺少 content", lambda: validate_tool_response({
        "role": "tool", "tool_call_id": "call_abc",
    }, "call_abc"))

    # 测试用例 3：role 应为 tool 却为 user
    _check("role 错误", lambda: validate_tool_response({
        "role": "user", "tool_call_id": "call_abc", "content": "ok",
    }, "call_abc"))

    # 测试用例 4：tool_call_id 与期望不符
    _check("ID 不匹配", lambda: validate_tool_response({
        "role": "tool", "tool_call_id": "call_xyz", "content": "ok",
    }, "call_abc"))

    # 测试用例 5：content 为 dict，应自动转 JSON
    fixed = validate_tool_response({
        "role": "tool", "tool_call_id": "call_abc", "content": {"temp": 25},
    }, "call_abc")
    # 打印自动转换后的 content
    print(f"[content 自动转 JSON] ✓ 通过 → {fixed['content']}")


# 脚本直接运行时执行
if __name__ == "__main__":
    # 配置 Windows 控制台 UTF-8 输出
    setup_stdout()
    # 运行全部演示用例
    run_demo()

8. 调试技巧 #

报错信息往往只指向「配对不完整」，不会告诉你哪一条 assistant 缺 tool。下面两个工具函数在开发阶段打印/校验完整 messages，发送 API 前跑一遍可省大量 400。

8.1 打印完整消息链 #

对应文件： 打印完整消息链.py
作用： 在第一次 LLM 调用并执行 tool 后，把当前 messages 逐条打印（role、content 摘要、tool_calls、tool_call_id），直观对照 §3 顺序。

8.1.1 tool_demo_common #

§7–§8 示例共用的 DeepSeek 客户端与三工具（get_weather / get_time / calculate），避免每个脚本重复配置。

# 工具调用示例共用：DeepSeek 客户端和三个工具。
"""Tool Calls 示例共用：DeepSeek 客户端 + 三工具。"""

# 导入json模块
import json
# 导入os模块
import os
# 从datetime模块中导入datetime、timedelta和timezone类
from datetime import datetime, timedelta, timezone

# 从dotenv模块导入load_dotenv函数
from dotenv import load_dotenv
# 从openai模块导入OpenAI类
from openai import OpenAI

# 加载环境变量，覆盖已存在变量
load_dotenv(override=True)

# 获取名为SSL_CERT_FILE的环境变量
_ssl_cert = os.getenv("SSL_CERT_FILE")
# 如果环境变量SSL_CERT_FILE存在且指定的文件不存在
if _ssl_cert and not os.path.isfile(_ssl_cert):
    # 从环境变量中移除SSL_CERT_FILE
    os.environ.pop("SSL_CERT_FILE", None)

# 创建OpenAI客户端对象
client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],  # 从环境变量获取API密钥
    base_url=os.getenv("OPENAI_BASE_URL"), # 获取基础URL
)
# 获取模型ID
MODEL = os.environ["MODEL_ID"]


# 定义获取天气的函数
def get_weather(city: str) -> dict:
    # 定义城市对应的天气示例数据
    data = {
        "Beijing": {"temp": 25, "condition": "sunny"},
        "Shanghai": {"temp": 30, "condition": "rainy"},
        "北京": {"temp": 25, "condition": "sunny"},
        "上海": {"temp": 30, "condition": "rainy"},
    }
    # 返回对应城市的天气数据，找不到则返回默认值
    return data.get(city, {"temp": "unknown", "condition": "unknown"})


# 定义获取当前时间的函数
def get_time(tz: str) -> dict:
    # 支持的时区与对应的时差
    offsets = {"Asia/Shanghai": 8, "Asia/Hong_Kong": 8, "UTC": 0}
    # 获取时区对应的小时
    hours = offsets.get(tz, 8)
    # 获取当前时区的当前时间
    now = datetime.now(timezone(timedelta(hours=hours)))
    # 返回时区、时间和日期
    return {"timezone": tz, "time": now.strftime("%H:%M:%S"), "date": now.strftime("%Y-%m-%d")}


# 定义计算数学表达式的函数
def calculate(expression: str) -> dict:
    # 允许的字符集合
    allowed = set("0123456789+-*/(). ")
    # 检查表达式是否为空或含有非法字符
    if not expression or not all(c in allowed for c in expression):
        return {"error": "表达式含非法字符"}
    try:
        # 安全地计算表达式的结果
        result = eval(expression, {"__builtins__": {}}, {})  # noqa: S307
        # 返回表达式和结果
        return {"expression": expression, "result": result}
    except Exception as e:
        # 返回错误信息
        return {"error": str(e)}


# 定义函数类型工具的描述函数
def _fn_tool(name: str, description: str, properties: dict, required: list[str]) -> dict:
    # 返回OpenAI工具格式字典
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {"type": "object", "properties": properties, "required": required},
        },
    }


# 工具列表，每个工具都定义了参数和功能描述
available_tools = [
    _fn_tool("get_weather", "查询指定城市的天气。", {"city": {"type": "string"}}, ["city"]),
    _fn_tool("get_time", "查询指定时区当前时间。", {"timezone": {"type": "string"}}, ["timezone"]),
    _fn_tool("calculate", "计算数学表达式。", {"expression": {"type": "string"}}, ["expression"]),
]

# 工具处理函数的映射
TOOL_HANDLERS = {
    "get_weather": lambda args: get_weather(args["city"]),
    "get_time": lambda args: get_time(args["timezone"]),
    "calculate": lambda args: calculate(args["expression"]),
}


# 定义执行工具的函数
def execute_tool(tool_call) -> dict:
    # 获取工具名称
    name = tool_call.function.name
    # 解析工具参数（JSON字符串）
    args = json.loads(tool_call.function.arguments or "{}")
    # 查找工具处理函数
    handler = TOOL_HANDLERS.get(name)
    # 如果找到处理函数则执行，否则返回错误
    if handler:
        return handler(args)
    return {"error": f"Unknown tool: {name}"}


# 定义将assistant消息转为dict的函数
def assistant_dict(message) -> dict:
    # 转为字典并排除None值
    data = message.model_dump(exclude_none=True)
    # 设置角色为assistant
    data["role"] = "assistant"
    # 返回字典数据
    return data


# 定义Windows下标准输出编码重设函数
def setup_stdout():
    # 导入sys模块
    import sys
    # 如果操作系统为Windows，则重新设置标准输出的编码为utf-8
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8")

8.1.2 code.py #

在共用模块之上调用 LLM、执行 tool，并在第二次 API 调用前调用 debug_messages(messages) 输出完整链路。

# 指定使用 Python 3 解释器
#!/usr/bin/env python3
# 模块文档：Tool Calls 第 8.1 节打印完整消息链
"""Tool Calls §8.1：打印完整消息链。"""

# 导入 json 模块
import json

# 从共用模块导入所需组件
from tool_demo_common import (
    MODEL,
    assistant_dict,
    available_tools,
    client,
    execute_tool,
    setup_stdout,
)


# 定义调试函数：格式化打印 messages 列表
def debug_messages(messages: list):
    # 打印分隔线
    print("=" * 50)
    # 打印标题
    print("MESSAGE CHAIN:")
    # 遍历每条消息及其索引
    for i, msg in enumerate(messages):
        # 打印序号与 role
        print(f"\n[{i}] Role: {msg['role']}")
        # 若有 content 则打印前 100 字符
        if msg.get("content"):
            text = str(msg["content"])
            print(f"    Content: {text[:100]}{'...' if len(text) > 100 else ''}")
        # 若有 tool_calls 则打印数量与详情
        if msg.get("tool_calls"):
            print(f"    Tool Calls: {len(msg['tool_calls'])}")
            for tc in msg["tool_calls"]:
                name = tc["function"]["name"]
                print(f"        - {name} ({tc['id']})")
        # 若是 tool 消息则打印 tool_call_id
        if msg.get("tool_call_id"):
            print(f"    Tool Call ID: {msg['tool_call_id']}")
    # 打印结束分隔线
    print("=" * 50)


# 定义主流程
def run(user_query: str) -> str:
    # 初始化 messages
    messages = [{"role": "user", "content": user_query}]

    # 第一次 LLM 调用
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=available_tools,
        tool_choice="auto",
    )
    assistant_msg = response.choices[0].message

    # 无 tool_calls：打印当前链并返回
    if not assistant_msg.tool_calls:
        debug_messages(messages)
        return assistant_msg.content or ""

    # 追加 assistant 消息
    messages.append(assistant_dict(assistant_msg))

    # 逐个同步执行工具
    for tool_call in assistant_msg.tool_calls:
        result = execute_tool(tool_call)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result, ensure_ascii=False),
        })

    # 工具执行后打印完整消息链（调试用）
    debug_messages(messages)

    # 第二次 LLM 调用获取最终回答
    final = client.chat.completions.create(
        model=MODEL, messages=messages, tools=available_tools,
    )
    return final.choices[0].message.content or ""


# 脚本直接运行
if __name__ == "__main__":
    setup_stdout()
    print(f"打印完整消息链（模型: {MODEL}）")
    print("输入问题，回车发送。直接回车使用默认问题。输入 q 退出。\n")

    default = "北京的天气怎么样？"

    while True:
        try:
            query = input(">> ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if query.lower() == "q":
            break
        if not query:
            query = default
        print(run(query))
        print()

8.2 验证消息完整性 #

对应文件： 验证消息完整性.py
要点： 扫描每条带 tool_calls 的 assistant，检查后续是否凑齐全部 tool_call_id；遇到下一个 user/assistant 即停止向后查找，模拟 API 的配对窗口。

# 指定 Python 解释器为 python3
#!/usr/bin/env python3
# 文件文档字符串，说明本文件用于工具调用 §8.2：验证消息完整性
"""Tool Calls §8.2：验证消息完整性。"""

# 从 tool_demo_common 模块中导入 MODEL 和 setup_stdout
from tool_demo_common import MODEL, setup_stdout

# 定义函数 validate_message_chain，参数为消息链列表，返回值为字符串列表
def validate_message_chain(messages: list) -> list[str]:
    # 函数文档字符串，说明功能是：验证消息链是否符合 API 要求。
    """验证消息链是否符合 API 要求。"""
    # 初始化一个空列表，用于保存错误信息
    errors = []

    # 遍历所有消息，i 为索引，msg 为元素
    for i, msg in enumerate(messages):
        # 如果当前消息是 assistant 且包含 tool_calls 字段
        if msg.get("role") == "assistant" and msg.get("tool_calls"):
            # 提取所有 tool_calls 的 id，组成一个列表
            tool_call_ids = [tc["id"] for tc in msg["tool_calls"]]

            # 初始化一个空列表，用于保存后续 tool 响应的 tool_call_id
            tool_responses = []
            # 从当前消息的下一个消息开始遍历
            for j in range(i + 1, len(messages)):
                # 如果后续消息的角色是 tool
                if messages[j].get("role") == "tool":
                    # 将该 tool 消息的 tool_call_id 加入 tool_responses
                    tool_responses.append(messages[j].get("tool_call_id"))
                # 如果遇到 assistant 或 user 角色，跳出循环
                elif messages[j].get("role") in ("assistant", "user"):
                    break

            # 检查每一个 tool_call_id，是否都在 tool_responses 列表中
            for tc_id in tool_call_ids:
                # 如果 tool_call_id 不在 tool_responses 中，记录错误
                if tc_id not in tool_responses:
                    errors.append(f"Missing tool response for {tc_id}")

    # 返回错误列表
    return errors

# 定义主演示函数
def run_demo():
    # 打印验证消息完整性的提示信息，展示模型名称
    print(f"验证消息完整性（模型: {MODEL}）\n")

    # 构造一个合法的消息链，包含 user、assistant、tool 和 assistant 各类消息
    valid = [
        {"role": "user", "content": "查天气"},
        {"role": "assistant", "content": None, "tool_calls": [
            {"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": "{}"}},
            {"id": "call_2", "type": "function", "function": {"name": "get_time", "arguments": "{}"}},
        ]},
        {"role": "tool", "tool_call_id": "call_1", "content": "{}"},
        {"role": "tool", "tool_call_id": "call_2", "content": "{}"},
        {"role": "assistant", "content": "完成"},
    ]

    # 构造一个非法（残缺）的消息链，缺少对 call_2 的 tool 响应
    broken = [
        {"role": "user", "content": "查天气"},
        {"role": "assistant", "content": None, "tool_calls": [
            {"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": "{}"}},
            {"id": "call_2", "type": "function", "function": {"name": "get_time", "arguments": "{}"}},
        ]},
        {"role": "tool", "tool_call_id": "call_1", "content": "{}"},
    ]

    # 对合法消息链调用验证函数，收集错误信息
    err_valid = validate_message_chain(valid)
    # 对非法消息链调用验证函数，收集错误信息
    err_broken = validate_message_chain(broken)

    # 打印合法消息链的错误信息，如果为 None，则显示“无”
    print(f"[合法消息链] errors={err_valid or '无'}")
    # 打印非法消息链的错误信息，如果为 None，则显示“无”
    print(f"[残缺消息链] errors={err_broken or '无'}")
    # 如果合法消息链没有错误，且非法消息链有错误，则表示验证器工作正常
    if not err_valid and err_broken:
        print("\n✓ 验证器工作正常")

# 判断是否为主模块
if __name__ == "__main__":
    # 设置 stdout 输出格式
    setup_stdout()
    # 运行主演示函数
    run_demo()

9. 总结 #

Tool Calls 的难点不在「怎么调函数」，而在 messages 协议：顺序、配对、格式任一出错都会 400。上线前对照下面清单过一遍；示例代码从 t2.py（单工具）到 §7–§8（链式/并行/调试）可按需选用。

9.1 核心要点 #

记住这五条，就能覆盖 90% 的 Tool Calls 集成问题：

消息顺序：user → assistant(tool_calls) → tool → assistant(final)
ID匹配：每个 tool_call 的 id 必须与对应的 tool 消息的 tool_call_id 匹配
完整性：所有 tool_calls 都必须有对应的 tool 响应
格式正确：arguments 必须是有效的JSON字符串
不能中断：在工具调用链中不能插入其他类型的消息

9.2 检查清单 #

发 API 前或单测里自动跑 validate_message_chain，人工再核对：

所有 tool_calls 都有对应的 tool 消息
tool_call_id 正确匹配
消息顺序正确
arguments 是有效JSON
content 是字符串格式
没有在 tool 消息之间插入其他角色消息
每个工具调用都有明确的错误处理

9.3 常见陷阱 #

以下七条是仓库里真实踩坑的汇总（含 s13 后台通知顺序错误）：

忘记添加 tool 消息
tool_call_id 不匹配
消息顺序错误
使用占位内容代替真实结果
后台任务未完成就发送结果
JSON格式错误
混合使用新旧API格式

导航菜单

1. 基础概念 #

1.1 什么是 Tool Calls？ #

1.2 关键角色 #

2. 消息格式 #

2.1 Assistant 消息（包含 tool_calls） #

2.2 Tool 消息（响应工具调用） #

2.3 完整的消息序列示例 #

3. 消息顺序规则 #

3.1 核心规则（必须遵守） #

3.2 规则详解 #

3.2.1 规则1：顺序严格匹配 #

3.2.2 规则2：完整的配对 #

3.2.3 规则3：不能插入其他消息 #

4. 完整工作流程 #

4.1 流程图 #

4.2 代码实现 #

5. 常见错误及解决方案 #

5.1 错误1：缺少 tool 消息 #

5.2 错误2：tool_call_id 不匹配 #

5.3 错误3：消息顺序错误 #

5.4 错误4：工具结果格式错误 #

5.5 错误5：混合使用旧版函数调用 #

6. 实战示例 #

6.1 实战1：单个工具调用 #

6.2 实战2：多个工具调用（并行） #

6.3 实战3：处理后台任务 #

7. 高级技巧 #

7.1 工具调用链（Chain of Tools） #

7.2 并行工具执行优化 #

7.3 错误处理和重试 #

7.4 流式处理工具调用 #

7.5 工具响应格式验证 #

8. 调试技巧 #

8.1 打印完整消息链 #

8.1.1 tool_demo_common #

8.1.2 code.py #

8.2 验证消息完整性 #

9. 总结 #

9.1 核心要点 #

9.2 检查清单 #

9.3 常见陷阱 #