AI Research 技术调研

Pydantic AI：类型化状态机驱动 agent loop + capability/hook 中间件的轻量 Python agent 框架

18分钟阅读

Pydantic AI

一句话定位

Pydantic AI 是 Pydantic 团队（Pydantic Validation / Logfire 同一团队，Samuel Colvin）出的 provider-agnostic Python agent 框架。它的范式是「lightweight 强 primitive + 强类型 + capability/hook 中间件」，不是 batteries-included harness：核心只 ship agent loop、各家 model provider 抽象、capabilities/hooks 扩展点，以及少量 capability；memory、上下文压缩、guardrails、文件系统访问、code execution 沙箱、multi-agent 编排都放在独立包 pydantic-ai-harness（本次未克隆）和第三方生态里。两个最独特的设计：(a) agent loop 是由 pydantic_graph 类型化状态机驱动的三节点图（可 agent.iter() 手动单步驱动），而非手写 while 循环；(b) capability 是带 30+ 生命周期 hook 的中间件链，是运行时能力扩展的统一入口。

读维度时的关键甄别：本 repo 内很多「重」能力是「抽象/扩展点齐备、具体实现不在此」。凡下文标「在 harness 包 / 第三方」的，都不等于「未实现」；证据见 docs/harness/overview.md L11-20。

核心架构总览（目录结构关键路径 + 引用的 commit）

分析基于 commit 8432869b2cd59f9f09aaa6fd8ed1c162b015638e（2026-07-10，fix: mark non-crypto hashes with usedforsecurity=False (FIPS precaution) #5053），git clone --depth 1 于 2026-07-11。

uv workspace 多包结构（根 AGENTS.md）：

  • pydantic_ai_slim/ — 核心 Agent 类 + agent loop + 各 provider 的 Model 类（本次主读对象）
  • pydantic_graph/ — 基于类型提示的图库，驱动 agent loop 的状态机
  • pydantic_evals/ — 离线 eval 框架
  • clai/ — CLI（可选 web UI）

核心包内关键路径（相对 repo 根）：

  • pydantic_ai_slim/pydantic_ai/_agent_graph.py（约 1795 行）— agent loop 核心，三节点状态机
  • pydantic_ai_slim/pydantic_ai/capabilities/abstract.py（约 915 行）— capability/中间件 hook 抽象，30+ 生命周期 hook
  • pydantic_ai_slim/pydantic_ai/tool_manager.py — ToolManager，dispatch / retry / deferred 消解
  • pydantic_ai_slim/pydantic_ai/tools.py — Tool / ToolDefinition
  • pydantic_ai_slim/pydantic_ai/toolsets/ — 工具集合抽象与各 WrapperToolset
  • pydantic_ai_slim/pydantic_ai/_system_prompt.py + _instructions.py — prompt / instructions 组装
  • pydantic_ai_slim/pydantic_ai/_instrumentation.py — OTel GenAI semconv
  • pydantic_ai_slim/pydantic_ai/models/ + providers/ + profiles/ — 模型协同三层
  • pydantic_ai_slim/pydantic_ai/durable_exec/ — Temporal / DBOS / Prefect 可恢复执行

「核心 vs harness 分包」是理解全局的前提：docs/harness/overview.md（L11-20）明确把 memory / context 压缩 / guardrails / 文件系统 / code execution 沙箱 / multi-agent 编排划到独立包 pydantic-ai-harness。

Agent Loop（主循环 / 何时继续何时停）

