PydanticAI 的图状态机 Agent 循环
PydanticAI 的图状态机 Agent 循环:类型驱动的编排
学习目标
补充理解:
- PydanticAI 如何将 Agent 循环实现为类型安全的有向图(而非 LangGraph 的 StateGraph)
- 四节点架构(UserPromptNode → ModelRequestNode → CallToolsNode → End)的状态流
- 流式路径与非流式路径的节点遍历差异
前置知识
- StateGraph Agent 循环 — 将 Agent 循环建模为有向图的通用概念
1. 与通用 StateGraph 的差异
通用 StateGraph(如 LangGraph)使用 State 字典作为共享状态容器,节点通过读写状态字典的键来通信。PydanticAI 采用类型安全 dataclass 作为状态:
@dataclassclass GraphAgentState: message_history: list[ModelMessage] # 消息历史 usage: RunUsage # 用量追踪 output_retries_used: int # 输出重试计数 run_step: int # 当前步数 run_id: str # 运行 UUID conversation_id: str # 会话 UUID metadata: dict | None # 用户元数据关键差异:
| 维度 | 通用 StateGraph | PydanticAI |
|---|---|---|
| 状态类型 | 字典/TypedDict | dataclass(强类型) |
| 节点间通信 | 读写状态字典的键 | 节点 run() 返回值决定下一个节点 |
| 图定义方式 | 状态图 + 条件边 | GraphBuilder builder API |
| 类型安全 | 运行时 key 错误可能 | 编译时类型检查 |
2. 四节点架构详解
PydanticAI 的 Agent 图只有四个节点,但覆盖了所有 Agent 执行路径:
每个节点的责任:
- UserPromptNode:入口节点。处理系统提示、动态指令、用户提示和延迟工具结果。决定进入
ModelRequestNode还是直接跳到CallToolsNode(当已有消息历史时) - ModelRequestNode:发出 LLM 请求。准备工具定义、检查用量限制、通过 capability 链包装模型调用
- CallToolsNode:处理模型响应。解析工具调用、执行工具、决定是回环还是结束
- SetFinalResult:流式路径的快捷终端节点
3. 依赖注入:GraphAgentDeps
与通用 StateGraph 不同,PydanticAI 将所有运行时依赖打包为一个 GraphAgentDeps dataclass:
@dataclassclass GraphAgentDeps: user_deps: DepsT # 用户提供的依赖(泛型) model: Model # LLM 模型 output_schema: OutputSchema # 输出类型 schema root_capability: AbstractCapability # 根 capability 链 tool_manager: ToolManager # 工具管理器 usage_limits: UsageLimits # 用量限制 end_strategy: EndStrategy # 结束策略 # ... 更多这种设计确保了:
- 所有依赖在编译时有类型约束
- 新增依赖只需添加字段,无需修改节点签名
- 图持久化时可以序列化/反序列化整个依赖快照
4. Capability 作为节点中间件
PydanticAI 的 capability 系统为 Agent 循环提供了中间件链。每个 capability 可以钩入不同的生命周期:
wrap_run(handler) # 包装整个 runwrap_node_run(node, handler) # 包装单个节点执行wrap_model_request(handler) # 包装模型调用wrap_tool_validate(handler) # 包装工具验证wrap_tool_execute(handler) # 包装工具执行这与通用 StateGraph 的中间件(如 LangGraph 的 Middleware)类似,但 PydanticAI 的 capability 提供了更细粒度的钩子——可以在工具验证阶段拦截,也可以在模型请求阶段拦截,还可以在节点级别拦截。
5. 流式路径的特殊处理
PydanticAI 的流式路径(agent.run_stream())使用cooperative hand-off 协议:
- 模型节点打开流,设置
stream_ready事件 - 消费者(用户代码)获取
AgentStream并开始消费 - 当检测到
FinalResultEvent时,停止流式并调用process_tool_calls() - 进入
SetFinalResult节点结束图遍历
这与非流式路径的关键差异在于:流式路径可以提前结束(在模型产生最终输出后),而非流式路径必须等待所有工具调用完成。