pydanticai

PydanticAI 和 PydanticGraph 专家技能

PydanticAI 是一个用于构建生产级 GenAI 应用的 Python Agent 框架，由 Pydantic 背后的团队打造。PydanticGraph 是与之配套的图／状态机库。

安装：

pip install pydantic-ai               # Full install (all providers)
pip install "pydantic-ai-slim[openai]" # Minimal install + your provider

快速参考

from pydantic_ai import Agent

# Basic agent — one line
agent = Agent('openai:gpt-5.2', instructions='Be concise.')

# Run it
result = agent.run_sync('What is the capital of France?')
print(result.output)

何时加载哪个参考文档

主题	加载时机	文件
Agent 创建与生命周期	需要创建、配置或运行 Agent——定义工具、依赖项、输出类型、运行方法或流式处理时	references/core-agents.md
能力与钩子	需要内置能力（Thinking、WebSearch、MCP 等）、按需加载、生命周期钩子或自定义能力时	references/capabilities-hooks.md
PydanticGraph	需要状态机、基于图的控制流、并行执行、BaseNode 子类或包含汇合／决策的 GraphBuilder 时	references/graph.md
模型、输出与流式处理	需要多模型设置、FallbackModel、流式输出、输出函数或带验证的结构化输出时	references/models-output.md
多 Agent 模式与集成	需要 Agent 委派、程序化交接、MCP 服务、持久化执行或 UI 适配器时	references/patterns.md
测试与评估	需要 TestModel、FunctionModel、pytest 模式、覆盖项或用于系统化评估的 Pydantic Evals 时	references/testing-evals.md
完整示例	需要完整且可运行的示例——银行客服 Agent、邮件反馈图、多 Agent 航班预订时	references/examples.md
框架边界	需要为项目比较 PydanticAI 与 LangGraph，或想将两者结合使用时	references/hybrid-pydanticai-langgraph.md——同时加载 skill_view(name='langgraph')
Agent 中的类型化 System One 决策	需要注入类型化决策服务、验证其响应结构，或将决策与生成式输出分开时	references/core-agents.md 和 System One；关于放置位置、权限、恢复和完整任务证据，请使用 harness-engineering
API 接口参考	需要快速查找正确的导入路径、类名或方法签名时	references/api-reference.md

常见模式一览

带工具和结构化输出的 Agent

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

class WeatherResult(BaseModel):
    temperature: float
    conditions: str

agent = Agent('openai:gpt-5.2', output_type=WeatherResult)

@agent.tool
async def get_weather(ctx: RunContext, city: str) -> str:
    """Get current weather for a city."""
    return f"24°C and sunny in {city}"

result = agent.run_sync('Weather in London?')
print(result.output.temperature)

→ 完整的 Agent 生命周期、运行方法和工具模式，请参阅 references/core-agents.md。

带依赖注入的 Agent

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    api_key: str
    db_conn: str

agent = Agent('openai:gpt-5.2', deps_type=MyDeps)

@agent.tool
async def query_db(ctx: RunContext[MyDeps], sql: str) -> str:
    return f"Query results using {ctx.deps.db_conn}"

→ 依赖注入模式和测试覆盖项，请参阅 references/core-agents.md。

包含多个节点的图

from dataclasses import dataclass
from pydantic_graph import BaseNode, End, GraphRunContext, GraphBuilder

@dataclass
class MyState:
    value: int = 0

@dataclass
class ProcessNode(BaseNode[MyState]):
    async def run(self, ctx: GraphRunContext[MyState]) -> End | NextNode:
        ctx.state.value += 1
        if ctx.state.value >= 5:
            return End(ctx.state.value)
        return NextNode()

→ BaseNode 和 GraphBuilder API、并行执行以及汇合／归约器模式，请参阅 references/graph.md。

如何选择运行方法