agent loop 由 pydantic_graph 类型化状态机驱动，不是手写 while 循环。三个节点在 _agent_graph.py：

  • UserPromptNode（约 L274-476）：处理 user prompt + system prompt + instructions，构建首个 ModelRequest；也处理 resume / deferred-tool-results 分支；_reevaluate_dynamic_prompts（L443）每轮重算 dynamic system prompt。
  • ModelRequestNode（约 L607-1121）：向 model 发一次请求。_make_request（L835）非流式、stream（L633）流式。请求经 root_capability.wrap_model_request 中间件链包裹（L713 / L876），响应后 _finish_handling 恒定转到 CallToolsNode。
  • CallToolsNode（约 L1124-1437）：决定继续还是停。核心分支在 _run_stream（L1172）：
    • 有 ToolCallPart → _handle_tool_calls（L1322）执行工具、收集 output_parts；若产出 final output（output tool 命中）→ End，否则组装新 ModelRequest 回到 ModelRequestNode（这就是 loop 回边，L1376）。
    • 无 tool call：按 output_schema 决定——有 text_processor 且有文本 → _handle_text_response → End；允许 image → 处理图；否则发 RetryPromptPart 让模型重试（L1305）。

图定义 build_agent_graph（L1608）：start → UserPromptNode → (ModelRequestNode ↔ CallToolsNode) → End。

停止条件：output tool 成功 / text 输出满足 schema / image 输出 / 空或 thinking-only 且 output_schema.allows_none（L1215）。

EndStrategy（_agent_graph.py L71-88，Literal['early','graceful','exhaustive']）控制 output tool 与 function tool 并存时何时结束：early 遇到第一个 output tool 就停；graceful（v2 起默认，L88 注明默认从 v1 的 early 改为 graceful）先跑完排在 output tool 前面的 function tool；exhaustive 全部工具都跑、第一个有效 output 按发出顺序成为结果。

循环预算：GraphAgentState.run_step（L146）计步；UsageLimits 提供 request_limit / total_tokens_limit / tool_calls_limit，在 _prepare_request 里 check_before_request（L1039）。输出重试预算 max_output_retries，consume_output_retry（L188）。State GraphAgentState（L138）含 message_history / usage / run_step / run_id / conversation_id。

因为 loop 是显式图，用户可 agent.iter() 拿到节点流手动单步驱动 —— 这是相对多数 harness 「黑盒 while 循环」的差异点。

记忆与上下文管理（压缩、长期记忆、会话持久化）

  • 会话标识：conversation_id（L147）+ run_id。resolve_conversation_id（L111）优先级：显式 'new' → 新 UUID7 / 显式串 / history 里最后一个非空 conversation_id / 否则新 UUID7。
  • 会话持久化：消息历史用 ModelMessagesTypeAdapter（Pydantic）可序列化到 DB/JSON（docs/message-history.md L269+），跨 run 传 message_history 即恢复上下文。框架本身不带存储后端 —— 是「历史可序列化 + 你自己存」。
  • 上下文压缩 / summarize：核心机制 = ProcessHistory capability（capabilities/process_history.py），在 before_model_request hook 里对 request_context.messages 跑用户提供的 history processor 函数。docs/message-history.md L709-822 给出「用便宜模型 summarize 最老 N 条」示例 ProcessHistory(summarize_old_messages)，多个 processor 按注册顺序链式。文档明确警告：summarize 时须保证 tool call / tool return 配对，否则报错。
  • provider-native compaction：messages.py 有 CompactionPart（_agent_graph.py L1261 处理），docs/harness/overview.md 说 compaction「via OpenAI 或 Anthropic APIs」是需模型支持的 capability；reinject_system_prompt capability 处理压缩管线丢失 system prompt 后的重注入。
  • 长期记忆：核心 repo 未实现。docs/harness/overview.md L16 明确 memory 在 pydantic-ai-harness 独立包；docs/capabilities.md L12 把「building a memory system」列为 capability 的典型用途，但需自建或用 harness 包。

