使用推理模型处理函数调用:完整指南

本文面向有编程基础的开发者,深入讲解如何将OpenAI推理模型(如o3、o4-mini)与函数调用结合,构建多步骤智能体工作流。文章从基础API调用出发,逐步演示自定义函数、网络搜索工具集成,并重点剖析推理模型特有的循环处理模式与手动对话编排技巧。通过完整代码示例,读者将掌握处理连续函数调用、保留思维链、优化上下文窗口等关键实践,为生产级AI应用开发打下坚实基础。

技术背景

传统聊天模型(如GPT-4o)在复杂任务中往往需要多次提示才能完成,且缺乏显式推理能力。推理模型通过强化学习训练,在回答前生成内部思维链,能更好地规划多步骤任务,尤其在函数调用场景中,模型可自主决定何时调用工具、按何种顺序调用,从而构建更智能的智能体。然而,推理模型的思维链消耗额外token,且函数调用可能连续多次,需要开发者设计循环处理机制。本文旨在帮助开发者掌握这些高级用法,避免常见陷阱。

核心原理

推理模型(如o4-mini)在生成最终回答前,会先产生一个隐藏的思维链(reasoning tokens),用于规划任务步骤。当需要外部数据时,模型会输出function_call类型的响应项,包含函数名和参数。开发者需执行对应函数,并将结果以function_call_output消息返回给模型,模型继续推理直至生成最终message。关键点:推理模型可能产生多个顺序依赖的函数调用,因此必须循环处理,直到响应中不再包含function_call。此外,思维链token计入上下文窗口,需注意管理。

架构分析

系统架构采用客户端-服务器模式:客户端(Python脚本)通过OpenAI Responses API与推理模型交互。核心组件包括:1) 模型配置(MODEL_DEFAULTS)指定模型、推理强度、工具列表;2) 工具定义(tools)和工具映射(tool_mapping)将函数名映射到实际Python函数;3) 响应处理循环,检查响应中的function_call并执行,将结果反馈给模型;4) 对话编排,使用previous_response_id自动管理历史,或手动维护conversation列表。设计上强调灵活性和可扩展性,支持并行工具调用(若模型支持)和异步执行。

完整实战教程

使用推理模型管理函数调用

OpenAI 现在支持使用推理模型进行函数调用。推理模型经过训练,能够遵循逻辑思维链,因此更适合处理复杂或多步骤任务。

_像 o3 和 o4-mini 这样的推理模型是通过强化学习训练的 LLM,能够进行推理。推理模型在回答之前会进行思考,产生一个较长的内部思维链,然后才回应用户。推理模型在复杂问题解决、编程、科学推理以及智能体工作流的多步骤规划方面表现出色。它们也是我们轻量级编码代理 Codex CLI 的最佳模型。_

在大多数情况下,通过 API 使用这些模型非常简单,与使用熟悉的“聊天”模型类似。

然而,有一些细微差别需要牢记,尤其是在使用函数调用等功能时。

本笔记本中的所有示例都使用较新的 Responses API,它提供了方便的状态管理抽象。但这里的原则也适用于较旧的聊天补全 API。

向推理模型发起 API 调用

# pip install openai
# Import libraries 
import json
from openai import OpenAI
from uuid import uuid4
from typing import Callable

client = OpenAI()
MODEL_DEFAULTS = {
    "model": "o4-mini", # 200,000 token context window
    "reasoning": {"effort": "low", "summary": "auto"}, # Automatically summarise the reasoning process. Can also choose "detailed" or "none"
}

让我们使用 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)

简单又轻松!

我们提出的问题相对复杂,可能需要模型推理出一个计划并逐步执行,但这种推理对我们来说是隐藏的——我们只是多等了一会儿才看到回复。

然而,如果我们检查输出,可以看到模型使用了隐藏的“推理”标记,这些标记包含在模型的上下文窗口中,但并未作为最终用户暴露给我们。

我们可以在响应中看到这些标记和推理摘要(但不是实际使用的字面标记)。

print(next(rx for rx in response.output if rx.type == 'reasoning').summary[0].text)
response.usage.to_dict()

了解这些推理标记非常重要,因为这意味着与传统的聊天模型相比,我们会更快地消耗可用的上下文窗口。