使用场景	方法	主要行为
同步代码中获取单个答案	run_sync()	阻塞直至完成，返回 RunResult
异步代码中获取单个答案	run()	异步执行，返回 RunResult
流式接收生成的文本	run_stream()	异步上下文管理器，生成 stream_text() / stream_output()
查看细粒度事件（工具调用、部分开始、增量内容）	run_stream_events()	生成 AgentStreamEvent 类型——FunctionToolCallEvent、PartStartEvent、FinalResultEvent
手动控制每个图步骤	iter()	遍历 Agent 的内部图节点（UserPromptNode → ModelRequestNode → CallToolsNode）
流式处理期间执行工具调用	run_stream_events() 或 run(event_stream_handler=...)	run_stream() 会在首个符合 output_type 的输出处停止，且不会执行后续工具调用

各运行方法的详细信息见 references/core-agents.md。

Graph API：BaseNode 与 GraphBuilder

因素	BaseNode（基于类）	GraphBuilder（基于函数）
风格	子类化 BaseNode[StateT]，实现 async run()	使用 @g.step 装饰异步函数
状态变更	在 run() 方法中通过 ctx.state 进行	在步骤函数中通过 ctx.state 进行
并行处理	手动编写分叉／汇合逻辑	内置 .map()，用于逐元素扇出；内置 .broadcast()，用于将相同输入发送到多个目标
汇合／聚合	在返回类型中手动聚合	内置归约器：reduce_list_append、reduce_sum、reduce_dict_update 等
边的声明	根据 run() 返回类型注解推断	通过 g.edge_from(source).to(target) 显式声明
适用场景	复杂节点逻辑、面向对象模式、有条件的边逻辑	简单线性流、并行数据处理、简洁语法

两种 API 均见 references/graph.md。

框架边界：PydanticAI 与 LangGraph

两个框架都能通过图和工具构建 Agent 系统，但设计理念截然不同。正确的选择取决于你的优化目标。

因素	PydanticAI + PydanticGraph	LangGraph	两者结合使用
设计理念	类型安全、以数据模式为驱动。感觉类似 FastAPI。	底层图原语（受 Pregel/Beam 启发）。感觉类似 NetworkX。	用 PydanticAI 构建 Agent 层；用 LangGraph 处理复杂编排
Agent 定义	Agent(model, tools, deps, output_type) — 声明式，只需一行	手动使用 StateGraph 节点进行消息传递	将 PydanticAI Agent 作为节点函数嵌入 LangGraph StateGraph
工具调用	@agent.tool 装饰器，根据类型提示自动生成模式，通过 RunContext 进行依赖注入	手动注册工具，tool_node = ToolNode(tools)	在 LangGraph 节点中使用 PydanticAI 的类型化工具定义
状态管理	GraphRunContext.state — 可变数据类，存储在内存中	使用类型化 reducer 和检查点保存器（SQLite/Postgres）的 State	使用 LangGraph 检查点保存器管理外层流程；使用 PydanticGraph 管理子图状态
多 Agent 模式	Agent 委派（工具调用）、程序化移交、基于图的模式	Supervisor（中央路由器）、swarm（直接移交）、分层模式（子图）	在 LangGraph supervisor 节点中进行 PydanticAI 委派
持久化	通过 Temporal、Inngest、Prefect、DBOS 实现持久执行	内置检查点保存器（MemorySaver、SqliteSaver、PostgresSaver）	在图级别使用 LangGraph 检查点保存器
流式传输	5 种方法：run、run_sync、run_stream、run_stream_events、iter	在编译后的图上调用 .stream() / .astream_events()	由 LangGraph .astream_events() 封装 PydanticAI 事件处理程序
学习曲线	较低 — 类型提示可为所有操作提供指引	较高 — 需要更多手动连接	最高 — 需要掌握两种思维模型
最适合	单 Agent、使用工具的工作流、类型安全的结构化输出、刚接触 Agent 的团队	复杂状态机、带分支/循环的多 Agent、HITL、现有 LangChain 用户	需要类型安全 Agent 和高级编排的大型系统

