第3章工具调用和执行¶

学习目标¶

掌握用 Python 函数定义 Tool 的方法
理解工具 schema 自动生成原理
学会异步执行工具与结果回传
掌握工具错误处理与重试策略
能够实现并行工具调用以提升效率

问题导入¶

某团队给 Agent 接了 5 个工具：查订单、查库存、查物流、发短信、建工单。上线后发现：①模型经常传错参数（把订单号当成了手机号）；②查订单和查库存本可同时进行，Agent 却串行调用，响应慢；③工具偶发超时，Agent 直接崩掉。

这三个问题的根因是工具层的工程化不足。工具是 Agent 的"手脚"——定义不清则模型乱用，执行不当则性能与稳定性都差。本章用 OpenAI Agents SDK 讲解如何规范地定义工具、自动生成 schema、异步执行、并行调用，以及如何优雅地处理工具错误。

核心概念¶

1. 用 Python 函数定义 Tool¶

OpenAI Agents SDK 提供 @function_tool 装饰器，把普通 Python 函数一键转为 Agent 可用的工具。函数的类型注解和 docstring 直接成为工具的 schema，无需手写 JSON Schema。

from agents import function_tool

@function_tool
def query_order(order_id: str) -> str:
    """查询订单状态。order_id：订单号，如 A1001"""
    return f"订单 {order_id} 状态：已发货"

要点：必须有完整类型注解；docstring 描述工具用途与参数；返回值应为字符串或可序列化对象。

2. 工具 Schema 自动生成¶

@function_tool 会读取函数的：

函数名 → 工具名
docstring → 工具描述（模型据此判断何时调用）
参数类型注解 → 参数 schema（type、required）
参数默认值 → optional 参数

这避免了手写 JSON Schema 的繁琐与错误，"函数即文档即 schema"是现代 Agent 框架的共识。

3. 异步执行与结果回传¶

工具执行可能耗时（HTTP 请求、数据库查询），应使用异步函数避免阻塞：

@function_tool
async def query_order(order_id: str) -> str:
    async with httpx.AsyncClient() as c:
        r = await c.get(f"{API}/orders/{order_id}")
    return r.text

SDK 的 Runner 是异步的，会自动 await 异步工具。工具返回值作为 observation 回传给 LLM，进入下一轮决策。

4. 工具错误处理¶

工具失败的三类与对策：

参数错误：模型传错参数。把错误信息回传，让模型修正后重试。
执行错误：工具内部异常（网络、权限）。捕获异常，返回结构化错误信息。
业务错误：如订单不存在。直接回传，让模型据此回答用户。

关键原则：永远不要让异常直接抛到 Agent 外层，而是转为可读的字符串回传，让模型自行决策。

5. 并行工具调用¶

现代 LLM（GPT-4o 等）支持单轮内并行发起多个工具调用。例如用户问"查我的订单 A1 和 A2 的状态"，模型可一次性返回两个 tool_call，SDK 会并发执行，再把两个结果一起回传。这显著降低了串行等待延迟。

触发并行的条件：①模型判断多个调用无依赖；②Prompt 鼓励并行；③框架支持并发执行（Agents SDK 默认支持）。

原理图¶

flowchart LR
    User[用户输入] --> LLM[LLM 决策]
    LLM --> |tool_call 1| T1[工具 A]
    LLM --> |tool_call 2| T2[工具 B]
    LLM --> |tool_call 3| T3[工具 C]

    T1 --> R1[结果1]
    T2 --> R2[结果2]
    T3 --> R3[结果3]

    R1 --> Merge[结果合并回传]
    R2 --> Merge
    R3 --> Merge
    Merge --> LLM2[LLM 下一轮决策]
    LLM2 --> Final[最终回答]

    subgraph 错误处理
        E1[参数错误] --> RT1[回传错误让模型修正]
        E2[执行异常] --> RT2[捕获并返回错误信息]
        E3[业务错误] --> RT3[回传让模型回答]
    end

    style T1 fill:#bbdefb
    style T2 fill:#bbdefb
    style T3 fill:#bbdefb
    style Merge fill:#c8e6c9

