Tool · 让 Agent 接入真实世界

一、从"一个普通函数"开始

Pydantic AI 的工具,其实就是一个普通 Python 函数。没有特殊基类,没有必须继承的东西。装饰器一贴,它就成了 Agent 可用的 tool:

from pydantic_ai import Agent

agent = Agent("openai:gpt-4o")

@agent.tool_plain
def add(a: int, b: int) -> int:
    """两数相加。"""
    return a + b

result = agent.run_sync("算一下 2378 加 5619")
print(result.output)   # 7997

这一段到底发生了什么?Pydantic AI 做了 4 件事:

1. 读函数签名,add(a: int, b: int) -> int 变成 JSON Schema {"a": int, "b": int}
2. 读 docstring "两数相加",塞进 schema 的 description
3. 把 schema 注册给 LLM,告诉它"你有个叫 add 的工具可以用"
4. LLM 决定调用时,Pydantic AI 用校验过的参数调你的 Python 函数,把返回值发回给 LLM

二、两个装饰器:@agent.tool vs @agent.tool_plain

@agent.tool
def lookup_user(ctx: RunContext[DbDeps], user_id: int) -> dict:
    return ctx.deps.db.fetch_user(user_id)   # 通过 ctx.deps 拿到注入的 DB 连接

三、RunContext:工具里能访问到什么

RunContext 是工具的"瞭望台",可以看到当前这次 run 的全部上下文:

from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class Deps:
    db: object
    user_id: int

agent = Agent("openai:gpt-4o", deps_type=Deps)

@agent.tool
def show_context(ctx: RunContext[Deps]) -> str:
    print(ctx.deps)            # 注入的依赖
    print(ctx.model)           # 当前用的模型
    print(ctx.usage)           # 目前累计用量
    print(ctx.prompt)          # 本次 run 的初始用户输入
    print(ctx.messages)        # 目前所有消息
    print(ctx.retry)            # 本工具当前第几次重试
    print(ctx.tool_name)        # 当前工具名(便于泛用 tool)
    return "ok"

四、自动推 schema:你写的 Python,LLM 一眼看懂

这是 Pydantic AI 最"省心"的部分。来看一个较复杂的签名:

from typing import Literal
from pydantic import Field

@agent.tool_plain
def search_orders(
    keyword: str,
    status: Literal["paid", "shipped", "delivered"] = "paid",
    limit: int = 10,
    include_refunded: bool = False,
) -> list[dict]:
    """根据关键词搜索订单。

    关键词会在订单的 user_name/product_name/note 三个字段中模糊匹配。
    status 过滤订单当前状态。
    limit 最多返回多少条(最大 50)。
    """
    ...

Pydantic AI 自动生成的 schema(送给 LLM 的)大致长这样:

{
  "name": "search_orders",
  "description": "根据关键词搜索订单。\n\n关键词会在订单的 user_name/product_name/note 三个字段中模糊匹配。\nstatus 过滤订单当前状态。\nlimit 最多返回多少条(最大 50)。",
  "parameters": {
    "type": "object",
    "properties": {
      "keyword": {"type": "string"},
      "status": {"type": "string", "enum": ["paid", "shipped", "delivered"], "default": "paid"},
      "limit": {"type": "integer", "default": 10},
      "include_refunded": {"type": "boolean", "default": false}
    },
    "required": ["keyword"]
  }
}

默认值自动识别、Literal 自动转 enum、docstring 自动进 description——你完全不用手写 JSON Schema。

参数描述的精细控制

三种方法把每个参数的描述告诉 LLM,择其一即可:

(a) docstring Google/Numpy/Sphinx 风格

@agent.tool_plain
def translate(text: str, target_lang: str) -> str:
    """翻译一段文本。

    Args:
        text: 要翻译的原文。
        target_lang: 目标语言的 ISO 639-1 代码,如 "en"、"zh"、"ja"。
    """
    ...

(b) Annotated + Field(推荐,可带校验)

from typing import Annotated
from pydantic import Field