工具体系（定义 / 调用协议 / 注册 / 权限）

  • 定义：tools.py Tool（约 L448）+ ToolDefinition（约 L699）。装饰器 @agent.tool（带 RunContext）/ @agent.tool_plain（不带）。函数签名 → JSON schema 由 _function_schema.py + function_signature.py 自动生成；docstring 用 griffe 解析（_griffe.py）填 description / 参数说明；GenerateToolJsonSchema（tools.py 约 L425）定制 schema。
  • ToolDefinition 关键字段：parameters_json_schema、kind（'function' / 'output' / 'external' / 'unapproved'，约 L742）、strict（OpenAI/Anthropic 严格 schema）、sequential（并发屏障）、timeout、defer_loading（工具搜索延迟暴露）、return_schema / include_return_schema、metadata、capability_id（归属 capability）、tool_kind（跨 provider typed part 判别，如 'tool-search'）、unless_native / with_native（本地工具 vs provider 原生工具的 fall-up）。
  • 注册 / 组织：toolsets/ 是核心抽象 —— 工具集合带生命周期 / instructions / 执行边界。组合器有 CombinedToolset，wrapper 系有 FilteredToolset / PrefixedToolset / RenamedToolset / PreparedToolset / ApprovalRequiredToolset / DeferredLoadingToolset；跨切面行为一律用 WrapperToolset 组合（AGENTS.md 相关规则）。MCP 工具经 mcp.py（约 70KB）接入。
  • 调用协议 / dispatch：ToolManager（tool_manager.py 约 L77）。for_run_step（L111）每步解析动态 toolset 并检查命名冲突。handle_call（L766）= validate_tool_call → execute_tool_call →（若 CallDeferred / ApprovalRequired 则 _resolve_single_deferred）。多工具并行执行（parallel_execution_mode，L96；process_tool_calls 在 _tool_execution.py）。retry：_check_max_retries（L172），ModelRetry → ToolRetryError → RetryPromptPart 回喂模型。
  • 权限：见「安全与权限」。

Prompt 设计（系统提示结构、动态组装）

两条并行通道：system_prompt 与 instructions。

  • system_prompt（_system_prompt.py）：静态串 + 函数（@agent.system_prompt，可 dynamic=True）。SystemPromptRunner（L15）自动探测函数是否 takes_ctx / async。dynamic part 带 dynamic_ref，每轮由 UserPromptNode._reevaluate_dynamic_prompts（_agent_graph.py L443）重算。system prompt 只在首个 request 生成（history 非空则假设已含，docs/message-history.md L151）。
  • instructions（_instructions.py）：官方推荐方式（每轮都重发、不进 history 持久部分）。支持 TemplateStr（模板）/ str / 函数 / 序列。resolve_instructions。
  • 动态组装 _get_instructions（_agent_graph.py L479）：base instructions（agent + capability 的 get_instructions()，abstract.py L253）+ toolset 的 get_instructions() 合并，按 InstructionPart.sorted 排序后 join；每步在 _prepare_request（约 L935）拉取（在动态 toolset 经 for_run_step 解析之后）。
  • capability 也能贡献 instructions；deferred capability 的 instructions 在模型调 load_capability 后才作为工具结果返回。

Router / 编排（任务分解、多 agent、子 agent）

docs/multi-agent-applications.md 明列 5 级：

  1. 单 agent。
  2. Agent delegation（核心模式）：agent 在 tool 里调另一个 agent，返回后收回控制。示例（L28-60）在 @agent.tool 里 await other_agent.run(..., usage=ctx.usage)（共享 usage 计费）。
  3. Programmatic hand-off：app 代码顺序调多个 agent，或用 output function 完全交棒。
  4. Graph based：用 pydantic_graph 显式状态机编排多 agent。
  5. Deep Agents：规划 + 文件操作 + 任务委派 + 沙箱执行。

核心无内置「主 router / 子 agent 调度器」 —— agent 无状态、全局设计，委派就是普通工具调用。子 agent capability 在第三方 subagents-pydantic-ai（SubAgentCapability：task / check_task / wait_tasks / list_active_tasks / soft_cancel_task / hard_cancel_task / answer_subagent，支持 sync/async/auto、嵌套、运行时建 agent；docs/capabilities.md 约 L1829）。另有一种内置的「subagent fallback」概念：ImageGeneration / XSearch 等 capability 在模型不支持 native 时用 fallback_model 委派给一个子 agent（capabilities.md 约 L475 / L672）。

Skill / 插件体系