调用自定义函数

如果我们向模型提出一个复杂的请求,同时还需要使用自定义工具,会发生什么?

* 假设我们还有关于奥运城市的问题,但我们有一个内部数据库,其中包含每个城市的ID。

* 模型可能需要在返回结果之前,在推理过程中调用我们的工具。

* 让我们创建一个生成随机UUID的函数,并让模型对这些UUID进行推理。


def get_city_uuid(city: str) -> str:
    """Just a fake tool to return a fake UUID"""
    uuid = str(uuid4())
    return f"{city} ID: {uuid}"

# The tool schema that we will pass to the model
tools = [
    {
        "type": "function",
        "name": "get_city_uuid",
        "description": "Retrieve the internal ID for a city from the internal database. Only invoke this function if the user needs to know the internal ID for a city.",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "The name of the city to get information about"}
            },
            "required": ["city"]
        }
    }
]

# This is a general practice - we need a mapping of the tool names we tell the model about, and the functions that implement them.
tool_mapping = {
    "get_city_uuid": get_city_uuid
}

# Let's add this to our defaults so we don't have to pass it every time
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

在推理步骤的同时,模型成功识别了进行工具调用的需求,并传回了发送给我们函数调用的指令。

让我们调用该函数并将结果发送给模型,以便它能够继续推理。

函数响应是一种特殊类型的消息,因此我们需要将下一条消息构造为一种特殊类型的输入:

{
"type": "function_call_output",
"call_id": function_call.call_id,
"output": tool_output
}
# Extract the function call(s) from the response
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) # Load the arguments as a dictionary
    tool_output = target_tool(**arguments) # Invoke the tool with the arguments
    new_conversation_items.append({
        "type": "function_call_output",
        "call_id": function_call.call_id, # We map the response back to the original function call
        "output": tool_output
    })
response = client.responses.create(
    input=new_conversation_items,
    previous_response_id=response.id,
    **MODEL_DEFAULTS
)
print(response.output_text)

在这里,这种方法效果很好——因为我们知道模型只需一次函数调用即可响应——但我们也需要考虑可能需要执行多次工具调用才能完成推理的情况。

让我们添加第二次调用来运行网络搜索。

OpenAI 的网络搜索工具在推理模型中并非开箱即用(截至 2025 年 5 月,这种情况可能很快就会改变),但使用 4o mini 或其他支持网络搜索的模型创建自定义网络搜索函数并不难。

def web_search(query: str) -> str:
    """Search the web for information and return back a summary of the results"""
    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": "Search the web for information and return back a summary of the results",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "The query to search the web for."}
            },
            "required": ["query"]
        }
    })
tool_mapping["web_search"] = web_search

按顺序执行多个函数

某些 OpenAI 模型支持 parallel_tool_calls 参数,该参数允许模型返回一个函数数组,我们可以并行执行这些函数。然而,推理模型可能会产生一系列必须按顺序执行的函数调用,尤其是当某些步骤可能依赖于先前步骤的结果时。

因此,我们应当定义一个通用模式,用于处理任意复杂的推理工作流:

* 在对话的每一步,初始化一个循环

* 如果响应包含函数调用,我们必须假设推理仍在进行,并应将函数结果(以及任何中间推理)反馈给模型以进行进一步推理

* 如果没有函数调用,而是收到一个类型为 'message' 的 response.output,我们可以安全地假设代理已完成推理,并可以跳出循环

# Let's wrap our logic above into a function which we can use to invoke tool calls.
def invoke_functions_from_response(response,
                                   tool_mapping: dict[str, Callable] = tool_mapping
                                   ) -> list[dict]:
    """Extract all function calls from the response, look up the corresponding tool function(s) and execute them.
    (This would be a good place to handle asynchroneous tool calls, or ones that take a while to execute.)
    This returns a list of messages to be added to the conversation history.
    """
    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."
)

# We fetch a response and then kick off a loop to handle the response
response = client.responses.create(
    input=initial_question,
    **MODEL_DEFAULTS,
)
while True:   
    function_responses = invoke_functions_from_response(response)
    if len(function_responses) == 0: # We're done reasoning
        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
        )

手动对话编排

