使用推理模型管理函数调用
OpenAI 现在提供使用 推理模型 的函数调用功能。推理模型经过训练,能够遵循逻辑思维链,因此更适合复杂或多步骤的任务。
像 o3 和 o4-mini 这样的推理模型是经过强化学习训练以执行推理的 LLM。推理模型在回答之前会进行思考,在响应用户之前会产生一个长长的内部思维链。推理模型在复杂的解决问题、编码、科学推理和代理工作流的多步规划方面表现出色。它们也是 Codex CLI(我们轻量级的编码代理)的最佳模型。
在大多数情况下,通过 API 使用这些模型非常简单,与使用熟悉的“聊天”模型相当。
但是,有一些细微差别需要注意,尤其是在使用函数调用等功能时。
此笔记本中的所有示例都使用较新的 Responses API,它提供了方便的抽象来管理对话状态。但是,这里的原理与使用旧的聊天补全 API 相关。
调用推理模型的 API
# pip install openai
# 导入库
import json
from openai import OpenAI
from uuid import uuid4
from typing import Callable
client = OpenAI()
MODEL_DEFAULTS = {
"model": "o4-mini", # 200,000 令牌上下文窗口
"reasoning": {"effort": "low", "summary": "auto"}, # 自动总结推理过程。也可以选择“详细”或“无”
}
让我们使用 Responses API 对推理模型进行简单调用。
我们指定较低的推理工作量,并通过有用的 output_text
属性检索响应。
我们可以提出后续问题,并使用 previous_response_id
让 OpenAI 自动管理对话历史记录。
response = client.responses.create(
input="Which of the last four Olympic host cities has the highest average temperature?",
**MODEL_DEFAULTS
)
print(response.output_text)
response = client.responses.create(
input="what about the lowest?",
previous_response_id=response.id,
**MODEL_DEFAULTS
)
print(response.output_text)
在最近的四届夏季奥运会主办城市——东京(2020 年)、里约热内卢(2016 年)、伦敦(2012 年)和北京(2008 年)——中,里约热内卢的气候最为温暖。年平均气温大约为:
• 里约热内卢:≈ 23 °C
• 东京:≈ 16 °C
• 北京:≈ 13 °C
• 伦敦:≈ 11 °C
所以里约热内卢的年平均气温最高。
在这四者中,伦敦的年平均气温最低,约为 11 摄氏度。
很简单!
我们提出的问题相当复杂,可能需要模型推断出一个计划并分步执行,但这种推理对我们来说是隐藏的——我们只需稍等片刻,然后就能看到响应。
但是,如果我们检查输出,可以看到模型使用了模型上下文窗口中包含但未暴露给我们的隐藏的“推理”令牌集。 我们可以在响应中看到这些令牌和推理的摘要(但不是使用的实际令牌)。
print(next(rx for rx in response.output if rx.type == 'reasoning').summary[0].text)
response.usage.to_dict()
**确定最低温度**
用户正在询问最近四届奥运会主办城市:东京、里约、伦敦和北京的最低平均气温。我看到伦敦的平均气温最低,约为 11 摄氏度。如果我仔细检查年平均气温:里约约为 23 摄氏度,东京约为 16 摄氏度,北京约为 13 摄氏度。所以,我的最终答案是伦敦,平均气温约为 11 摄氏度。我可以为用户提供这些近似值。
{'input_tokens': 136,
'input_tokens_details': {'cached_tokens': 0},
'output_tokens': 89,
'output_tokens_details': {'reasoning_tokens': 64},
'total_tokens': 225}
了解这些推理令牌很重要,因为这意味着我们的可用上下文窗口消耗速度将比传统聊天模型更快。
调用自定义函数
如果我们要求模型处理一个复杂的请求,该请求还需要使用自定义工具,会发生什么情况?
- 假设我们有更多关于奥运城市的问题,但我们还有一个包含每个城市 ID 的内部数据库。
- 模型可能需要在推理过程的中间调用我们的工具,然后才能返回结果。
- 让我们创建一个生成随机 UUID 的函数,并要求模型对这些 UUID 进行推理。
def get_city_uuid(city: str) -> str:
"""只是一个返回假 UUID 的假工具"""
uuid = str(uuid4())
return f"{city} ID: {uuid}"
# 我们将传递给模型的工具模式
tools = [
{
"type": "function",
"name": "get_city_uuid",
"description": "从内部数据库检索城市的内部 ID。仅当用户需要知道城市的内部 ID 时才调用此函数。",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "要获取信息的城市名称"}
},
"required": ["city"]
}
}
]
# 这是一个通用实践——我们需要一个映射,将我们告诉模型的工具名称与实现它们的函数进行映射。
tool_mapping = {
"get_city_uuid": get_city_uuid
}
# 让我们将其添加到我们的默认设置中,这样我们就不必每次都传递它了
MODEL_DEFAULTS["tools"] = tools
response = client.responses.create(
input="What's the internal ID for the lowest-temperature city?",
previous_response_id=response.id,
**MODEL_DEFAULTS)
print(response.output_text)
这次我们没有收到 output_text
。让我们看看响应输出。
response.output
[ResponseReasoningItem(id='rs_68246219e8288191af051173b1d53b3f0c4fbdb0d4a46f3c', summary=[], type='reasoning', status=None),
ResponseFunctionToolCall(arguments='{"city":"London"}', call_id='call_Mx6pyTjCkSkmASETsVASogoC', name='get_city_uuid', type='function_call', id='fc_6824621b8f6c8191a8095df7230b611e0c4fbdb0d4a46f3c', status='completed')]
除了推理步骤之外,模型还成功识别了工具调用的需求,并传回了发送到我们函数调用的指令。
让我们调用该函数,并将结果发送给模型,以便它能够继续推理。 函数响应是一种特殊类型的消息,因此我们需要将下一条消息构建为一种特殊类型的输入:
{
"type": "function_call_output",
"call_id": function_call.call_id,
"output": tool_output
}
# 从响应中提取函数调用
new_conversation_items = []
function_calls = [rx for rx in response.output if rx.type == 'function_call']
for function_call in function_calls:
target_tool = tool_mapping.get(function_call.name)
if not target_tool:
raise ValueError(f"No tool found for function call: {function_call.name}")
arguments = json.loads(function_call.arguments) # 将参数加载为字典
tool_output = target_tool(**arguments) # 使用参数调用工具
new_conversation_items.append({
"type": "function_call_output",
"call_id": function_call.call_id, # 我们将响应映射回原始函数调用
"output": tool_output
})
response = client.responses.create(
input=new_conversation_items,
previous_response_id=response.id,
**MODEL_DEFAULTS
)
print(response.output_text)
伦敦的内部 ID 是 816bed76-b956-46c4-94ec-51d30b022725。
这在这里效果很好——因为我们知道只需要一个函数调用就可以让模型响应——但我们也需要考虑可能需要执行多个工具调用才能完成推理的情况。
让我们添加第二个调用来运行网络搜索。
OpenAI 的网络搜索工具并非开箱即用,与推理模型一起提供(截至 2025 年 5 月——这可能很快会改变),但使用 4o mini 或其他支持网络搜索的模型创建自定义网络搜索函数并不难。
def web_search(query: str) -> str:
"""搜索网络信息并返回结果摘要"""
result = client.responses.create(
model="gpt-4o-mini",
input=f"Search the web for '{query}' and reply with only the result.",
tools=[{"type": "web_search_preview"}],
)
return result.output_text
tools.append({
"type": "function",
"name": "web_search",
"description": "搜索网络信息并返回结果摘要",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "要搜索网络的信息。"}
},
"required": ["query"]
}
})
tool_mapping["web_search"] = web_search
按顺序执行多个函数
一些 OpenAI 模型支持 parallel_tool_calls
参数,该参数允许模型返回一个函数数组,然后我们可以并行执行这些函数。但是,推理模型可能会产生一系列必须按顺序执行的函数调用,特别是当某些步骤可能依赖于先前步骤的结果时。
因此,我们应该定义一个可以用于处理任意复杂推理工作流的通用模式:
- 在对话的每个步骤,初始化一个循环
- 如果响应包含函数调用,我们必须假设推理正在进行,并且我们应该将函数结果(以及任何中间推理)反馈给模型以进行进一步推理
- 如果没有函数调用,而是收到类型为“message”的
Response.output
,我们可以安全地假设代理已完成推理,然后我们可以跳出循环。
# 让我们将上面的逻辑包装成一个函数,我们可以用它来调用工具。
def invoke_functions_from_response(response,
tool_mapping: dict[str, Callable] = tool_mapping
) -> list[dict]:
"""从响应中提取所有函数调用,查找相应的工具函数并执行它们。
(这是处理异步工具调用或耗时较长的工具调用的好地方。)
这将返回要添加到对话历史记录的消息列表。
"""
intermediate_messages = []
for response_item in response.output:
if response_item.type == 'function_call':
target_tool = tool_mapping.get(response_item.name)
if target_tool:
try:
arguments = json.loads(response_item.arguments)
print(f"Invoking tool: {response_item.name}({arguments})")
tool_output = target_tool(**arguments)
except Exception as e:
msg = f"Error executing function call: {response_item.name}: {e}"
tool_output = msg
print(msg)
else:
msg = f"ERROR - No tool registered for function call: {response_item.name}"
tool_output = msg
print(msg)
intermediate_messages.append({
"type": "function_call_output",
"call_id": response_item.call_id,
"output": tool_output
})
elif response_item.type == 'reasoning':
print(f'Reasoning step: {response_item.summary}')
return intermediate_messages
现在让我们演示一下我们之前讨论过的循环概念。
initial_question = (
"What are the internal IDs for the cities that have hosted the Olympics in the last 20 years, "
"and which of those cities have recent news stories (in 2025) about the Olympics? "
"Use your internal tools to look up the IDs and the web search tool to find the news stories."
)
# 我们获取一个响应,然后启动一个循环来处理响应
response = client.responses.create(
input=initial_question,
**MODEL_DEFAULTS,
)
while True:
function_responses = invoke_functions_from_response(response)
if len(function_responses) == 0: # 我们完成了推理
print(response.output_text)
break
else:
print("More reasoning required, continuing...")
response = client.responses.create(
input=function_responses,
previous_response_id=response.id,
**MODEL_DEFAULTS
)
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Beijing'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'London'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Rio de Janeiro'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Tokyo'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Paris'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Turin'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Vancouver'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Sochi'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Pyeongchang'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': '2025 Beijing Olympics news'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': '2025 London Olympics news'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': '2025 Rio de Janeiro Olympics news'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': '2025 Tokyo Olympics news'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': '2025 Paris Olympics news'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': '2025 Turin Olympics news'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': '2025 Vancouver Olympics news'})
More reasoning required, continuing...
Reasoning step: [Summary(text='**Focusing on Olympic News**\n\nI need to clarify that the Invictus Games are not related to the Olympics, so I should exclude them from my search. That leaves me with Olympic-specific news focusing on Paris. I also want to consider past events, like Sochi and Pyeongchang, so I think it makes sense to search for news related to Sochi as well. Let’s focus on gathering relevant Olympic updates to keep things organized.', type='summary_text')]
Invoking tool: web_search({'query': '2025 Sochi Olympics news'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': '2025 Pyeongchang Olympics news'})
More reasoning required, continuing...
Reasoning step: []
以下是过去 20 年(2005-2025 年)举办过奥运会的城市内部 ID,以及在 2025 年有关于奥运会的新闻报道的城市:
1. 北京(2008 年夏季;2022 年冬季)
• UUID:5b058554-7253-4d9d-a434-5d4ccc87c78b
• 2025 年奥运新闻?2025 年没有重要的奥运专项新闻
2. 伦敦(2012 年夏季)
• UUID:9a67392d-c319-4598-b69a-adc5ffdaaba2
• 2025 年奥运新闻?否
3. 里约热内卢(2016 年夏季)
• UUID:ad5eaaae-b280-4c1d-9360-3a38b0c348c3
• 2025 年奥运新闻?否
4. 东京(2020 年夏季)
• UUID:66c3a62a-840c-417a-8fad-ce87b97bb6a3
• 2025 年奥运新闻?否
5. 巴黎(2024 年夏季)
• UUID:a2da124e-3fad-402b-8ccf-173f63b4ff68
• 2025 年奥运新闻?是
– 奥运圣火气球将于 2028 年每年飘浮在巴黎上空([美联社新闻])
– 国际奥委会将在 2025 年 3 月的会议上更换有缺陷的巴黎 2024 年奖牌([NDTV Sports])
– 国际奥委会选举科斯蒂·考文垂为 2025 年 3 月会议主席([维基百科])
– MLB 取消了计划于 2025 年在巴黎举行的常规赛季比赛([美联社新闻])
6. 都灵(2006 年冬季)
• UUID:3674750b-6b76-49dc-adf4-d4393fa7bcfa
• 2025 年奥运新闻?否(举办了世界特殊奥林匹克冬季运动会,但不是主流奥运会)
7. 温哥华(2010 年冬季)
• UUID:22517787-5915-41c8-b9dd-a19aa2953210
• 2025 年奥运新闻?否
8. 索契(2014 年冬季)
• UUID:f7efa267-c7da-4cdc-a14f-a4844f47b888
• 2025 年奥运新闻?否
9. 平昌(2018 年冬季)
• UUID:ffb19c03-5212-42a9-a527-315d35efc5fc
• 2025 年奥运新闻?否
2025 年奥运相关新闻城市摘要:
• 巴黎(a2da124e-3fad-402b-8ccf-173f63b4ff68)
手动对话编排
到目前为止一切顺利!看到模型暂停执行以运行函数然后再继续,这真的很酷。 实际上,上面的示例非常简单,而生产用例可能要复杂得多:
- 我们的上下文窗口可能会变得太大,我们可能希望修剪掉较旧的、不太相关的消息,或者总结到目前为止的对话
- 我们可能希望允许用户在对话中来回导航并重新生成答案
- 我们可能希望将消息存储在自己的数据库中以供审计,而不是依赖 OpenAI 的存储和编排
- 等等。
在这些情况下,我们可能希望完全控制对话。我们不使用 previous_message_id
,而是可以将 API 视为“无状态”的,并维护一个对话项数组,每次都将其作为输入发送给模型。
这带来了一些推理模型特有的细微差别需要考虑。
- 特别重要的是,我们必须在对话历史记录中保留任何推理和函数调用响应。
- 这是模型跟踪它已经运行过的思维链步骤的方式。如果缺少这些,API 将会报错。
让我们再次回顾上面的示例,手动编排消息并跟踪令牌使用情况。
请注意,下面的代码结构是为了提高可读性——在实践中,您可能希望考虑一种更复杂的流程来处理边缘情况
# 让我们用第一条用户消息初始化我们的对话
total_tokens_used = 0
user_messages = [
(
"Of those cities that have hosted the summer Olympic games in the last 20 years - "
"do any of them have IDs beginning with a number and a temperate climate? "
"Use your available tools to look up the IDs for each city and make sure to search the web to find out about the climate."
),
"Great thanks! We've just updated the IDs - could you please check again?"
]
conversation = []
for message in user_messages:
conversation_item = {
"role": "user",
"type": "message",
"content": message
}
print(f"{'*' * 79}\nUser message: {message}\n{'*' * 79}")
conversation.append(conversation_item)
while True: # Response loop
response = client.responses.create(
input=conversation,
**MODEL_DEFAULTS
)
total_tokens_used += response.usage.total_tokens
reasoning = [rx.to_dict() for rx in response.output if rx.type == 'reasoning']
function_calls = [rx.to_dict() for rx in response.output if rx.type == 'function_call']
messages = [rx.to_dict() for rx in response.output if rx.type == 'message']
if len(reasoning) > 0:
print("More reasoning required, continuing...")
# 确保我们捕获任何推理步骤
conversation.extend(reasoning)
print('\n'.join(s['text'] for r in reasoning for s in r['summary']))
if len(function_calls) > 0:
function_responses = invoke_functions_from_response(response)
# 在函数调用次数过多时保留函数调用和输出的顺序(目前推理模型不支持,但值得考虑)
interleaved = [val for pair in zip(function_calls, function_outputs) for val in pair]
conversation.extend(interleaved)
if len(messages) > 0:
print(response.output_text)
conversation.extend(messages)
if len(function_calls) == 0: # 没有更多函数了 = 推理完成,我们已准备好处理下一条用户消息
break
print(f"Total tokens used: {total_tokens_used} ({total_tokens_used / 200_000:.2%} of o4-mini's context window)")
*******************************************************************************
User message: Of those cities that have hosted the summer Olympic games in the last 20 years - do any of them have IDs beginning with a number and a temperate climate? Use your available tools to look up the IDs for each city and make sure to search the web to find out about the climate.
*******************************************************************************
More reasoning required, continuing...
**Clarifying Olympic Cities**
The user is asking about cities that hosted the Summer Olympics in the last 20 years. The relevant years to consider are 2004 Athens, 2008 Beijing, 2012 London, 2016 Rio de Janeiro, and 2020 Tokyo. If we're considering 2025, then 2004 would actually be 21 years ago, so I should focus instead on the years from 2005 onwards. Therefore, the cities to include are Beijing, London, Rio, and Tokyo. I’ll exclude Paris since it hasn’t hosted yet.
Reasoning step: [Summary(text="**Clarifying Olympic Cities**\n\nThe user is asking about cities that hosted the Summer Olympics in the last 20 years. The relevant years to consider are 2004 Athens, 2008 Beijing, 2012 London, 2016 Rio de Janeiro, and 2020 Tokyo. If we're considering 2025, then 2004 would actually be 21 years ago, so I should focus instead on the years from 2005 onwards. Therefore, the cities to include are Beijing, London, Rio, and Tokyo. I’ll exclude Paris since it hasn’t hosted yet.", type='summary_text')]
Invoking tool: get_city_uuid({'city': 'Beijing'})
Invoking tool: get_city_uuid({'city': 'London'})
Invoking tool: get_city_uuid({'city': 'Rio de Janeiro'})
Invoking tool: get_city_uuid({'city': 'Tokyo'})
More reasoning required, continuing...
Reasoning step: []
Invoking tool: web_search({'query': 'London climate'})
Invoking tool: web_search({'query': 'Tokyo climate'})
More reasoning required, continuing...
我查找了过去 20 年夏季奥运会主办城市的内部 ID 和气候:
• 北京
– ID:937b336d-2708-4ad3-8c2f-85ea32057e1e(以“9”开头)
– 气候:湿润性大陆性气候(冬季寒冷,夏季炎热)→ 非温带
• 伦敦
– ID:ee57f35a-7d1b-4888-8833-4ace308fa004(以“e”开头)
– 气候:温带海洋性气候(温和,降雨适中)
• 里约热内卢
– ID:2a70c45e-a5b4-4e42-8d2b-6c1dbb2aa2d9(以“2”开头)
– 气候:热带(炎热/潮湿)
• 东京
– ID:e5de3686-a7d2-42b8-aca5-6b6e436083ff(以“e”开头)
– 气候:湿润性亚热带气候(夏季炎热潮湿;冬季温和)
只有北京(“9…”)和里约(“2…”)的 ID 以数字开头,但这两个城市都没有温带气候。因此,过去 20 年的举办城市中,没有一个城市同时拥有以数字开头的 ID 和温带气候。
*******************************************************************************
User message: Great thanks! We've just updated the IDs - could you please check again?
*******************************************************************************
More reasoning required, continuing...
Reasoning step: []
Invoking tool: get_city_uuid({'city': 'Beijing'})
Invoking tool: get_city_uuid({'city': 'London'})
Invoking tool: get_city_uuid({'city': 'Rio de Janeiro'})
Invoking tool: get_city_uuid({'city': 'Tokyo'})
这是更新后的 ID 及其气候:
• 北京
– ID:8819a1fd-a958-40e6-8ba7-9f450b40fb13(以“8”开头)
– 气候:湿润性大陆性气候 → 非温带
• 伦敦
– ID:50866ef9-6505-4939-90e7-e8b930815782(以“5”开头)
– 气候:温带海洋性气候
• 里约热内卢
– ID:5bc1b2de-75da-4689-8bff-269e60af32cb(以“5”开头)
– 气候:热带 → 非温带
• 东京
– ID:9d1c920e-e725-423e-b83c-ec7d97f2e79f(以“9”开头)
– 气候:湿润性亚热带气候 → 非温带
在这些城市中,唯一具有温带气候的城市是伦敦,但其 ID 以“5”(数字)开头——因此它满足“ID 以数字开头且气候温带”的条件。
Total tokens used: 17154 (8.58% of o4-mini's context window)
摘要
在此 cookbook 中,我们确定了如何将函数调用与 OpenAI 的推理模型相结合,以演示依赖于外部数据源(包括网络搜索)的多步任务。
重要的是,我们涵盖了函数调用过程中特定于推理模型的细微差别,特别是:
- 模型可以选择按顺序进行多次函数调用或推理步骤,并且某些步骤可能依赖于先前步骤的结果
- 我们无法知道这些步骤会有多少,因此我们必须通过循环来处理响应
- Responses API 使用
previous_response_id
参数简化了编排,但在需要手动控制时,保持对话项的正确顺序以保留“思维链”非常重要。
此处使用的示例相当简单,但您可以想象这项技术如何扩展到更实际的用例,例如:
- 查看客户的交易历史和最近的通信记录,以确定他们是否有资格获得促销优惠
- 调用最近的交易日志、地理位置数据和设备元数据,以评估交易欺诈的可能性
- 查看内部人力资源数据库以获取员工的福利使用情况、任期和最近的政策变更,以回答个性化的人力资源问题
- 阅读内部仪表板、竞争对手新闻源和市场分析,以编译一份根据其重点领域量身定制的每日高管简报。