用 Pydantic AI 构建类型安全的 LLM Agent

2026-05-12 19 预计阅读时间：1 分钟
来源：realpython.com AI 摘要 原文链接

免责声明：本文为 AI 摘要整理，建议结合原文阅读。摘要可能省略上下文、版本差异或边界条件，不作为官方说明。

预计阅读时间：10 分钟

写 LLM Agent 最让人头疼的不是调 prompt，而是运行时才发现模型返回的 JSON 缺了一个字段、函数签名传了错误类型、依赖注入全靠全局变量硬凑。Pydantic AI 把 Python 类型系统从头到尾拉进 Agent 开发——从输入参数到模型输出再到工具函数签名，每一步都有静态检查兜底，IDE 能提示，mypy 能校验，运行时还能兜住边界情况。

下面拆解它的三个核心机制：结构化输出、函数调用（Function Calling）和依赖注入，最后给一个可直接跑的完整示例。

结构化输出：让模型吐出你想要的形状

裸字符串响应在简单对话里够用，但 Agent 需要可编程的数据——状态码、决策分支、嵌套对象。Pydantic AI 的做法是：你定义一个 BaseModel，框架把它转成 JSON Schema 告给模型，模型按 schema 生成，框架再拿 Pydantic 校验反序列化。校验失败直接抛 ModelRetry，你可以在 retry 逻辑里给模型反馈修正。

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class Diagnosis(BaseModel):
    symptom: str = Field(description="患者主诉症状")
    severity: int = Field(ge=1, le=10, description="严重程度 1-10")
    recommendation: str = Field(description="建议处理方式")

doctor_agent = Agent(
    model="openai:gpt-4o",
    result_type=Diagnosis,  # 关键：声明结构化输出类型
)

result = doctor_agent.run_sync("我最近头疼得厉害，持续三天了")
print(result.data)  # Diagnosis 对象，字段齐全、类型正确

result_type 不填时默认返回字符串；填了 BaseModel，框架自动完成 schema 生成 → 模型调用 → 校验反序列化这条链路。Field(description=) 的内容会进入 schema 的 description 字段，直接影响模型的填写行为——写得越具体，输出越稳定。

函数调用：给 Agent 装上可校验的工具

Agent 不只是聊天，它要调数据库、发请求、读文件。Pydantic AI 用 @agent.tool 注册工具函数，函数的参数类型和返回类型全部参与校验：

from pydantic_ai import Agent, RunContext

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

@weather_agent.tool
async def get_weather(ctx: RunContext[None], city: str) -> str:
    """查询指定城市的当前天气"""
    # 实际项目中这里调 API，示例用硬编码
    mock_data = {"北京": "晴 22°C", "上海": "阴 18°C", "深圳": "雨 26°C"}
    return mock_data.get(city, f"{city} 暂无数据")

result = weather_agent.run_sync("上海今天天气怎么样？")
print(result.data)  # 模型会自动调用 get_weather，返回天气信息

几个细节值得注意：

  • 函数签名里的类型注解（city: str）会被转成 tool schema 的 parameters，模型只会在参数类型匹配时调用。
  • RunContext 是框架注入的上下文对象，第一个泛型参数对应依赖类型，后面会展开。
  • docstring 直接变成 tool 描述，模型靠它决定何时调用——写得模糊，模型就乱调或不调。

依赖注入：把状态从全局变量里解放出来

Agent 的工具函数经常需要数据库连接、API key、用户 session。硬编码或全局变量在单脚本里能跑，多用户并发时就炸。Pydantic AI 的依赖注入通过 RunContext 和 Agent 的泛型参数绑定：

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
    api_key: str
    db_url: str

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

@db_agent.tool
async def query_user(ctx: RunContext[Deps], user_id: int) -> str:
    """按 ID 查询用户信息"""
    # ctx.deps 拿到注入的依赖，类型安全
    print(f"使用 db_url: {ctx.deps.db_url}")
    return f"用户 {user_id}: 张三, VIP"

deps = Deps(api_key="sk-xxx", db_url="postgresql://localhost/mydb")
result = db_agent.run_sync("查一下用户 42 的信息", deps=deps)
print(result.data)

