你有没有遇到过:Agent 跑完了,返回的是一段自由文本,你想拿里面的"情感标签"或"来源列表"去接着写代码,却只能自己再写正则去抠?Pydantic AI 把这件事从"约定俗成"变成"类型保证"——它是 Pydantic 团队官方的 Python Agent 框架,Agent 的输出直接是一个校验过的 Pydantic 模型实例,工具就是普通函数,依赖注入干净,还能用离线 test 模型先把逻辑跑通。本教程从最小 Agent 到"结构化输出 + 工具 + 依赖注入 + 输出校验重试",每一步都可运行。
一、为什么需要"类型安全"的 Agent
Pydantic AI 的核心主张:Agent 的每一次运行结果都应该是类型明确的对象,而不是一段要你事后解析的字符串。它和另外两个常见选择定位不同:
| 方案 | 擅长 | 不适合 |
|---|---|---|
| Pydantic AI | 需要 Agent 循环 + 工具 + 依赖 + 类型化输出 | 只想做一次性结构化抽取 |
| Instructor | 在现有 OpenAI / Anthropic 客户端上做单次结构化抽取 | 需要多轮工具调用 |
| LangGraph | 工作流是带持久状态的图 | 想用最少的抽象快速起步 |
模型字符串有固定格式:provider:model,比如 openai:gpt-4o、anthropic:claude-...。写错前缀会直接报找不到模型的错。下面示例统一用 openai:gpt-4o,你换成自己账号里可用的任意模型即可。
Step 1:安装与最小 Agent
先用离线 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)
定义好模型,赋给 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)
在函数上加 @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)
工具里需要数据库连接、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)
光有结构还不够,有时模型会"漏填"。用输出校验器在跑完后再把关,不满足就抛 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)
开发期用 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 报错 | 校验条件太严,放宽或检查提示词 |