跳到主要内容

02 · 工具与工具集

本章讲两件事：① 一个普通 Python 函数怎么自动变成「模型能调的工具」；② 多个工具来源怎么组合、调用时怎么路由到对的地方。

2.1 核心魔法：类型注解 → 工具 schema​

它要解决的小问题： 模型要调工具，得先知道「这工具叫什么、要哪些参数、参数什么类型」——也就是一份 JSON schema。手写 schema 又烦又容易和真实函数签名不同步。

思路： 既然 Pydantic 本来就能把类型注解变成 schema、又能拿 schema 校验数据，那就直接拿函数签名当 schema 源。这是 Pydantic AI 整个「FastAPI 手感」的来源。

function_schema（_function_schema.py:103）就是这个翻译器。它做的事：

  • 用 inspect.signature + get_type_hints 读出每个参数的名字和类型（_function_schema.py:131,138）。
  • 第一个参数如果标注成 RunContext[...]，自动识别为「带上下文工具」并跳过它（不进 schema）——_function_schema.py:150-165。
  • 其余参数逐个用 Pydantic 的 FieldInfo 建字段，必填/可选由「有没有默认值」决定（_function_schema.py:184-190）。
  • docstring 被解析成「工具描述 + 每个参数的描述」（doc_descriptions，_function_schema.py:146）。
# 示意，非源码：这个函数会被翻成什么
@agent.tool
async def get_weather(ctx: RunContext[Deps],  # RunContext → 跳过，不进 schema
                      city: str,               # 必填 string
                      units: str = 'celsius'   # 可选，有默认值
                      ) -> str:
    """查城市天气。

    Args:
        city: 城市名
        units: 温度单位
    """
    ...
# 生成的 schema ≈ {name: 'get_weather', description: '查城市天气。',
#   parameters: {city: {type: string, description: '城市名'}, units: {...}},
#   required: ['city']}

关键细节： 是否「带 ctx」可以不显式声明——靠看第一个参数的注解推断（takes_ctx is None 时自动判断，_function_schema.py:150）。这就是为什么 @agent.tool（带 ctx）和 @agent.tool_plain（不带）能并存。

2.2 工具的运行期表示​

注册后的工具最终是一个 ToolDefinition（schema 那一面）加一个可调用体。FunctionToolset（toolsets/function.py:44）是最常见的工具来源——@agent.tool 注册的函数都进这里。它实现两个抽象方法：get_tools（列工具，function.py:599）和 call_tool（调工具，function.py:634）。

call_tool 很薄，但藏了一个贴心设计——超时自动转成 ModelRetry：

# toolsets/function.py:634 —— 调工具，超时 → 让模型重试（简化展示）
async def call_tool(self, name, tool_args, ctx, tool):
    timeout = tool.timeout if tool.timeout is not None else self.timeout
    if timeout is not None:
        try:
            with anyio.fail_after(timeout):
                return await tool.call_func(tool_args, ctx)
        except TimeoutError:
            raise ModelRetry(f'Timed out after {timeout} seconds.') from None
    else:
        return await tool.call_func(tool_args, ctx)

注意 ModelRetry：工具超时不是直接报错崩掉，而是变成「告诉模型超时了，请重试」喂回去——这是贯穿全库的容错哲学（同样的还有参数校验失败、ModelRetry 由工具主动抛出）。

2.3 toolset：工具来源的抽象​

它要解决的小问题： 工具可能来自好几处——你 @agent.tool 注册的、MCP server 提供的、输出工具、外部延迟工具……得有个统一接口。

这个接口是 AbstractToolset（toolsets/abstract.py:74），职责三件：列工具 / 校验参数 / 调工具。关键抽象方法：

  • get_tools(ctx)（abstract.py:164）—— 返回 {工具名: ToolsetTool}。
  • call_tool(name, args, ctx, tool)（abstract.py:169）—— 调指定工具。

还有两个生命周期钩子很重要：

  • for_run(ctx)（abstract.py:110）—— 每次运行调一次，可返回「本次运行专用的新实例」做状态隔离。
  • for_run_step(ctx)（abstract.py:118）—— 每轮（每次模型请求前）调一次，用于「随对话推进而变的工具集」。