deps_type=Deps 声明依赖类型，run_sync(deps=deps) 注入实例，工具函数通过 ctx.deps 访问——全程类型可追踪，IDE 自动补全 ctx.deps.db_url，传错类型 mypy 直接报红。

完整示例：类型安全的订单查询 Agent

把三个机制串起来，写一个能跑的完整例子。你需要先安装依赖：

pip install pydantic-ai openai

然后设置环境变量：

export OPENAI_API_KEY="sk-your-key-here"

以下是完整代码，保存为 order_agent.py 直接运行：

import asyncio
from dataclasses import dataclass
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

# ---- 1. 定义结构化输出 ----
class OrderResult(BaseModel):
    order_id: str = Field(description="订单编号")
    status: str = Field(description="订单状态：pending / shipped / delivered")
    total: float = Field(gt=0, description="订单金额")
    summary: str = Field(description="一句话概括订单情况")

# ---- 2. 定义依赖 ----
@dataclass
class OrderDeps:
    mock_orders: dict[str, dict]  # 模拟数据库

# ---- 3. 创建 Agent 并注册工具 ----
order_agent = Agent(
    model="openai:gpt-4o",
    result_type=OrderResult,
    deps_type=OrderDeps,
    system_prompt="你是一个订单查询助手。根据用户提供的订单号查询信息，用结构化数据返回。",
)

@order_agent.tool
async def lookup_order(ctx: RunContext[OrderDeps], order_id: str) -> str:
    """按订单号查询订单详情"""
    order = ctx.deps.mock_orders.get(order_id)
    if not order:
        return f"订单 {order_id} 不存在"
    return f"状态: {order['status']}, 金额: {order['total']} 元, 商品: {order['items']}"

# ---- 4. 准备数据并运行 ----
mock_db = {
    "ORD-001": {"status": "shipped", "total": 299.0, "items": "机械键盘"},
    "ORD-002": {"status": "pending", "total": 59.0, "items": "鼠标垫"},
}

deps = OrderDeps(mock_orders=mock_db)

async def main():
    result = await order_agent.run("帮我查一下 ORD-001 的情况", deps=deps)
    print(result.data.model_dump())

asyncio.run(main())

运行预期输出大致如下（模型生成的 summary 内容会有差异）：

{'order_id': 'ORD-001', 'status': 'shipped', 'total': 299.0, 'summary': '机械键盘订单已发货，金额 299 元'}

这个例子把三个机制全串起来了：OrderResult 锁定输出形状，lookup_order 的参数类型约束模型只能传字符串订单号，OrderDeps 把模拟数据库注入到工具函数里。每一步都有类型守门，模型乱填字段会触发校验重试，工具参数类型不对框架直接拦截。

采纳建议与边界

适合用的场景：

  • Agent 需要稳定输出给下游系统消费（API、数据库写入、自动化流程）。
  • 多工具 Agent，工具之间有共享依赖（DB 连接、认证 token）。
  • 团队协作项目，类型检查能减少"模型返回了什么形状"的口头沟通成本。

需要注意的边界：

  • 结构化输出依赖模型的 function calling / JSON mode 能力，小模型或开源模型支持程度不一，测试时先确认模型端是否可靠按 schema 生成。
  • Pydantic 校验失败会触发 ModelRetry，但重试次数有限（默认 2 次），复杂 schema 在弱模型上可能反复失败——此时简化 schema 或换模型比加 retry 更实际。
  • 依赖注入是同步构造的，如果依赖本身需要异步初始化（比如建数据库连接池），要在 Agent 外部完成，把已初始化的对象作为 deps 传入。

上手路径建议：

  1. 先用最简 Agent（result_type=str，无工具）跑通模型调用，确认环境正常。
  2. 加一个 BaseModel 做 result_type，感受校验和 schema 描述对输出的约束力。
  3. 加一个 @agent.tool，体会参数类型如何限制模型的调用行为。
  4. 最后引入 deps_type，把硬编码的连接和配置替换成注入对象。

类型安全不是银弹，但它把 Agent 开发里最不可控的"模型输出形状"和"工具调用参数"这两件事拉进了可校验的范围。Pydantic AI 做的不是让模型更聪明，而是让模型犯的类型错误能被程序提前拦住——这在生产环境里比调 prompt 更省心。

相关推荐