边界条件 — 以下情况可考虑 LangGraph：

• 需要为长时间运行的对话提供内置检查点保存/持久化功能（内置 SQLite、Postgres 后端）
• 多 Agent 系统需要通过隔离的状态命名空间组合子图
• 需要人在回路模式（中断/恢复、状态编辑、审批工作流）
• 已在使用 LangChain，且希望保持一致
• 图需要通过 Send() 实现循环或动态扇出

以下情况可考虑 PydanticAI：

• 类型安全和 IDE 自动补全是优先事项
• 希望通过声明式 Agent 最大限度地减少样板代码
• 需要具有自动验证和重试功能的结构化输出
• 多 Agent 需求仅涉及简单委派或顺序移交
• 重视可组合的能力系统（Thinking、WebSearch、MCP 作为插件）

以下情况可考虑两者结合使用：

• 外层循环需要 LangGraph 的编排能力（检查点保存、子图、HITL），但内层 Agent 逻辑需要 PydanticAI 的类型安全 Agent 定义和工具模式
• 团队成员需求各异：有些 Agent 适合使用 PydanticAI 的类型系统，有些则需要 LangGraph 的底层控制能力
• 完整示例请参阅 references/hybrid-pydanticai-langgraph.md。

LangGraph 技能： skill_view(name='langgraph') — 涵盖 supervisor/swarm/分层模式、持久化、生产环境部署和评估。

错误处理快速选择

from pydantic_ai import UnexpectedModelBehavior, capture_run_messages

with capture_run_messages() as messages:
    try:
        result = agent.run_sync('Query')
    except UnexpectedModelBehavior as e:
        cause = e.__cause__  # Often ModelRetry('reason')
        print(f"Root cause: {cause}")
        print("Full conversation:", messages)  # Inspect every message
        # Common recovery: raise ModelRetry from tools with clear instructions

异常	含义	恢复方法
UnexpectedModelBehavior	超出重试次数上限，或模型返回了意外响应	检查 e.__cause__，检查消息，调整指令或工具重试设置
ModelRetry（由工具引发）	工具希望模型使用不同的参数重试	让异常继续传播——PydanticAI 会在达到 retries 上限前自动处理
ModelAPIError	提供商返回了 4xx/5xx	检查 API 密钥、速率限制和模型可用性
UsageLimitExceeded	Token／请求预算已用尽	增加 UsageLimits 或优化提示词
HookTimeoutError	生命周期钩子超时	增加钩子超时时间或优化钩子逻辑

主要 CLI 命令

pip install pydantic-ai                    # Everything
pip install "pydantic-ai-slim[openai]"      # Minimal
pip install "pydantic-ai-slim[openai,google,anthropic]"  # Multi-provider

目录结构

pydanticai/
├── SKILL.md
├── references/
│   ├── core-agents.md         # Agent lifecycle, tools, deps, output
│   ├── capabilities-hooks.md  # Capabilities system & lifecycle hooks
│   ├── graph.md               # PydanticGraph (BaseNode + GraphBuilder)
│   ├── models-output.md       # Models, streaming, structured output
│   ├── patterns.md            # Multi-agent patterns & integrations
│   ├── testing-evals.md       # Testing & evaluation framework
│   ├── examples.md            # Complete worked examples
│   ├── hybrid-pydanticai-langgraph.md  # PydanticAI as LangGraph node (hybrid pattern)
│   └── api-reference.md       # Quick API surface reference

注意事项

• 输出类型 = 仅限最终响应： output_type 限制的是最终响应。模型在运行过程中仍可以调用工具（函数工具）。输出函数则不同，它们会被强制调用，并结束本次运行。

• pydantic-graph 不依赖 pydantic-ai： 它是一个独立的库。你可以将它用于非 GenAI 状态机。使用 pip install pydantic-graph 安装。

