前言

前面的 OpenAI API 文章演示了基本的对话。这篇讲 API 中最重要也最容易被低估的功能:Function Calling——让模型不仅能”说话”,还能”做事”。

Function Calling 不是让模型执行你的代码,而是让模型决定什么时候该调用哪个函数,以及用什么参数。模型返回函数名和参数,你的代码负责执行,然后把结果返回给模型。


一、基本机制

1
2
3
4
5
6
7
用户: "今天北京天气怎么样?"

1. 你的代码发送请求,同时告诉模型有哪些工具可用
2. 模型分析用户意图,返回: {"function": "get_weather", "arguments": {"city": "北京"}}
3. 你的代码执行 get_weather("北京"),拿到结果 "晴,22°C"
4. 你的代码把结果发给模型
5. 模型生成最终回复: "北京今天晴朗,气温22°C,适合户外活动"

二、定义工具(Tool)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from openai import OpenAI
import json

client = OpenAI(api_key="sk-xxx")

# 工具定义(JSON Schema 格式)
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的实时天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称,如 北京、上海"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "发送邮件",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string", "description": "收件人邮箱"},
"subject": {"type": "string", "description": "邮件主题"},
"body": {"type": "string", "description": "邮件正文"}
},
"required": ["to", "subject", "body"]
}
}
}
]

三、完整的调用流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import json

def get_weather(city: str, unit: str = "celsius") -> str:
"""模拟天气查询"""
weather = {
"北京": {"celsius": "晴,22°C", "fahrenheit": "晴,72°F"},
"上海": {"celsius": "多云,25°C", "fahrenheit": "多云,77°F"},
}
return weather.get(city, {}).get(unit, f"未找到 {city} 的天气")

def send_email(to: str, subject: str, body: str) -> str:
"""模拟发送邮件"""
return f"邮件已发送至 {to}{subject}」"

# 函数名到实际函数的映射
available_functions = {
"get_weather": get_weather,
"send_email": send_email,
}

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

# 第一步:发送请求,获取可能的函数调用
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto" # 让模型自己判断是否需要调用函数
)

response_message = response.choices[0].message
tool_calls = response_message.tool_calls

# 第二步:如果模型要求调用函数,逐个执行
if tool_calls:
messages.append(response_message)

for tool_call in tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)

print(f"[调用] {func_name}({func_args})")

# 执行函数
func = available_functions[func_name]
result = func(**func_args)

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

# 第三步:把函数执行结果发给模型,生成最终回复
final_response = client.chat.completions.create(
model="gpt-4o",
messages=messages
)
return final_response.choices[0].message.content

# 模型没有调用函数,直接返回回答
return response_message.content

# 测试
print(run_conversation("北京今天天气如何?如果气温超过 20 度就提醒我少穿点"))

四、并行函数调用

当用户的一句话触发多个不相关的函数时,模型会并行返回:

1
2
3
4
5
6
7
8
# 用户说了一句话,需要同时查两个城市的天气
user_input = "北京和上海的天气分别怎么样?"

# 模型一次返回两个 tool_calls:
# tool_calls[0]: get_weather("北京")
# tool_calls[1]: get_weather("上海")

# 两个调用可以并行执行,显著减少延迟

确认你的 tools 定义之间没有依赖关系(两个函数的参数不互相依赖),模型就会自动并行调用。


五、强制函数调用和结构化输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 强制模型必须调用某个函数
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "张三 25岁 深圳"}],
tools=[{
"type": "function",
"function": {
"name": "extract_person_info",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string"},
"age": {"type": "integer"},
"city": {"type": "string"}
},
"required": ["name", "age", "city"]
}
}
}],
tool_choice={"type": "function", "function": {"name": "extract_person_info"}}
)

import json
info = json.loads(response.choices[0].message.tool_calls[0].function.arguments)
print(info) # {"name": "张三", "age": 25, "city": "深圳"}

六、实战:智能日程助手

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
import json
from datetime import datetime, timedelta
from openai import OpenAI

client = OpenAI(api_key="sk-xxx")

# 模拟日程数据
schedule_db = []

# 工具定义
tools = [
{
"type": "function",
"function": {
"name": "add_event",
"description": "添加日程事件",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "事件名称"},
"date": {"type": "string", "description": "日期,格式 YYYY-MM-DD"},
"time": {"type": "string", "description": "时间,格式 HH:MM,可选"},
"duration_minutes": {"type": "integer", "description": "持续分钟数"}
},
"required": ["title", "date"]
}
}
},
{
"type": "function",
"function": {
"name": "query_schedule",
"description": "查询指定日期的日程",
"parameters": {
"type": "object",
"properties": {
"date": {"type": "string", "description": "日期,YYYY-MM-DD 格式"}
},
"required": ["date"]
}
}
},
{
"type": "function",
"function": {
"name": "delete_event",
"description": "删除日程",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "要删除的事件名称(支持模糊匹配)"}
},
"required": ["title"]
}
}
}
]

def add_event(title, date, time=None, duration_minutes=60):
event = {"title": title, "date": date, "time": time or "全天",
"duration": duration_minutes}
schedule_db.append(event)
return json.dumps({"status": "ok", "event": event})

def query_schedule(date):
events = [e for e in schedule_db if e["date"] == date]
if not events:
return json.dumps({"date": date, "events": [], "message": "当天没有日程"})
return json.dumps({"date": date, "events": events})

def delete_event(title):
global schedule_db
before = len(schedule_db)
schedule_db = [e for e in schedule_db if title not in e["title"]]
deleted = before - len(schedule_db)
return json.dumps({"status": "ok", "deleted": deleted})

available_functions = {
"add_event": add_event,
"query_schedule": query_schedule,
"delete_event": delete_event,
}

SYSTEM_PROMPT = """你是一个日程管理助手。根据用户描述,调用合适的函数管理日程。
- 如果用户说"明天下午",请计算出具体日期和时间
- 操作完成后简洁地告知用户结果
- 当前日期是 {}""".format(datetime.now().strftime("%Y-%m-%d"))

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

while True:
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message

if not message.tool_calls:
return message.content

messages.append(message)
for tc in message.tool_calls:
func = available_functions[tc.function.name]
args = json.loads(tc.function.arguments)
result = func(**args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})

# 使用
print(chat("下周一上午9点开会,讨论项目进度,预计1.5小时"))
print(chat("下周一下午3点去牙科复诊"))
print(chat("告诉我下周一的日程"))

七、注意事项

1. tool description 是给模型”看”的,不是给机器看的

把 description 写好,就像给同事交代接口一样。模糊的 description 会导致模型选错工具或填错参数。

2. 校验参数,不要完全信任模型

模型可能传一个不存在的城市名,或者把 name 填成 age。在函数内部做类型和值的校验,出错了把错误信息返回给模型,它会自己纠正。

3. 复杂的多层调用

模型在一次 response 中返回多个 tool_calls 时,这些调用之间没有先后顺序保证。如果需要 A 的结果才能决定 B 的参数,需要分两次请求。

Function Calling 和 Agent 之间的界限很模糊。当你把 Function Calling、多轮对话、错误重试打包起来,本质上就是一个 Agent 了。