要分清两层含义：

  • (a) 运行时能力扩展 = Capabilities（核心主扩展点）：AbstractCapability（capabilities/abstract.py）。一个 capability 可贡献 instructions / model settings / toolset / native tools / wrapper toolset / 30+ 生命周期 hook。内置有 WebSearch / WebFetch / Thinking / MCP / ImageGeneration / XSearch / ProcessHistory / ReinjectSystemPrompt / Hooks / IncludeToolReturnSchemas 等（capabilities/ 目录）。组合用 CombinedCapability，按中间件语义拓扑排序（CapabilityOrdering：position outermost/innermost + wraps/wrapped_by + requires，abstract.py 约 L98-141）。
  • On-demand capabilities / Skills（渐进披露）：Capability 类 + defer_loading=True（abstract.py 约 L183）。模型先看到 catalog（id + description），调保留工具名 load_capability 才把该 bundle 的 instructions + tools + model settings + hooks 一起激活（capabilities.md L55-130）。这显式对标 Anthropic Agent Skills（capabilities.md L93 原文「same idea generalised」），但更强：一个 Anthropic skill 是 markdown，而 on-demand capability 额外带 typed 工具 / model settings / hooks；可从 markdown + YAML frontmatter 加载（load_skill(path) → Capability，capabilities.md 约 L409-453）。loaded 状态跨 run 可恢复：存在 message history 的 load_capability call/return 对里（不在 agent 对象上），因此要求稳定显式 id，从而可跨 run、跨 provider 恢复。
  • (b) coding-agent skills（docs/coding-agent-skills.md）：给写 pydantic-ai 应用的编码 agent（Claude Code / Cursor 等）装的框架知识 skill（pydantic/skills 仓库），与运行时无关，别混淆。

自进化能力（自我改进 / 学习型记忆 / eval 驱动纠错）

未实现自我进化 / 学习型记忆 / trajectory 反哺。框架不改自身权重或 prompt。最接近的是运行内纠错闭环：ModelRetry（工具或输出校验失败）→ RetryPromptPart 回喂模型让它改（tool_manager retry + _agent_graph.py L1305 / L889）；capability 的 on_*_error hook 可把错误转成 retry。这是 in-run 反馈，不是跨 run 学习。独立 pydantic_evals/ 包能评估含 LLM/agent 的任意函数，但它是离线评估工具，不自动回灌到 agent 行为里。该维度整体判为「未实现 / 不适用」。

可观测性（日志 / trace 格式）

一等公民是 OpenTelemetry GenAI semantic conventions。_instrumentation.py（约 21KB）+ models/instrumented.py（InstrumentedModel / InstrumentationSettings）。属性用标准 key：gen_ai.system / gen_ai.request.model / gen_ai.provider.name / gen_ai.agent.name（baggage）/ gen_ai.conversation.id / gen_ai.agent.call.id（_instrumentation.py 约 L30-60）。DEFAULT_INSTRUMENTATION_VERSION = 5。记录 TTFT（gen_ai.client.operation.time_to_first_chunk，_agent_graph.py 约 L683-701）。AGENTS.md 有规则要求 _otel_*.py 只实现 spec 定义的东西、不夹私货以防 spec drift；_otel_messages.py 把消息转成 OTel event。官方托管方案是 Logfire（Pydantic 自家，docs/logfire.md），有 Logfire MCP server 可查 agent run / tool call / model request。trace_include_content（RunContext 约 L55）控制内容是否进 trace（隐私）；另有 Instrumentation capability（capabilities/instrumentation.py）。