到目前为止一切顺利!看着模型暂停执行以运行函数然后再继续,真的很酷。

实际上,上面的例子相当简单,生产环境中的用例可能复杂得多:

* 我们的上下文窗口可能变得太大,我们可能希望修剪较旧且不太相关的消息,或总结到目前为止的对话

* 我们可能希望允许用户在对话中来回导航并重新生成答案

* 我们可能希望将消息存储在我们自己的数据库中用于审计目的,而不是依赖OpenAI的存储和编排

* 等等。

在这些情况下,我们可能希望完全控制对话。与其使用previous_message_id,我们可以将API视为“无状态”的,并维护一个对话项数组,每次将其作为输入发送给模型。

这带来了一些特定于推理模型的细微差别需要考虑。

* 特别是,我们必须保留对话历史中的任何推理和函数调用响应。

* 这是模型跟踪其已执行的思维链步骤的方式。如果这些未包含,API将报错。

让我们再次运行上面的例子,自己编排消息并跟踪令牌使用情况。

# Let's initialise our conversation with the first user message
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...")
            # Ensure we capture any reasoning steps
            conversation.extend(reasoning)
            print('\n'.join(s['text'] for r in reasoning for s in r['summary']))
        if len(function_calls) > 0:
            function_outputs = invoke_functions_from_response(response)
            # Preserve order of function calls and outputs in case of multiple function calls (currently not supported by reasoning models, but worth considering)
            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:  # No more functions = We're done reasoning and we're ready for the next user message
            break
print(f"Total tokens used: {total_tokens_used} ({total_tokens_used / 200_000:.2%} of o4-mini's context window)")

总结

在本教程中,我们介绍了如何将函数调用与OpenAI的推理模型结合,以演示依赖于外部数据源(包括搜索网络)的多步骤任务。

重要的是,我们涵盖了函数调用过程中推理模型特有的细微差别,具体包括:

* 模型可能会连续进行多次函数调用或推理步骤,某些步骤可能依赖于先前步骤的结果

* 我们无法预知这些步骤的数量,因此必须使用循环来处理响应

* Responses API通过previous_response_id参数简化了编排,但在需要手动控制的情况下,保持对话项的正确顺序以保留“思维链”至关重要

这里使用的示例相对简单,但您可以想象如何将此技术扩展到更实际的应用场景,例如:

* 查询客户的交易历史和近期通信,以确定他们是否符合促销优惠的资格

* 调用近期交易日志、地理位置数据和设备元数据,以评估交易欺诈的可能性

* 查看内部人力资源数据库,获取员工的福利使用情况、任职年限和最新政策变更,以回答个性化的人力资源问题

* 阅读内部仪表板、竞争对手新闻源和市场分析,以编制针对高管关注领域的每日简报

常见问题

常见错误:1) 未处理连续函数调用,导致只执行一次就停止;2) 未保留推理步骤,手动编排时遗漏reasoning项导致API报错;3) 函数参数解析失败,需确保json.loads正确;4) 工具未注册,需检查tool_mapping。排查方法:打印响应output类型,确认是否包含function_call;检查usage中的token消耗;使用try-except捕获函数执行异常。生产优化:使用异步工具调用、缓存结果、限制推理强度(effort)以节省token、定期总结对话历史以控制上下文长度。

企业应用扩展

本文示例可扩展至实际场景:1) 客户资格评估:查询交易历史和通信记录,判断促销资格;2) 欺诈检测:结合交易日志、地理位置和设备元数据评估风险;3) 个性化HR问答:查询员工福利、任职年限和政策变更;4) 高管简报:整合内部仪表板、竞品新闻和市场分析。此外,可探索并行工具调用(若模型支持)、流式输出、以及将推理摘要用于审计。

总结

本文面向有编程基础的开发者,深入讲解如何将OpenAI推理模型(如o3、o4-mini)与函数调用结合,构建多步骤智能体工作流。文章从基础API调用出发,逐步演示自定义函数、网络搜索工具集成,并重点剖析推理模型特有的循环处理模式与手动对话编排技巧。通过完整代码示例,读者将掌握处理连续函数调用、保留思维链、优化上下文窗口等关键实践,为生产级AI应用开发打下坚实基础。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员,解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部API 与工程实践教程