Skip to content

引言

让我们跟踪一个真实的请求:用户问「北京今天天气怎么样?适合出门吗?」看看从提问到最终回答,到底发生了什么。

你可能以为只是一次 API 调用。实际上,这是一个至少 6 步的循环过程,涉及两次模型调用和一次外部 API 调用。理解这个流程,是构建任何 AI Agent 的基础。


完整的六步流程

用户提问:"北京今天天气怎么样?适合出门吗?"

  Step 1: 发送消息 + 工具定义给模型
    ──────────────────────────────────────→

  Step 2: 模型决定调用工具,生成参数
    ←──────────────────────────────────────
    {"name": "get_weather", "arguments": {"city": "北京"}}

  Step 3: 你的代码执行工具,获取结果
    → 调用天气 API → {"temp": 28, "condition": "晴", "aqi": 42}

  Step 4: 把工具结果追加到消息,再次调用模型
    ──────────────────────────────────────→

  Step 5: 模型基于工具结果生成最终回答
    ←──────────────────────────────────────
    "北京今天晴天,28°C,空气质量优,很适合出门!"

  Step 6: 返回最终回答给用户

完整代码实现

python
import json
from openai import OpenAI

client = OpenAI()

# === 工具定义 ===
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "获取指定城市的当前天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称,如「北京」"}
            },
            "required": ["city"]
        }
    }
}]

# === 工具实现 ===
def get_weather(city: str) -> dict:
    """实际调用天气 API(这里用模拟数据)"""
    # 真实场景中,这里会调用和风天气、心知天气等 API
    weather_db = {
        "北京": {"temp": 28, "condition": "晴", "aqi": 42, "humidity": 35},
        "上海": {"temp": 32, "condition": "多云", "aqi": 65, "humidity": 72},
    }
    return weather_db.get(city, {"temp": 20, "condition": "未知", "aqi": 50})

# === 工具调度器 ===
def execute_tool(name: str, arguments: str) -> str:
    """根据模型输出的工具名和参数,执行对应函数"""
    args = json.loads(arguments)

    if name == "get_weather":
        result = get_weather(args["city"])
    else:
        result = {"error": f"未知工具: {name}"}

    return json.dumps(result, ensure_ascii=False)

# === 主循环 ===
def chat_with_tools(user_message: str) -> str:
    messages = [
        {"role": "system", "content": "你是一个有帮助的助手。利用提供的工具来回答问题。"},
        {"role": "user", "content": user_message}
    ]

    # Step 1 & 2: 发送消息,检查模型是否要调用工具
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools
    )

    message = response.choices[0].message

    # 如果模型要调用工具
    if message.tool_calls:
        # 把模型的工具调用消息加入历史
        messages.append(message)

        # Step 3: 执行每个工具调用
        for tool_call in message.tool_calls:
            print(f"[工具调用] {tool_call.function.name}({tool_call.function.arguments})")

            result = execute_tool(tool_call.function.name, tool_call.function.arguments)
            print(f"[工具结果] {result}")

            # Step 4: 把工具结果加入消息历史
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # Step 5: 再次调用模型,让它基于工具结果生成回答
        final_response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools
        )

        return final_response.choices[0].message.content

    # 如果模型不需要调用工具,直接返回
    return message.content

# === 运行 ===
answer = chat_with_tools("北京今天天气怎么样?适合出门吗?")
print(f"\n最终回答:{answer}")

运行日志

[工具调用] get_weather({"city": "北京"})
[工具结果] {"temp": 28, "condition": "晴", "aqi": 42, "humidity": 35}

最终回答:北京今天天气很好!晴天,气温 28°C,空气质量指数 42(优),
湿度 35% 也很舒适。非常适合出门活动,记得做好防晒!

多工具并行调用

定义多个工具

python
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "获取股票当前价格",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "股票代码"}
                },
                "required": ["symbol"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索互联网获取信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        }
    }
]

# 用户问:北京天气如何?顺便帮我查一下特斯拉的股价
# 模型可能同时调用 get_weather 和 get_stock_price(并行调用)

并行调用的消息结构

messages 数组在一次并行调用后的状态:

  [
    {"role": "system", "content": "..."},
    {"role": "user", "content": "北京天气如何?特斯拉股价多少?"},

    ← 模型返回多个 tool_calls
    {
      "role": "assistant",
      "tool_calls": [
        {"id": "call_1", "function": {"name": "get_weather", "arguments": "{...}"}},
        {"id": "call_2", "function": {"name": "get_stock_price", "arguments": "{...}"}}
      ]
    },

    ← 每个工具调用结果分别返回
    {"role": "tool", "tool_call_id": "call_1", "content": "{...}"},
    {"role": "tool", "tool_call_id": "call_2", "content": "{...}"},
  ]

Tool Choice 控制

python
# 模式 1: auto(默认)——模型自己决定是否调用工具
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto"    # 默认值
)

# 模式 2: none——禁止调用任何工具
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="none"    # 即使定义了工具也不用
)

# 模式 3: required——必须调用某个工具
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="required"   # 模型必须调用至少一个工具
)

# 模式 4: 指定具体工具
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "get_weather"}}
    # 强制只调用 get_weather
)

错误处理

python
def execute_tool_safely(name: str, arguments: str) -> str:
    """带错误处理的工具执行"""
    try:
        args = json.loads(arguments)

        if name == "get_weather":
            city = args.get("city")
            if not city:
                return json.dumps({"error": "缺少 city 参数"})
            result = get_weather(city)
            return json.dumps(result, ensure_ascii=False)

        return json.dumps({"error": f"未知工具: {name}"})

    except json.JSONDecodeError:
        return json.dumps({"error": "参数格式错误"})
    except Exception as e:
        return json.dumps({"error": f"执行失败: {str(e)}"})
    # 模型会看到错误信息,并据此调整回答
    # 例如:"抱歉,天气信息暂时无法获取。"

本节小结

概念要点
六步流程定义工具 → 模型决策 → 生成参数 → 执行 → 回传 → 生成回答
工具调度器用 if/else 或注册表映射工具名到函数
并行调用模型可同时调用多个工具,每个结果分别回传
tool_choiceauto/none/required/指定工具——控制模型的调用行为
错误处理把错误信息作为工具结果返回,模型会据此调整回答

思考题

  1. 为什么 Tool Call 需要两次模型调用(一次决定调用什么,一次基于结果生成回答)?如果只调用一次会怎样?
  2. 如果工具执行时间很长(比如搜索一个大型数据库要 30 秒),这个流程有什么问题?你会怎么优化?
  3. 设计一个完整的工具集,让 AI 能帮你管理工作日程。需要哪些工具?每个工具的参数是什么?