• pydantic-ai-slim 与 pydantic-ai： slim 软件包仅包含核心依赖项和 OpenTelemetry。完整的 pydantic-ai 是一个元软件包，额外包含 openai、anthropic、google、cli、mcp、evals、web、retries 和 logfire 扩展。

• 默认情况下，流式传输期间的工具调用不会执行： run_stream() 会在遇到第一个符合输出类型的输出时停止。使用 run_stream_events()，或使用带有 event_stream_handler 的 run()，以便继续执行工具调用。

• 系统提示词 ≠ 指令： 系统提示词属于消息历史的一部分，会在往返过程中保留。指令则位于服务端，不会出现在发送给客户端的消息中。重用 message_history 时，除非添加 ReinjectSystemPrompt，否则 Agent 的新系统提示词不会自动发送。

• conversation_id 需手动设置以创建分支： 传入 conversation_id='new'，即可从现有历史记录开始新的对话链。此操作不会自动完成。

• 模型名称采用 provider:model_name 格式 —— PydanticAI 会根据字符串前缀自动解析模型类。对于自定义端点，请使用 OpenAIChatModel(model_name, provider=OpenAIProvider(base_url=...))。

• TestModel 无法模拟原生工具： 如果 Agent 使用了 WebSearch 等工具，请在测试中使用 agent.override(model=TestModel(), native_tools=[]) 覆盖设置。

• 在模块级别声明可测试的 Agent 时使用 defer_model_check=True： 在模块级别（函数外）声明 Agent，并在测试中使用 TestModel 和 agent.override(model=TestModel()) 时，请在构造函数中设置 defer_model_check=True。否则，Agent 会在导入时尝试解析模型字符串；即使在任何测试运行前已覆盖实际模型，如果没有 API 凭据，解析仍会失败。

• 消息历史中的配对关系必须保持完整： 对历史记录进行切片时，工具调用及其返回结果必须保持配对，否则 LLM 会报错。

• stream_text() 不适用于 BaseModel 输出类型： 当 output_type 为 BaseModel（结构化输出）时，调用 result.stream_text() 会引发 UserError('stream_text() can only be used with text responses')。请改用 result.stream_output()，以获取流式传入的部分已验证对象。如果需要对结构化输出进行文本级流式传输，请使用 run_stream_events()，并检查包含 TextPartDelta 增量的 PartDeltaEvent。这两种方法适用于不同的输出模式：文本输出 → stream_text()，结构化输出 → stream_output()。

• graph.run() 返回的是 OutputT，而不是状态对象： 即使向 graph.run() 传入 state=MyState()，返回值仍是图的 output_type（例如 list[int]），而不是状态。执行过程中，state 对象会被原地修改（因为它是可变数据类），因此请单独保留一个引用：

  state = MyState(items_processed=0)
  result = await graph.run(state=state, inputs=[1, 2, 3])
  # result -> [2, 4, 6]       (OutputT = list[int])
  # state.items_processed -> 3 (state mutated in-place)
  

  这种陷阱最常见于并行 .map() 模式，读者可能会以为 result.items_processed 可以正常使用。实际上不行。items_processed 计数保存在你传入的状态对象上，而不在返回值中。

非聊天应用提案

当应用调用 Agent 来填充可编辑的拟议操作时，请阅读类型化应用提案。将类型化数据返回到应用拥有的状态中；将命令权限保留在模型之外。交互行为请参考 product-design-and-ux，架构请参考 software-architecture，部署位置请参考 harness-engineering，评估请参考 agent-evals-and-observability，生产环境权限与恢复请参考 agent-production-operations。

对于非聊天提案，请根据应用提供的范围验证候选 ID。将审批身份和命令保留在模型输出之外。执行时，应用必须重新检查当前权限、版本、策略和确切的已批准操作；仅在审批时检查角色并不足够。通过已持久化的 ID 核对状态不确定的命令，再重试。