跳转到内容

PydanticAI 的图状态机 Agent 循环

PydanticAI 的图状态机 Agent 循环:类型驱动的编排

学习目标

补充理解:

  • PydanticAI 如何将 Agent 循环实现为类型安全的有向图(而非 LangGraph 的 StateGraph)
  • 四节点架构(UserPromptNode → ModelRequestNode → CallToolsNode → End)的状态流
  • 流式路径与非流式路径的节点遍历差异

前置知识


1. 与通用 StateGraph 的差异

通用 StateGraph(如 LangGraph)使用 State 字典作为共享状态容器,节点通过读写状态字典的键来通信。PydanticAI 采用类型安全 dataclass 作为状态

@dataclass
class 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 # 用户元数据

关键差异

维度通用 StateGraphPydanticAI
状态类型字典/TypedDictdataclass(强类型)
节点间通信读写状态字典的键节点 run() 返回值决定下一个节点
图定义方式状态图 + 条件边GraphBuilder builder API
类型安全运行时 key 错误可能编译时类型检查

2. 四节点架构详解

PydanticAI 的 Agent 图只有四个节点,但覆盖了所有 Agent 执行路径:

每个节点的责任

  1. UserPromptNode:入口节点。处理系统提示、动态指令、用户提示和延迟工具结果。决定进入 ModelRequestNode 还是直接跳到 CallToolsNode(当已有消息历史时)
  2. ModelRequestNode:发出 LLM 请求。准备工具定义、检查用量限制、通过 capability 链包装模型调用
  3. CallToolsNode:处理模型响应。解析工具调用、执行工具、决定是回环还是结束
  4. SetFinalResult:流式路径的快捷终端节点

3. 依赖注入:GraphAgentDeps

与通用 StateGraph 不同,PydanticAI 将所有运行时依赖打包为一个 GraphAgentDeps dataclass:

@dataclass
class 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) # 包装整个 run
wrap_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 协议

  1. 模型节点打开流,设置 stream_ready 事件
  2. 消费者(用户代码)获取 AgentStream 并开始消费
  3. 当检测到 FinalResultEvent 时,停止流式并调用 process_tool_calls()
  4. 进入 SetFinalResult 节点结束图遍历

这与非流式路径的关键差异在于:流式路径可以提前结束(在模型产生最终输出后),而非流式路径必须等待所有工具调用完成。