安全与权限（审批门、密钥管理）

  • 人在环审批（HITL）：ApprovalRequiredToolset（toolsets/approval_required.py）—— call_tool 时若 not ctx.tool_call_approved and approval_required_func(...) 则 raise ApprovalRequired；工具也可自己 raise ApprovalRequired / CallDeferred。
  • deferred tool 机制：ApprovalRequired / CallDeferred 由 ToolManager.handle_call（L766）捕获 → 交给 capability 的 handle_deferred_tool_calls（abstract.py 约 L879，accumulation dispatch，逐 capability 消解）；无 handler 则冒泡为 DeferredToolRequests，由外层 app / 用户批准后用 DeferredToolResults 回填（ToolApproved / ToolDenied，tools.py 约 L327-400）。ToolDefinition 有 kind='unapproved'。文档 docs/deferred-tools.md（约 475 行）。
  • guardrails：核心无内置 guardrail 类，但 capability 抽象是 guardrail 的载体（capabilities.md 约 L1701 给 PII 脱敏示例：在 after_model_request 里改 response）。现成 guardrail 在 harness 包 / 第三方 pydantic-ai-shields（CostTracking / ToolGuard / InputGuard / PromptInjection / PiiDetector / SecretRedaction 等，capabilities.md 约 L1835）。
  • SSRF 防护：_ssrf.py（约 21KB），web fetch 等的服务端请求伪造防护（本次未细读，存在即证）。
  • 密钥管理：provider 从环境变量 / Provider 对象取 API key（providers/ 目录），框架无自建 secret vault。
  • 本次分析的 commit 本身即 FIPS 预防（非加密 hash 标 usedforsecurity=False），可见对安全合规有意识。

沙箱与执行隔离

核心 repo 无沙箱、无子进程执行隔离（grep subprocess 命中的是无关的 async executor 用法，run_in_executor 只是用线程池跑同步工具函数，非隔离）。Code Mode 在 harness 包：docs/harness/code-mode.md —— CodeMode capability 把所有工具包成单个 run_code 工具，模型写 Python（含 loop / 条件 / asyncio.gather）在 Monty 沙箱（github.com/pydantic/monty，Pydantic 自研）里跑，一次 tool call 完成多工具编排、省 round-trip。核心仅提供扩展点：handle_call docstring（L766）提到 wrap_validation_errors=False 供「sandboxed tool dispatch」的嵌套调用者用 —— 说明框架为沙箱化工具分发留了钩子，但沙箱本体未实现于核心。

与模型的协同设计

  • 强 provider-agnostic 三层：models/（各 provider Model 类）+ providers/（认证 / endpoint）+ profiles/（模型能力画像）。AGENTS.md 规则：能力检测用 profile 布尔 flag（如 bedrock_supports_prompt_caching）而非散落 isinstance；provider-specific API 行为放 Provider.model_profile()，模型内在特征放 profile 函数。
  • typed native tools 跨 provider 归一：_narrow_tool_call_parts（_agent_graph.py 约 L1640）把裸 ToolCallPart 按 tool_kind 提升为 typed 子类（如 ToolSearchCallPart）—— adapter 作者只发裸 part，框架管 typed 身份。unless_native / with_native（tools.py 约 L789-818）实现「本地实现 ↔ provider 原生」自动切换（fall-up 模式）。
  • 降级翻译：Model.prepare_messages（约 L1017）/ prepare_request 把 history 翻译成当前 provider 能上线的形状（如不支持 ToolSearchTool 时降级）。FallbackModel 延迟到选定底层模型再定 profile。
  • count_tokens（约 L1036）+ UsageLimits.count_tokens_before_request；thinking/reasoning part 统一抽象（_thinking_part.py，ThinkingPart）。direct.py（约 15KB）提供绕过 agent loop 直接调 model 的低层 API。

轨迹利用（session/trajectory 是否反哺训练/评测）

不反哺训练（无 RL/SFT 数据导出闭环）。session / trajectory = 可序列化的 message history（all_messages() / new_messages()，ModelMessagesTypeAdapter），用途是持久化续聊、跨 Python/JS 交换、喂 evals（docs/message-history.md L269 明说 for evals）。durable_exec/（Temporal / DBOS / Prefect，AGENTS.md 另提 Restate）把 run context / deps / history / retries / model 选择 / toolset 生命周期跨 durable 边界保持 —— 面向「可恢复长运行」而非训练。配合 pydantic_evals 可拿真实 run 轨迹做回归评测（配 pytest-recording / vcrpy 录制回放，AGENTS.md 约 L98），但「评测 → 改 agent」仍是人工。结论：训练反哺未实现；评测利用部分支持（离线、手动）。

