第3章函数调用¶

学习目标¶

理解函数调用（Function Calling）在 Agent 体系中的核心地位
掌握工具函数的声明方式与 JSON Schema 规范
能够解析模型的工具调用返回并执行真实函数
区分 auto / none / required 三种工具选择策略
理解并行函数调用（Parallel Function Calling）的机制与处理方式

问题导入¶

大语言模型本身只能"输出文本"，无法查数据库、调 API、执行代码。要让 Agent 真正"做事"，必须给模型一种能力：在回答用户前，先声明"我需要调用某个函数获取信息"，由应用代码执行该函数再把结果回填给模型。这就是 Function Calling。

OpenAI 在 Chat Completions 时代就引入了函数调用，并在 Responses API 中将其升级为更统一的"工具（tools）"机制。模型不再凭空编造天气、股价、库存，而是通过结构化 JSON 调用真实服务，从根本上缓解"幻觉"。对 Agent 工程师而言，函数调用是工具使用的基石——MCP、LangGraph 的工具节点、AutoGen 的工具代理，底层都依赖这一机制。本章将带你从声明、调用、解析到并行执行，完整掌握函数调用闭环。

核心概念¶

1. 工具声明与 JSON Schema¶

工具通过 tools 参数声明，每个工具是一个 function 对象，包含 name、description、parameters。其中 parameters 遵循 JSON Schema 规范，描述参数的类型、必填项、枚举值等。模型在训练时学习了如何读懂 Schema，并据此生成符合约束的参数 JSON。

tools = [{
    "type": "function",
    "name": "get_weather",
    "description": "查询某城市当前天气",
    "parameters": {
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "城市名称，如北京"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
        },
        "required": ["city"],
        "additionalProperties": False,
    },
}]

description 与参数描述是模型选对工具的关键，务必写得清晰具体。

2. 工具调用的返回结构¶

当模型决定调用工具时，response.output 中会出现 function_call 类型的 output_item，包含：

call_id：本次调用的唯一标识，回填结果时必须带上。
name：要调用的函数名。
arguments：JSON 字符串形式的参数。

此时 output_text 通常为空。应用代码需根据 name 路由到真实函数，解析 arguments，执行后把结果以 function_call_output 形式回填。

3. 工具选择策略：auto / none / required¶

tool_choice 参数控制模型是否调用工具：

auto（默认）：模型自行判断是否需要调用工具。
none：禁止调用工具，强制直接回答，适合纯闲聊。
required：必须至少调用一个工具，适合强制走工具链的 Agent 流程。
指定函数：{"type": "function", "name": "get_weather"}，强制调用特定函数。

合理设置 tool_choice 能减少不必要的工具调用开销与延迟。

4. 并行函数调用¶

较新的模型支持在一次响应中返回多个 function_call，例如用户问"北京和上海的天气"，模型可能同时返回两个 get_weather 调用。应用代码应：

遍历所有 function_call 项。
并发执行（建议用 asyncio.gather）。
把每个结果按 call_id 回填。

并行调用能显著降低多工具场景的延迟，但要确保函数无副作用冲突。

5. 自动生成 Schema¶

手写 JSON Schema 容易出错。社区常用 pydantic + pydantic_function_tool 或 langchain 的工具装饰器自动生成。Responses API 也兼容直接传入 Python 函数签名（通过 SDK 的工具辅助方法），降低声明成本。

原理图¶

sequenceDiagram
    participant U as 用户
    participant App as 应用代码
    participant API as Responses API
    participant Tool as 真实函数

    U->>App: "北京天气如何？"
    App->>API: responses.create(input, tools=[get_weather])
    API-->>App: function_call(name=get_weather, args={"city":"北京"})
    App->>Tool: get_weather(city="北京")
    Tool-->>App: {"temp":26,"condition":"晴"}
    App->>API: responses.create(input=function_call_output, call_id)
    API-->>App: response.output_text="北京当前 26 度，晴。"
    App-->>U: 回答

图：函数调用的完整闭环——声明、调用、回填、总结

示例代码¶

import os
import json
import asyncio
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


# 1. 定义真实函数
def get_weather(city: str, unit: str = "celsius") -> dict:
    """模拟天气查询"""
    data = {"北京": 26, "上海": 28, "广州": 31}
    temp = data.get(city, 25)
    if unit == "fahrenheit":
        temp = temp * 9 / 5 + 32
    return {"city": city, "temp": temp, "unit": unit, "condition": "晴"}


def get_stock(symbol: str) -> dict:
    """模拟股价查询"""
    return {"symbol": symbol, "price": 152.3, "currency": "USD"}


# 2. 声明工具 Schema
tools = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "查询某城市当前天气",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称，如北京"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
            },
            "required": ["city"],
            "additionalProperties": False,
        },
    },
    {
        "type": "function",
        "name": "get_stock",
        "description": "查询股票当前价格",
        "parameters": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "股票代码，如 AAPL"},
            },
            "required": ["symbol"],
            "additionalProperties": False,
        },
    },
]

# 函数名到真实函数的路由表
FUNCTION_MAP = {
    "get_weather": get_weather,
    "get_stock": get_stock,
}


# 3. 单轮工具调用闭环
def run_with_tools(user_query: str) -> str:
    # 第一次调用：模型决定是否调用工具
    resp = client.responses.create(
        model="gpt-4.1",
        input=user_query,
        tools=tools,
        tool_choice="auto",
    )

    # 收集所有 function_call 并执行
    tool_outputs = []
    for item in resp.output:
        if item.type == "function_call":
            func = FUNCTION_MAP[item.name]
            args = json.loads(item.arguments)
            result = func(**args)
            tool_outputs.append({
                "type": "function_call_output",
                "call_id": item.call_id,
                "output": json.dumps(result, ensure_ascii=False),
            })

    if not tool_outputs:
        return resp.output_text

    # 第二次调用：把工具结果回填，让模型总结
    final = client.responses.create(
        model="gpt-4.1",
        input=resp.output + tool_outputs,  # 注意要带上之前的 output
        tools=tools,
    )
    return final.output_text


