企业官网建设流程全解析

用 Pydantic AI Harness 的 Step Persistence 能力，只需几行代码就能把 AI Agent 的会话变成"可保存、可恢复、可分叉"：进程崩溃后从最近的检查点继续运行（resume），也能从任意快照点 fork 出一条新分支。它是基于 Pydantic AI 构建的 Agent 持久化底座，专为多轮对话、多进程协作和故障恢复场景设计。

🧭 为什么 AI Agent 会话需要"存档点"

跑过 Agent 的人都遇到过这些场景：

• 🔥进程崩溃：Agent 跑到一半被 kill，上下文全丢，只能从头再来；
• 🧵跨进程续聊：想在新进程里接着上一轮的调查继续问，却不知道从哪里"接"；
• 🌿分支实验：希望从某个中间状态分叉出一条新路径，而不是覆盖原对话；
• 🔍审计追溯：想知道 Agent 到底调了哪些工具、每步花了多久、副作用是否真的发生。

Step Persistence 就是为解决这些问题而生的：它把 Agent 每次run的每一步都记录成事件与快照，让你随时**恢复（continue）或分叉（fork）**运行。

⚡ Step Persistence 的四大能力

上这类轨迹（上图是一次典型 Agent run 的模型请求与工具调用时序）背后的每一步，Step Persistence 都会落盘。它提供四类核心能力：

1. 追加式步骤事件（StepEvent）：run 开始/结束、模型请求、工具调用、失败等每个边界都会追加一条事件记录，即使进程在工具调用中途被杀死，也留有完整的事件线索；
2. 可续接快照（ContinuableSnapshot）：在工具调用全部完成的稳定边界保存快照。每个快照带state：complete（工具结果齐全）或interrupted（崩溃时留下的未完结工具调用）；
3. 工具副作用账本（ToolEffectRecord）：按(run_id, tool_call_id)记录每个工具的生命周期（started/completed/failed），崩溃后能判断"副作用到底发生没有"；
4. 谱系元数据：conversation_id（多轮序列）与parent_run_id（父-子 Agent 层级）两个独立维度，让"一次对话"和"一次委托调用"都能精确查询。

💡 它记录的是"消息级检查点"，不是图状态（graph-state）检查点——能力内部状态、工作区快照的恢复不在当前范围内。

🚀 三步开启 Step Persistence

先安装（Mongo 后端需额外安装[mongodb]扩展）：

uv add pydantic-ai-harness # 或使用 pip：pip install pydantic-ai-harness

最小接入只有几行，核心就是把StepPersistence挂到 Agent 的 capabilities 上：

from pydantic_ai import Agent from pydantic_ai_harness import StepPersistence from pydantic_ai_harness.step_persistence import InMemoryStepStore store = InMemoryStepStore() agent = Agent( 'openai:gpt-5', capabilities=[StepPersistence(store=store, agent_name='librarian')], ) await agent.run('帮我调查一个符号的定义', conversation_id='conv-1')

run_id天然对应每次Agent.run调用；多轮对话的分组请用conversation_id=，这样同一对话的所有轮次都能按组查询。

💾 四种存储后端：按规模选择

所有后端实现同一套异步StepStore协议，可随时替换：

长对话的快照默认不裁剪，可用max_snapshots_per_run限制单 run 的快照数量，防止存储膨胀：

store = FileStepStore('runs', max_snapshots_per_run=8)

后端实现位于 pydantic_ai_harness/step_persistence/_store.py。

🔄 会话恢复：continue_run 一键续接

快照里保存的正是上图这类完整消息历史（输入、回复、工具调用与参数）。恢复会话的核心是continue_run：它读取该 run 最新的complete快照消息，直接回传给Agent.run(message_history=...)：

from pydantic_ai_harness.step_persistence import continue_run # 找到对话的最后一轮 run prior_run = (await store.list_runs(conversation_id='conv-1'))[-1].run_id history = await continue_run(store, run_id=prior_run) # 换个进程也没关系，接着聊 await agent.run('继续：检查一下这个路径', message_history=history, conversation_id='conv-1')

默认只返回"安全"的complete快照；若确认崩溃点可安全续接，可传include_interrupted=True。

🌿 分叉运行：fork_run 开辟新分支

fork_run(store, run_id=...)返回与continue_run相同形状的消息历史，但语义是从该快照点开辟一条新分支——新 run 会拿到新的run_id（通常还有新的conversation_id），不会污染原对话：

from pydantic_ai_harness.step_persistence import fork_run history = await fork_run(store, run_id='libr-3f2a') # 传给一个全新的 delegate run：新的 agent_name / conversation_id

典型用法：调查中途发现方向不对，从某个快照点分叉重查；或主对话继续推进，同时另开一条分支做实验。

🛟 崩溃恢复：工具副作用账本是关键

进程中途被 kill 时，"工具到底执行了没有"是个经典难题。Step Persistence 的答案：

• store.list_unresolved_tool_effects(run_id=...)列出所有只有started记录、没有终态的工具——这类工具标记为unknown_after_crash：副作用可能已发生，也可能没有；
• inspect_recovery(store=..., run_id=...)（位于 pydantic_ai_harness/step_persistence/recovery.py）一次性返回最新快照、已结算快照与未决效果清单，便于决策是continue_run还是fork_run；
• 有外部副作用的工具（改 issue 标签、发 PR 等）建议在执行前调用annotate_tool_effect(store, ctx, idempotency_key=..., effect_summary=...)，把幂等键和效果摘要写进账本，恢复时即可判断能否安全重放。

⚠️ 注意：回到更早的complete快照不会撤销已发生的外部副作用，去重决策由你的编排层负责。

📊 检查运行树与 SnapshotSaved 通知

list_runs支持conversation_id、parent_run_id等过滤器组合查询，结果按开始时间升序排列：

# 某个编排器 run 下的所有 delegate 子运行 delegates = await store.list_runs(parent_run_id='orch-3f2a') # 一次对话的全部轮次 turns = await store.list_runs(conversation_id='conv-abc')

子 Agent 委托在进程内调用时，parent_run_id会通过ContextVar自动推断填充，无需手动传参；跨进程场景可显式设置parent_run_id=。

每次检查点写入完成后，会发出SnapshotSaved类型化能力事件（携带persistence_run_id、conversation_id、step_index、state），订阅它就能实时感知快照落盘时机。

📏 边界说明：它不做什么

• ❌ 不恢复能力内部状态、图节点状态、重试计数器或进行中的流式响应；
• ❌ 不自动去重重放的外部副作用（需配合annotate_tool_effect自行决策）；
• ❌ 默认不裁剪事件与快照，也不回收已外化的媒体文件；
• ❌ 不发出 OpenTelemetry spans（pydantic_ai 的Instrumentation能力已覆盖 agent run / chat / running tool 的追踪）。

📚 参考资料

📌 项目处于 0.x 版本，API 可能在次要版本间调整；届时弃用警告与迁移指南会说明升级方法。

掌握 Step Persistence 后，你的 AI Agent 就从"一次性进程"升级为有存档、可续命、能分叉的持久会话系统——无论是崩溃恢复还是多 Agent 协作编排，都多了一手底牌。