引言
让我们跟踪一个真实的请求:用户问「北京今天天气怎么样?适合出门吗?」看看从提问到最终回答,到底发生了什么。
你可能以为只是一次 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_choice | auto/none/required/指定工具——控制模型的调用行为 |
| 错误处理 | 把错误信息作为工具结果返回,模型会据此调整回答 |
思考题
- 为什么 Tool Call 需要两次模型调用(一次决定调用什么,一次基于结果生成回答)?如果只调用一次会怎样?
- 如果工具执行时间很长(比如搜索一个大型数据库要 30 秒),这个流程有什么问题?你会怎么优化?
- 设计一个完整的工具集,让 AI 能帮你管理工作日程。需要哪些工具?每个工具的参数是什么?