# 4. 并行函数调用处理（异步）
async def run_parallel(user_query: str) -> str:
    resp = await asyncio.to_thread(
        client.responses.create,
        model="gpt-4.1",
        input=user_query,
        tools=tools,
    )

    calls = [item for item in resp.output if item.type == "function_call"]
    if not calls:
        return resp.output_text

    # 并发执行所有工具调用
    async def execute(call):
        func = FUNCTION_MAP[call.name]
        args = json.loads(call.arguments)
        # 这里用 to_thread 包装同步函数
        result = await asyncio.to_thread(func, **args)
        return {
            "type": "function_call_output",
            "call_id": call.call_id,
            "output": json.dumps(result, ensure_ascii=False),
        }

    outputs = await asyncio.gather(*[execute(c) for c in calls])

    final = await asyncio.to_thread(
        client.responses.create,
        model="gpt-4.1",
        input=resp.output + outputs,
        tools=tools,
    )
    return final.output_text


if __name__ == "__main__":
    print(run_with_tools("北京今天多少度？"))
    print(asyncio.run(run_parallel("北京和上海现在的天气，以及 AAPL 股价？")))

企业案例¶

案例 1：Agent 工具集成平台¶

某企业内部 Agent 平台把数据库查询、Jira 工单、Confluence 搜索等封装为标准工具，统一用 Responses API 的 tools 暴露给模型。通过 tool_choice="required" 强制走工具链，避免模型直接编造答案，工具调用日志全量审计。

案例 2：智能运维（AIOps）¶

运维 Agent 声明了 restart_service、query_logs、scale_replicas 等工具，模型根据告警描述自动选择调用顺序。对有副作用的工具（如重启），在 Schema 中用 description 标注"需人工确认"，由应用层插入审批环节。

案例 3：电商导购¶

电商导购 Agent 同时调用 search_product、get_coupon、check_inventory 三个工具，并行返回结果后由模型汇总推荐，整体响应时间从串行的 6 秒降至 2 秒。

常见问题¶

Q1：模型生成的参数不符合 Schema 怎么办？ 设置 strict: true 开启结构化输出（部分模型支持），可保证参数严格匹配 Schema；否则应在应用层做校验并提示模型重试。

Q2：函数调用结果回填后模型还在调用工具，正常吗？ 正常。复杂任务可能需要多轮工具调用。应设置最大轮次（如 5 次）防止死循环，超出后强制让模型基于已有信息作答。

Q3：call_id 可以省略吗？ 不可以。call_id 是模型关联"调用"与"结果"的唯一凭证，缺失或错乱会导致模型无法正确引用工具结果。

Q4：如何让模型不调用某些敏感工具？ 动态构建 tools 列表，按用户权限过滤；或在 tool_choice 中明确排除。永远不要依赖模型"自觉"不调用。

Q5：并行调用时结果顺序重要吗？ 不重要。模型按 call_id 匹配结果，与回填顺序无关。但要确保每个 call_id 都有对应输出，否则模型会等待缺失结果。

面试题¶

1. 简述函数调用的完整流程。 答：① 应用声明 tools（含 JSON Schema）；② 模型返回 function_call（含 name、arguments、call_id）；③ 应用执行真实函数；④ 把结果以 function_call_output 形式连同 call_id 回填；⑤ 模型基于结果生成最终回复。

2. tool_choice 的三种模式及适用场景？ 答：auto（默认，模型自主决定）、none（禁用工具，适合闲聊）、required（强制调用，适合 Agent 流程）。还可指定具体函数名强制调用某一工具。

3. 如何处理并行函数调用？ 答：遍历 response.output 中所有 function_call 项，用 asyncio.gather 并发执行，按 call_id 回填所有结果。注意无副作用冲突，并设置超时。

4. 为什么工具的 description 比参数名更重要？ 答：模型主要依据 description 判断"何时调用哪个工具"，描述模糊会导致选错工具或漏调用。参数名影响 JSON 生成，description 影响工具选择决策。

5. 函数调用如何缓解幻觉？ 答：把"事实性信息"（天气、库存、价格）交给真实函数返回，模型只负责理解意图与总结结果，从源头避免模型凭训练数据编造可能过时的事实。

练习题¶

实现一个支持 add / subtract / multiply / divide 四则运算的 Agent，处理除零错误并把错误信息回填给模型。
给定一个本地 SQLite 数据库，封装 query_db(sql) 工具，让模型能回答"销量最高的商品"——注意在 Schema 中约束只读查询。
扩展并行调用示例，加入超时控制：单个工具超过 3 秒则回填 {"error":"timeout"}，让模型决定是否重试。

小结¶

函数调用是 Agent 从"能说"到"能做"的关键一跃。本章覆盖了工具声明（JSON Schema）、返回解析、tool_choice 策略与并行调用处理。核心要点：description 要写得具体；call_id 必须正确回填；并行调用用 asyncio.gather；对有副作用的工具加人工审批。掌握函数调用后，你就具备了构建工具型 Agent 的地基，后续 LangGraph 与 MCP 都建立在此之上。下一章转向 Embedding，进入向量与语义搜索的世界。

第3章 函数调用¶