进阶 📋 6 个步骤 第 598 / 599 篇

用 Pydantic AI 写类型安全的 Python Agent(结构化输出 + 工具调用)

Pydantic AI 是 Pydantic 团队官方的 Python Agent 框架,把"类型安全"带到 Agent 开发:Agent 输出用 Pydantic 模型约束、工具即函数、依赖注入干净、还能用离线 test 模型先跑通。本教程从最小 Agent 到"结构化输出 + 工具 + 依赖注入 + 输出校验重试",每一步都可运行。

2026.11.04· 18 分钟阅读· 约 1480 字· 🐍 Pydantic AI / 🧩 结构化输出

你有没有遇到过:Agent 跑完了,返回的是一段自由文本,你想拿里面的"情感标签"或"来源列表"去接着写代码,却只能自己再写正则去抠?Pydantic AI 把这件事从"约定俗成"变成"类型保证"——它是 Pydantic 团队官方的 Python Agent 框架,Agent 的输出直接是一个校验过的 Pydantic 模型实例,工具就是普通函数,依赖注入干净,还能用离线 test 模型先把逻辑跑通。本教程从最小 Agent 到"结构化输出 + 工具 + 依赖注入 + 输出校验重试",每一步都可运行。

🧩 本教程适合:写过 Python、重视类型与工程化、受够了"解析 LLM 自由文本"的开发者。需要 Python 3.10+ 和一个 OpenAI(或任意支持的)API Key。

一、为什么需要"类型安全"的 Agent

Pydantic AI 的核心主张:Agent 的每一次运行结果都应该是类型明确的对象,而不是一段要你事后解析的字符串。它和另外两个常见选择定位不同:

方案擅长不适合
Pydantic AI需要 Agent 循环 + 工具 + 依赖 + 类型化输出只想做一次性结构化抽取
Instructor在现有 OpenAI / Anthropic 客户端上做单次结构化抽取需要多轮工具调用
LangGraph工作流是带持久状态的图想用最少的抽象快速起步

模型字符串有固定格式:provider:model,比如 openai:gpt-4o、anthropic:claude-...。写错前缀会直接报找不到模型的错。下面示例统一用 openai:gpt-4o,你换成自己账号里可用的任意模型即可。

Step 1:安装与最小 Agent

1 装包并跑一句

先用离线 test 模型验证整条链路,不花一分钱:

pip install pydantic-ai

from pydantic_ai import Agent

# 'test' 是内置离线模型,无需 Key,适合先跑通逻辑
agent = Agent("test")
result = agent.run_sync("用一句话介绍 Pydantic AI。")
print(result.output)
💡 想接真实模型时,设好 OPENAI_API_KEY 后把模型字符串改成 "openai:gpt-4o" 即可,其余代码一行不用动。

别用网上过时的模型名。各厂商模型名会变,以你账号里实际可用的为准。本教程示例用 gpt-4o 这种稳定型号,跑通后再按需替换。

Step 2:结构化输出(output_type)

2 让返回值是 Pydantic 模型

定义好模型,赋给 output_type,Agent 会自动校验 LLM 的回复是否符合结构:

from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent

class Sentiment(BaseModel):
    label: Literal["positive", "negative", "neutral"]
    score: float = Field(ge=-1, le=1)

agent = Agent("openai:gpt-4o", output_type=Sentiment)
result = agent.run_sync("大家对 Extract 这个 App 评价如何?")
print(result.output.label, result.output.score)
# result.output 是 Sentiment 实例,不是字符串
🔑 result.output 的类型被推导为 Sentiment,IDE 能自动补全字段,后续代码直接用 result.output.label 即可,不用再解析文本。

Step 3:给 Agent 加工具(@agent.tool)

3 工具即函数,靠签名和 docstring 生成 schema

在函数上加 @agent.tool,第一个参数通常是 ctx: RunContext[...]:

from pydantic_ai import Agent, RunContext

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

@agent.tool
def recent_reviews(ctx: RunContext, product: str) -> list[str]:
    """获取某产品最近的评论片段。Args: product: 产品名。"""
    # 真实项目里换成你的数据库 / HTTP 调用
    return ["新版本修好了我吐槽的所有问题!", "客服响应有点慢。"]

result = agent.run_sync("Extract 的口碑怎么样?")
print(result.output)

docstring 就是工具说明书。模型靠它判断"什么时候该调这个工具、传什么参数"。参数类型和说明要对得上,否则会传错或干脆不调。

Step 4:依赖注入(deps_type)

4 把连接 / 配置通过 deps 传进去

工具里需要数据库连接、API 客户端?别在工具内硬编码,用 deps_type 注入,测试时换假数据即可:

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class ReviewDeps:
    product_id: str

agent = Agent("openai:gpt-4o", deps_type=ReviewDeps, output_type=Sentiment)

@agent.tool
def recent_reviews(ctx: RunContext[ReviewDeps]) -> list[str]:
    """读取该产品的评论。"""
    pid = ctx.deps.product_id
    return [f"评论({pid}):整体不错"]

result = agent.run_sync(
    "总结下这个产品的口碑。",
    deps=ReviewDeps(product_id="extract-app"),
)
print(result.output)
💡 测试时把 deps 换成内存里的假对象,工具就能脱离真实数据库单测,这是 Pydantic AI 设计上最舒服的一点。

Step 5:输出校验与重试(@agent.output_validator + ModelRetry)

5 结果不对就让它重来

光有结构还不够,有时模型会"漏填"。用输出校验器在跑完后再把关,不满足就抛 ModelRetry 让它重试:

from pydantic_ai import Agent, ModelRetry

class ResearchBrief(BaseModel):
    summary: str
    sources: list[str]
    follow_up: list[str]

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

@agent.output_validator
def require_sources(brief: ResearchBrief) -> ResearchBrief:
    if not brief.sources:
        raise ModelRetry("请补充引用来源。")
    return brief

result = agent.run_sync("写一份关于类型安全 Agent 的简报,并给出来源。")
print(result.output.sources)

重试有预算。校验器抛 ModelRetry 会消耗重试额度,超出后框架会报错。把校验逻辑写得精准,别让模型陷入无限重试。

Step 6:可观测与下一步(Logfire / 离线 test)

6 先离线跑通,再上真实模型和可观测

开发期用 test 模型快速迭代,上线前接 Logfire 看每一步调用:

import logfire
from pydantic_ai import Agent

logfire.configure(send_to_logfire="if-token-present")
logfire.instrument_pydantic_ai()

agent = Agent("openai:gpt-4o", output_type=Sentiment)
result = agent.run_sync("这条用户反馈是正面还是负面?")
print(result.output)
🚀 推荐工作流:① 用 Agent("test") 把结构化输出、工具、校验的逻辑全跑通;② 再换成真实模型字符串;③ 接 Logfire(或任意 OpenTelemetry 后端)做可观测。这样既不烧钱又能早发现问题。

常见问题速查

现象大概率原因 & 解决
报错找不到模型模型字符串格式错了,确认是 provider:model
output 是字符串不是对象没设 output_type,或字段类型写错
工具没被调用docstring 不清楚,或 instructions 没提示用工具
一直 ModelRetry 报错校验条件太严,放宽或检查提示词
← 返回教程中心