这就是为什么工具列表可以逐轮变化（比如某些工具只在前几轮可用）——ModelRequestNode._prepare_request 每轮都会调 tool_manager.for_run_step（见 01 章 §1.4）。

2.4 toolset 的组合：装饰器模式​

Pydantic AI 用装饰器/包装器模式把 toolset 叠起来。toolsets/ 目录下一堆 XxxToolset 大多是包装别的 toolset 再改一点行为：

Toolset干什么
CombinedToolset把多个 toolset 合成一个（最常用）
PrefixedToolset给工具名加前缀，避免重名
RenamedToolset重命名工具
FilteredToolset按条件过滤掉一些工具
PreparedToolset每轮动态改写工具定义
ApprovalRequiredToolset给工具加「需人类审批」标记
WrapperToolset包装基类，上面这些大多继承它

AbstractToolset 上还有 .prefixed()、.filtered() 这类便捷方法（abstract.py:192,203），直接返回对应的包装 toolset，可链式叠加。

2.5 调用时怎么路由：CombinedToolset​

模型说「调 get_weather」，框架怎么知道这个名字属于哪个 toolset？答案在 CombinedToolset（toolsets/combined.py:26）：它的 get_tools 把各子 toolset 的工具合并（combined.py:66），并记住「哪个工具名归哪个子 toolset」；call_tool（combined.py:90）按名字找回对应子 toolset 再转发调用。

名字冲突在这一层被主动报错（合并时检测重名），错误信息里还会带 tool_name_conflict_hint（abstract.py:106），提示你用 PrefixedToolset 解决。

2.6 三种「不立即返回结果」的工具​

普通工具调完就有返回值。但有三类工具的「结果」要绕一圈：

  • 延迟工具（deferred） —— 工具抛 CallDeferred（exceptions.py:80），表示「这个调用得交给外部系统异步处理」。本轮收集成 DeferredToolRequests，运行暂停，外部处理完再用 deferred_tool_results 续跑。
  • 需审批工具（approval / HITL） —— 工具抛 ApprovalRequired（exceptions.py:98），同样挂起，等人类批准/拒绝（ToolApproved / ToolDenied，tools.py:328,338）后续跑。
  • 外部工具（external） —— 工具定义只声明 schema，执行在框架外。

这三类共用同一套「挂起 → 收集请求 → 外部回填结果 → 续跑」机制，入口是 UserPromptNode._handle_deferred_tool_results（_agent_graph.py:385，见 01 章 §1.3）。这是 Pydantic AI 做 human-in-the-loop 的基础。

2.7 巧妙之处​

  • schema 永远和函数同步：因为 schema 是从签名实时生成的（_function_schema.py:103），改了函数签名 schema 自动跟着变，不会出现「文档写的参数和实现对不上」。
  • 容错优先于报错：工具超时（function.py:646）、参数校验失败都转成 ModelRetry 喂回模型，而不是让整个运行崩掉。让 agent 有「自我纠错」的回合。
  • 组合优于配置：toolset 用包装器模式叠加（toolsets/wrapper.py），加新行为 = 写一个新包装器，不用改核心。

2.8 边界​

  • 工具重试次数有上限，由 ToolManager 的 default_max_retries 把关（tool_manager.py），超了仍会失败。
  • 工具的并行/串行执行受 end_strategy 和 sequential 标记影响——见 03 章。

2.9 代码地图​

主题文件关键符号
函数→schemapydantic_ai_slim/pydantic_ai/_function_schema.pyfunction_schema, FunctionSchema
toolset 抽象pydantic_ai_slim/pydantic_ai/toolsets/abstract.pyAbstractToolset, ToolsetTool, get_tools, call_tool, for_run_step
函数工具集pydantic_ai_slim/pydantic_ai/toolsets/function.pyFunctionToolset, call_tool
组合路由pydantic_ai_slim/pydantic_ai/toolsets/combined.pyCombinedToolset
工具管理pydantic_ai_slim/pydantic_ai/tool_manager.pyToolManager
延迟/审批pydantic_ai_slim/pydantic_ai/exceptions.pyCallDeferred, ApprovalRequired
延迟/审批数据pydantic_ai_slim/pydantic_ai/tools.pyDeferredToolRequests, ToolApproved, ToolDenied