图：工具调用流程——LLM 发起多个并行 tool_call，框架并发执行后合并结果回传，错误统一转字符串回传

示例代码¶

1. 工具定义与 schema 自动生成¶

"""
第3章示例：定义工具函数并让 Agent 调用
依赖：pip install openai-agents
"""
import asyncio
from agents import Agent, Runner, function_tool


@function_tool
def query_order(order_id: str) -> str:
    """查询订单状态。

    Args:
        order_id: 订单号，格式如 A1001
    """
    db = {"A1001": "已发货", "A1002": "处理中", "A1003": "已完成"}
    return f"订单 {order_id} 状态：{db.get(order_id, '不存在')}"


@function_tool
def query_inventory(sku: str) -> str:
    """查询商品库存。

    Args:
        sku: 商品 SKU 编码
    """
    db = {"SKU001": 100, "SKU002": 0, "SKU003": 30}
    stock = db.get(sku, 0)
    return f"商品 {sku} 库存：{stock} 件" + ("（缺货）" if stock == 0 else "")


# 查看自动生成的 schema
print("=== query_order schema ===")
print(query_order.to_json_schema())


agent = Agent(
    name="电商助手",
    instructions="你是电商客服。可查询订单和库存。回答要简洁。",
    tools=[query_order, query_inventory],
    model="gpt-4o-mini",
)


async def main():
    result = await Runner.run(agent, "帮我查下订单 A1001 和 A1003 的状态")
    print("\n回复：", result.final_output)


if __name__ == "__main__":
    asyncio.run(main())

运行结果示例：

