何时使用
用 Python 构建类型安全、可测试的 LLM 智能体时使用。PydanticAI 出自 Pydantic 团队,把 Pydantic 的校验保证带到 LLM 应用:结构化输出(Pydantic 模型校验)、工具调用、依赖注入、流式响应、多轮对话,统一覆盖 OpenAI / Anthropic / Gemini / Groq / Mistral / Ollama。
该用:
- 需要模型返回校验过的强类型结构而非裸字符串。
- 智能体要调用工具(查 DB、调 API)并产出结构化结果。
- 想不打真实 LLM 就为智能体逻辑写单测。
- 想在不重写业务代码的前提下切换模型供应商。
- 用户提到
Agent、@agent.tool、RunContext、ModelRetry、result_type。
不该用(边界):
- 一次性裸提问、demo 脚本 → 直接调 SDK,别套框架。
- 任务只是设计/迭代提示词 → 用
prompt-template-designer。 - 需要文档/知识库检索增强 → 用
rag-pipeline-builder,再用本技能包装为 Agent。 - 非 Python 技术栈 → 本技能不适用。
步骤 / 指令
- 安装:按供应商装 extras。
pip install pydantic-ai pip install 'pydantic-ai[openai]' # OpenAI / Azure pip install 'pydantic-ai[anthropic]' # Anthropic pip install 'pydantic-ai[gemini]' # Gemini - 定输出契约:用
BaseModel定义result_type,生产环境不要返回裸字符串。字段尽量用Optional/default,过严 schema 会反复ValidationError。 - 建 Agent:
Agent('供应商:模型', result_type=..., system_prompt=...);API Key 走环境变量,勿写进Agent()参数。 - 挂工具:
@agent.tool装饰函数,docstring 必填(会作为工具描述发给 LLM,缺了模型不会调);需上下文时签名用RunContext[Deps]。 - 依赖注入:用
@dataclass定义Deps,建 Agent 时传deps_type=Deps;调用时agent.run(msg, deps=...)按次传入,依赖不是全局的。 - 业务校验/重试:
@agent.result_validator里抛ModelRetry(...)触发模型重答;Agent(..., retries=N)限制重试上限防死循环。 - 测试:单测用
agent.override(model=TestModel()),断言结构而非措辞;需确定性输出用FunctionModel。 - 流式 / 多轮:长输出用
run_stream;多轮对话用result.all_messages()取历史,下一轮传message_history=。
调用形式:同步 agent.run_sync(...);异步 await agent.run(...)(FastAPI 路由里直接 await,勿包 asyncio.run());取结果用 result.data,取用量用 result.usage()。
示例
最小结构化输出 Agent:
from pydantic import BaseModel
from pydantic_ai import Agent
class MovieReview(BaseModel):
title: str
year: int
rating: float # 0.0~10.0
recommended: bool
agent = Agent('openai:gpt-4o', result_type=MovieReview,
system_prompt='You are a film critic. Return structured reviews.')
review = agent.run_sync('Review Inception (2010)').data # 强类型实例
print(f"{review.title} ({review.year}): {review.rating}/10")
工具 + 依赖注入(客服场景,节选):
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
@dataclass
class Deps:
db: Database
user_id: str
class SupportResponse(BaseModel):
message: str
escalate: bool
support_agent = Agent('openai:gpt-4o-mini', deps_type=Deps,
result_type=SupportResponse,
system_prompt='You are a support agent. Use the tools to help customers.')
@support_agent.tool
async def get_order_history(ctx: RunContext[Deps]) -> list[dict]:
"""Fetch recent orders for the current user.""" # docstring 即工具描述
return await ctx.deps.db.get_orders(ctx.deps.user_id, limit=5)
async def handle(user_id: str, msg: str):
deps = Deps(db=get_db(), user_id=user_id)
return (await support_agent.run(msg, deps=deps)).data
业务校验触发重试:
from pydantic_ai import ModelRetry
@agent.result_validator
async def validate(ctx, result: StrictJson) -> StrictJson:
if result.value > 1000:
raise ModelRetry('Value must be under 1000. Try again with a smaller number.')
return result
不打真实 LLM 的单测:
from pydantic_ai.models.test import TestModel
def test_escalates():
with support_agent.override(model=TestModel()):
r = support_agent.run_sync('I want to cancel',
deps=Deps(db=FakeDb(), user_id='u-123'))
assert isinstance(r.data, SupportResponse)
assert isinstance(r.data.escalate, bool) # 测结构,不测措辞
注意事项
- API Key 走环境变量(
OPENAI_API_KEY/ANTHROPIC_API_KEY…),绝不硬编码、绝不写进Agent()参数。 - 工具 docstring 必写且具体:它是发给 LLM 的工具描述,写空了模型不会调用。
- 依赖按次传
deps=:RunContext里deps为None,多半是run()时漏传;依赖非全局。 - 结构化输出老是校验失败:放宽
result_type,加Optional/default,别堆过严字段。 result_type优先于裸字符串;用result_validator补 Pydantic 之外的业务规则;retries=设上限防死循环。- 会变更数据的工具(写库、发邮件、调支付)生产环境应要求显式确认后再让 Agent 调用;对有后果的动作记录
result.all_messages()做审计。 - 别广捕
ValidationError:交给ModelRetry让框架重试可恢复的输出错误。 - 跨 async 任务勿共享单一 Agent 实例(若 deps 不同):按请求建实例或按次传 deps。
- FastAPI 中直接
await agent.run(),不要包asyncio.run()。
互见
- related:
prompt-template-designer—— 智能体的system_prompt与工具描述可由其产出更稳定的模板。 - combines_with:
rag-pipeline-builder—— 把检索管道封装成工具挂到 Agent 上,得到带引用的检索增强智能体。
采编自 sickn33/antigravity-awesome-skills(MIT),适配重写。