与同类 harness 的关键差异（1-3 条）

  1. agent loop = pydantic_graph 类型化状态机（三节点、可 agent.iter() 手动单步驱动），而非黑盒 while 循环。停/继续判定集中在 CallToolsNode._run_stream（L1172），EndStrategy 三档且 v2 默认改 graceful。这是相对 CrewAI / AutoGen / smolagents 等的结构性差异。
  2. capabilities = 30+ 生命周期 hook 的中间件链（before/after/on_*_error 三件套 + wrap_run / wrap_node_run / wrap_model_request / wrap_tool_validate / wrap_tool_execute / wrap_output_validate / wrap_output_process，见 abstract.py L363-856）。运行时能力、guardrail、memory、skill 全部收敛到这一个中间件抽象，是它区别于「插件即工具列表」型 harness 的核心。
  3. 类型安全贯穿全局（AGENTS.md 明列为设计目标）：ToolDefinition 的 JSON schema、structured output（output tool / prompted / native）、RunContext[DepsT] 依赖注入、capability 泛型。同时「核心轻 + 重能力分包（pydantic-ai-harness）+ 第三方」的分层，使核心刻意保持 primitive 而非 batteries-included。

原始源码定位

  • repo: https://github.com/pydantic/pydantic-ai
  • commit/version analyzed: 8432869b2cd59f9f09aaa6fd8ed1c162b015638e（2026-07-10，#5053）
  • 关键文件列表（相对 repo 根）：
    • pydantic_ai_slim/pydantic_ai/_agent_graph.py（agent loop 三节点状态机）
    • pydantic_ai_slim/pydantic_ai/capabilities/abstract.py（capability / 30+ hook 抽象）
    • pydantic_ai_slim/pydantic_ai/capabilities/process_history.py（历史处理 / 压缩 capability）
    • pydantic_ai_slim/pydantic_ai/tool_manager.py（ToolManager：dispatch / retry / deferred）
    • pydantic_ai_slim/pydantic_ai/tools.py（Tool / ToolDefinition）
    • pydantic_ai_slim/pydantic_ai/toolsets/approval_required.py（ApprovalRequiredToolset）
    • pydantic_ai_slim/pydantic_ai/_system_prompt.py + _instructions.py（prompt / instructions 组装）
    • pydantic_ai_slim/pydantic_ai/_run_context.py（RunContext 字段）
    • pydantic_ai_slim/pydantic_ai/_instrumentation.py（OTel GenAI semconv）
    • AGENTS.md（根，项目哲学 / 仓库结构 / 包划分）
    • docs：docs/harness/overview.md、docs/harness/code-mode.md、docs/capabilities.md、docs/coding-agent-skills.md、docs/message-history.md、docs/multi-agent-applications.md、docs/deferred-tools.md

一手源存档（sources/）

存于 /Users/zhao/projects/self-wiki/ai-research/sources/harness/pydantic-ai/：

  • NOTES.md — 第一阶段源码级调研笔记（12 维度 + 行号级证据）
  • src/_agent_graph.py — agent loop 全量（三节点、EndStrategy、停止条件、loop 回边）
  • src/capabilities_abstract.py — capability 抽象与全 hook 清单
  • src/capabilities_process_history.py — ProcessHistory capability
  • src/tool_manager.py — ToolManager dispatch / retry / deferred
  • src/tools.py — Tool / ToolDefinition
  • src/toolsets_approval_required.py — ApprovalRequiredToolset
  • src/_system_prompt.py + src/_instructions.py — prompt / instructions 组装
  • src/_run_context.py — RunContext 字段
  • src/docs_harness_overview.md — 核心 vs harness 分包证据
  • src/docs_harness_code-mode.md — Code Mode / Monty 沙箱
  • src/root_AGENTS.md — 项目哲学 / 仓库结构

关系图谱

反向链接