=== query_order schema ===
{'name': 'query_order', 'description': '查询订单状态...', 'parameters': {...}

回复： 订单 A1001 状态：已发货；订单 A1003 状态：已完成。

2. 异步工具与并行调用¶

"""
异步工具 + 并行调用：模拟多个无依赖查询并发执行
"""
import asyncio
import time
from agents import Agent, Runner, function_tool


@function_tool
async def query_order(order_id: str) -> str:
    """查询订单状态（模拟网络耗时）"""
    await asyncio.sleep(1)  # 模拟 IO
    return f"订单 {order_id}：已发货"


@function_tool
async def query_inventory(sku: str) -> str:
    """查询库存（模拟网络耗时）"""
    await asyncio.sleep(1)
    return f"商品 {sku}：库存 50 件"


agent = Agent(
    name="并行助手",
    instructions="遇到多个独立查询时，一次性发起所有工具调用，不要串行。",
    tools=[query_order, query_inventory],
    model="gpt-4o-mini",
)


async def main():
    start = time.time()
    result = await Runner.run(
        agent, "查订单 A1001 状态，再查商品 SKU001 库存"
    )
    elapsed = time.time() - start
    print("回复：", result.final_output)
    print(f"总耗时：{elapsed:.2f}s（并行应约 1s，串行约 2s）")


if __name__ == "__main__":
    asyncio.run(main())

3. 工具错误处理¶

"""
工具错误处理：捕获异常，转字符串回传，让模型自行决策
"""
import asyncio
from agents import Agent, Runner, function_tool


@function_tool
def risky_api(endpoint: str) -> str:
    """调用不稳定的 API。

    Args:
        endpoint: 接口路径
    """
    import random
    if random.random() < 0.6:
        raise RuntimeError(f"{endpoint} 调用超时")
    return f"{endpoint} 返回：OK"


# 包装工具，统一捕获异常
def safe_tool(fn):
    @function_tool(name=fn.__name__, description=fn.__doc__)
    def wrapper(*args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except Exception as e:
            return f"[工具错误]{fn.__name__} 执行失败：{e}，请重试或换方案"
    return wrapper


safe_risky = safe_tool(risky_api)

agent = Agent(
    name="容错助手",
    instructions=(
        "你调用工具时可能失败。若工具返回 [工具错误]，"
        "分析原因后重试（最多 2 次），仍失败则向用户说明并给兜底方案。"
    ),
    tools=[safe_risky],
    model="gpt-4o-mini",
)


async def main():
    result = await Runner.run(agent, "请调用 risky_api 查询 /users 数据", max_turns=8)
    print("回复：", result.final_output)


if __name__ == "__main__":
    asyncio.run(main())

企业案例¶

案例 1：多工具电商客服 Agent¶

某电商客服 Agent 接入 12 个工具：订单查询、物流查询、库存查询、退款、换货、优惠券查询、发票开具、地址修改、投诉建单、FAQ 检索、人工转接、满意度回访。团队用 @function_tool 统一定义，靠 docstring 区分用途。关键经验：①工具命名规范（动词_名词），降低模型选错概率；②每个工具的 docstring 写明"何时该用"和"参数格式"；③副作用工具（退款、建单）加二次确认；④并行调用让多查询场景响应从 6s 降到 1.5s。

案例 2：工具错误降级¶

某金融 Agent 调用行情接口偶发超时。团队设计三级降级：①首次失败指数退避重试（1s/2s）；②二次失败切换备用数据源；③备用源也失败则返回缓存数据并标注"数据可能延迟"。错误信息结构化回传给模型，由模型决定是否告知用户。这套机制让工具层可用性从 95% 提升到 99.9%。

常见问题¶

Q1：工具函数返回复杂对象（dict/list）怎么办？

A：SDK 会自动 JSON 序列化。但建议返回精简字符串，避免长 JSON 撑爆上下文。复杂对象先做摘要再回传。

Q2：如何让模型更准确地选对工具？

A：①docstring 写清楚"何时用"而非只是"做什么"；②工具命名语义化；③工具数量控制在 10 个以内，过多用分组或子 Agent 路由；④在 system prompt 中给出工具选择示例。

Q3：并行工具调用一定更快吗？

A：不一定。无依赖时并行更快；有依赖（后一个需要前一个的结果）必须串行。模型有时会误判依赖，可在 prompt 中明确"以下查询相互独立可并行"。

Q4：副作用工具（如发邮件、退款）如何安全调用？

A：①在工具内加幂等性校验（同一请求只执行一次）；②高风险操作要求二次确认（模型先问用户"确认退款？"再调）；③记录审计日志；④生产环境加白名单和限流。

Q5：工具 schema 自动生成有什么坑？

A：①无类型注解的参数会被忽略；②Optional 参数需显式默认值；③docstring 格式不规范会导致描述缺失；④复杂嵌套类型（如自定义类）不被支持，应拆为基本类型。

面试题¶

简答题：@function_tool 是如何从 Python 函数生成工具 schema 的？

参考要点：读取函数名、docstring、参数类型注解、默认值，自动生成 name/description/parameters。

设计题：一个 Agent 有 20 个工具，如何组织以降低模型选错率？

参考要点：按业务域分组、用子 Agent 路由、工具数量控制在单 Agent 10 个内、命名规范化。

实现题：如何让两个无依赖的工具并行执行？请写出关键代码。

参考要点：用 async 异步工具 + 模型并行 tool_call + asyncio.gather；或依赖 SDK 自动并发。

原理题：为什么工具异常不应直接抛出，而要转字符串回传？

参考要点：抛出会终止整个 Agent；回传让模型决策（重试/换方案/告知用户），保持循环连续性。

对比题：手写 JSON Schema 与自动 schema 生成各有什么优劣？

练习题¶

工具定义：为"日程管理 Agent"定义 3 个工具（创建日程、查询日程、删除日程），观察自动生成的 schema。
并行优化：写一个 Agent 同时查询 3 个城市的天气，验证并行调用是否将总耗时从 3s 降到约 1s。
错误降级：为示例 3 增加备用数据源，主工具失败时自动切换，统计主备成功率。

小结¶

工具是 Agent 干预真实世界的接口。本章核心：用 @function_tool 把 Python 函数声明为工具，schema 自动生成；异步工具避免阻塞，并行调用降低延迟；错误统一转字符串回传，让模型自主决策重试或降级；副作用工具需幂等与确认。掌握工具后，Agent 有了"手脚"，下一章解决 Agent 的"记忆"——如何在多轮交互中保持状态。

第3章 工具调用和执行¶