@agent.tool_plain
def transfer(
    amount: Annotated[float, Field(gt=0, le=1_000_000, description="转账金额,大于 0 且不超过 100 万")],
    to_account: Annotated[str, Field(min_length=10, pattern=r"^\d+$", description="收款账号,至少 10 位纯数字")],
) -> str:
    ...

(c) 用 Pydantic 模型当参数

class TransferReq(BaseModel):
    amount: float = Field(gt=0, le=1_000_000)
    to_account: str = Field(min_length=10)

@agent.tool_plain
def transfer(req: TransferReq) -> str:
    return do_transfer(req.amount, req.to_account)

五、同步工具 vs 异步工具

两种都支持,Agent 里可以混用——Pydantic AI 自动识别:

@agent.tool_plain
def sync_one(x: int) -> int:
    return x * 2

@agent.tool_plain
async def async_one(x: int) -> int:
    await asyncio.sleep(0.1)
    return x * 3

六、并行工具调用:一次多调几个

现代 LLM(GPT-4o、Claude、Gemini)都支持一次响应里发起多个工具调用。Pydantic AI 会并发执行这些工具:

@agent.tool_plain
async def get_weather(city: str) -> str:
    await asyncio.sleep(0.3)  # 模拟 API
    return f"{city}: 晴,20°C"

@agent.tool_plain
async def get_time(city: str) -> str:
    await asyncio.sleep(0.3)
    return f"{city}: 14:30"

agent = Agent("openai:gpt-4o", tools=[get_weather, get_time])

result = await agent.run(
    "告诉我北京和上海现在的天气和时间",
    model_settings={"parallel_tool_calls": True},  # 默认就是 True
)

这里模型大概率会一次发起 4 个工具调用(北京/上海 × 天气/时间),Pydantic AI 并发跑完再一起发回去。如果串行跑要 1.2s,并行大概 0.3s。

七、ModelRetry:主动要求重试

LLM 给的参数不合理(业务层面),你希望它重选一次?在工具里抛 ModelRetry:

from pydantic_ai import Agent, ModelRetry, RunContext

@agent.tool
async def get_stock_price(ctx: RunContext[Deps], symbol: str) -> float:
    if symbol not in KNOWN_SYMBOLS:
        raise ModelRetry(
            f"未知股票代码 '{symbol}'。"
            f"请使用 6 位数字 A 股代码(如 '600519')或美股英文 ticker(如 'AAPL')。"
        )
    return await price_api.fetch(symbol)

模型收到这段错误文本后,会重新挑选参数再调一次。和 output_validator 的 ModelRetry 一样,最多重试次数 = 工具的 retries 或 Agent 的 retries(工具级优先)。

工具级 retries 的写法:

@agent.tool_plain(retries=3)
async def get_stock_price(symbol: str) -> float:
    ...

八、工具的三种注册方式

(a) 装饰器(最常用)

@agent.tool_plain
def my_tool(x: int) -> int:
    return x + 1

(b) 构造 Agent 时通过 tools= 传

from pydantic_ai import Agent, Tool

def my_tool(x: int) -> int:
    return x + 1

agent = Agent("openai:gpt-4o", tools=[my_tool])
# 或显式:
agent = Agent("openai:gpt-4o", tools=[Tool(my_tool, name="increment", max_retries=2)])

(c) 动态工具准备函数 toolsets

某些场景下,可用工具集本身依赖运行时状态——比如登录用户的权限决定能调哪些工具。Pydantic AI 有 FunctionToolset / PrepareTools 的玩法:

from pydantic_ai.toolsets import FunctionToolset

admin_ts = FunctionToolset(tools=[delete_user, promote_user])
user_ts = FunctionToolset(tools=[list_my_orders, get_profile])

def select_toolsets(ctx: RunContext[UserDeps]):
    if ctx.deps.is_admin:
        return [admin_ts, user_ts]
    return [user_ts]

agent = Agent(
    "openai:gpt-4o",
    deps_type=UserDeps,
    toolsets=select_toolsets,
)

这是第 7-8 章讲多 Agent / 复杂路由时会深入的机制,先知道有这回事即可。

九、工具返回值能是什么?

Pydantic AI 很宽容——工具返回值只要是 可以序列化的东西 就行:

• 基础类型:str / int / float / bool / None
• 容器:list / dict / tuple / set(内部必须可序列化)
• Pydantic BaseModel / dataclass / TypedDict
• ToolReturn:如果想精细控制发给 LLM 的内容 + 附加 metadata

from pydantic_ai.messages import ToolReturn

@agent.tool_plain
def complex_result() -> ToolReturn:
    return ToolReturn(
        return_value={"answer": 42},             # 给业务代码看
        content="查询完成,答案是 42",              # 给 LLM 看的自然语言表述
        metadata={"source": "internal-db"},       # trace 附加信息
    )

十、一个完整示例:天气 Agent(工具 + 重试 + 并行)

import asyncio
from dataclasses import dataclass
from typing import Literal, Annotated
from pydantic import BaseModel, Field
from pydantic_ai import Agent, ModelRetry, RunContext

@dataclass
class Deps:
    http_client: object       # 真实项目里是 httpx.AsyncClient

class WeatherReport(BaseModel):
    city: str
    temp_c: float
    condition: Literal["sunny", "cloudy", "rain", "snow", "fog"]

agent = Agent(
    "openai:gpt-4o-mini",
    deps_type=Deps,
    output_type=list[WeatherReport],
    system_prompt="你是天气助手。当用户询问多个城市时,使用工具并行查询。",
)

@agent.tool
async def geocode(
    ctx: RunContext[Deps],
    city: Annotated[str, Field(description="中文或英文城市名")],
) -> dict:
    """把城市名转成经纬度。"""
    # 这里模拟 API 调用
    coords = {"北京": (39.90, 116.41), "上海": (31.23, 121.47)}.get(city)
    if not coords:
        raise ModelRetry(f"未知城市 '{city}',请使用'北京'、'上海'等标准中文名。")
    return {"city": city, "lat": coords[0], "lon": coords[1]}

@agent.tool
async def fetch_weather(
    ctx: RunContext[Deps],
    lat: float,
    lon: float,
) -> dict:
    """根据经纬度获取天气。"""
    # 假装调 open-meteo
    return {"temp_c": 14.0, "weather_code": "cloudy"}

async def main():
    deps = Deps(http_client=None)
    result = await agent.run("告诉我北京和上海的天气", deps=deps)
    for r in result.output:
        print(r)

asyncio.run(main())

这里就能看出 Pydantic AI 的工程感了:

• 工具返回 dict,但 Agent 最终结构化输出 list[WeatherReport]——两层格式都被类型约束
• 模型先并行调两次 geocode,再并行调两次 fetch_weather,最后一次总结
• 未知城市触发 ModelRetry——模型自动修正为标准名

十一、工具的执行顺序与"agent loop"

Agent 内部循环大致如下:

循环会一直进行到:① 成功获得 output;② 达到最大迭代(result.all_messages() 里轮次过多,默认限制受 provider 实现影响,一般 50 轮)。

十二、八个常见坑

1. 工具返回巨型对象:几千行数据塞回去,token 爆表。返回前做摘要/分页,给 LLM 看的是"视图"。
2. 工具里抛 raw Exception:会直接冒出来,模型看不到。业务错用 ModelRetry,系统错才让它抛。
3. 工具签名里混入 **kwargs:Pydantic AI 推不出 schema,会忽略。要么写死参数,要么用 BaseModel 做参数。
4. 工具名冲突:两个 add 函数,同一个 Agent 里,后注册覆盖前一个(不报错)。生产务必用语义清晰、全局唯一的工具名。
5. 忘记 await async 工具:你自己直接调 my_async_tool(1, 2) 得到的是 coroutine 对象,不是结果。LLM 调是 Pydantic AI 帮你 await,这个坑只在你手工测时才出现。
6. 工具写在循环里:for city in cities: @agent.tool def f()...——装饰器闭包陷阱,所有 f 最终都引用同一个城市。用 toolsets 或 factory pattern 解决。
7. 同步工具里调阻塞 IO:FastAPI 并发场景下全站变慢。全部 async 化。
8. ModelRetry 信息写得太模糊:"参数错误"这种让模型没法改正。写具体:期望什么格式、取值在哪、例子是什么。

十三、本章小结