Agent 开发之 Checkpoint：AI 工作流的状态恢复机制

开发 Agent 的人大多体会过一种痛改了一行 prompt就要把整个流程从头跑一遍。LLM 调用、工具调用、推理链——全重来。如果你的 Agent 有 10 步、每步调一次 GPT-5一次调试的 token 账单一拉就是六位数。更糟的是Agent 不像普通程序那样崩溃即退出。它可能在第 7 步卡住、在第 4 步给出错误的中间结论然后继续往下走、在第 9 步耗尽上下文窗口。没有 checkpoint 时你只能从零重新启动祈祷这次能走得更远。一旦给 Agent 加上 checkpoint你就能从失败的那一步接着跑——不是重放历史是真正地从状态快照恢复执行。Checkpoint ≠ 代码断点。在 LangGraph 里你用interrupt()暂停了一个节点的执行人类审核后调用Command(resumeTrue)——Agent 不是从interrupt()的那一行继续而是重新执行整个节点。这个设计意味着 checkpoint 记录的永远是super-step 结束时的完整状态而不是某行代码执行到一半的寄存器快照。理解这一点有助于理解 Pregel 模型、channel 版本号、pending writes 这些概念。痛点没有 checkpointAgent 调试是一件很烧钱的事没有 checkpoint 的 Agent 开发流程是这样写代码 → 跑 Agent → 第 N 步出问题 → 修代码 → 从第 1 步重新跑 → 又出问题 → 再修再跑。每一步重跑都在消耗 LLM token。上一次跑通过的步骤这一次可能因为模型非确定性走了不同的工具调用路径产出的中间结果也不同。你修的是一行 prompt但行为变化可能出现在完全不相干的步骤——因为模型的推理链重组了。LangGraph 社区里一个常见的场景调 Agent 的工具选择逻辑时每次都要等前面 6 步的 RAG 检索、意图识别、工具调度跑完才能验证第 7 步的改动是否生效。这 6 步消耗的 token 是纯浪费。checkpoint 解决的就是这个问题——让 Agent 的状态可持久化、可恢复、可回溯。但不是靠记录每一行代码执行到哪里而是靠一种来自大规模图计算领域的设计BSPBulk Synchronous Parallel批量同步并行模型。原理checkpoint 的数学基础不在 LangGraph在 Google 2010 年的一篇论文LangGraph 的 checkpoint 机制不是原创——它继承自 PregelGoogle 在 2010 年 SIGMOD 上发表的 BSP 图计算系统。Pregel 的 super-step 三阶段Pregel 把图计算分成若干个super-step。每个 super-step 严格遵循三个阶段Plan规划根据当前 channel 的版本号变化计算下一轮需要执行哪些节点Execute执行并行运行所有被标记的节点每个节点读取上游 channel 的数据、产出写给下游 channelUpdate更新把所有节点的输出写入 channel递增版本号生成 checkpointsuper-step 之间的边界是天然的 checkpoint 点。因为所有节点都完成了本轮计算、状态一致、没有执行到一半的节点存在。Pregel 的设计哲学是在 step 之间做快照不在 step 内部做快照——这也是为什么 LangGraph 的interrupt()不会产生一个半行代码级的断点。LangGraph 的运行时核心PregelLoop完整实现了这个三阶段循环prepare_next_tasks → execute → apply_writes → create_checkpoint → 循环每一次循环都是一个 super-step。checkpoint 记录的是apply_writes完成后的完整图状态。Checkpoint 的数据模型一个 checkpoint 在 LangGraph 内部是一个TypedDict六个字段vintcheckpoint 格式版本号用于向后兼容idstrcheckpoint 的唯一标识ULID 格式按时间排序tsstrISO 8601 时间戳channel_valuesdict所有 channel 的当前值是图状态的完整快照channel_versionsdict每个 channel 的版本号格式为{channel_name: version_int}versions_seendict每个节点上次执行时看到的各 channel 版本号格式为{node_name: {channel_name: version_int}}channel_versions和versions_seen是调度引擎的核心。判断一个节点是否需要在下一个 super-step 执行的规则很简单channel_versions[ch] versions_seen[node][ch] → node 需要运行如果某个节点依赖的 channel 版本号比它上次执行时看到的版本号更高说明这个 channel 在上一个 super-step 被别的节点更新了——当前节点需要重新计算。这就是一个分布式的脏标记机制。不依赖中心化的调度器告诉每个节点你该跑了每个节点自己通过版本号比对就能判断。BaseCheckpointSaver 的五个接口LangGraph 通过BaseCheckpointSaver抽象了所有 checkpoint 存储后端。五个接口职责清晰put(config, checkpoint, metadata, new_versions)在 super-step 边界写入完整 checkpoint。new_versions由 PregelLoop 计算——它是本轮所有 channel 的新版本号集合put_writes(config, writes, task_id, task_path)在每个节点执行完成后立即调用写入该节点的中间输出pending writes。它不等 super-step 结束——这是一种容错设计get_tuple(config)按thread_idcheckpoint_id检索一个 checkpoint返回CheckpointTuple包含 checkpoint、metadata、parent_config、pending_writeslist(config, filter, before, limit)列出某个 thread 的所有 checkpoint最新在前。这是 Time Travel 的检索入口delete_thread(thread_id)删除一个 thread 的所有 checkpointput和put_writes的分离是关键设计决策。put在 super-step 结束时调用代表完整快照put_writes在每个节点完成时立即调用代表部分写入。如果 super-step 中途崩溃已经通过put_writes持久化的中间输出可以用于恢复——不需要重跑已经成功的兄弟节点。两张表的存储模型大多数BaseCheckpointSaver的实现SqliteSaver、PostgresSaver使用两张表checkpoints 表列类型说明thread_idTEXT线程标识主键之一checkpoint_nsTEXTcheckpoint 命名空间默认空字符串checkpoint_idTEXTULID按时间排序parent_checkpoint_idTEXT父 checkpoint 的 ID构成链表checkpointJSONB序列化后的 checkpoint 数据大值溢出到独立checkpoint_blobs表该表为 BYTEAmetadataJSONBCheckpointMetadatasource、step、parents 等writes 表列类型说明thread_idTEXT线程标识checkpoint_nsTEXT命名空间checkpoint_idTEXT所属 checkpointtask_idTEXT节点 IDchannelTEXT写入的 channel 名valueBYTEA序列化后的写入值writes 表的存在让部分恢复成为可能。如果 super-step 里有 5 个节点并行执行其中 3 个成功、1 个崩溃、1 个还没开始已有 3 个节点的 writes 持久化——恢复时只需要重跑崩溃和未开始的那两个。一个具体的时间线假设这个简单图START → node_a → node_b → END整个运行过程中会产生4 个 checkpointstep -1source inputnext (__start__,)。这是 PregelLoop 在接受外部输入后创建的第一个 checkpoint——内容就是用户传入的input_data。channel_versions全部初始化为 1step 0source loopnext (node_a,)。输入被应用到 channel 后创建。versions_seen[__start__]记录了输入 channel 的版本号PregelLoop 计算后发现node_a依赖的 channel 版本号高于它见过的版本——所以node_a被放入nextstep 1source loopnext (node_b,)。node_a执行完毕后它的输出被写入 channelchannel_versions递增PregelLoop 发现node_b依赖的 channel 有新版本——node_b进入next。此时的channel_values包含了node_a的完整输出step 2source loopnext ()。node_b执行完毕next为空——图执行结束。这是最终 checkpointchannel_values包含完整结果在这 4 个 checkpoint 之间put_writes被调用 2 次node_a和node_b各一次writes 表里对应有 2 条记录。序列化方面LangGraph 默认使用JsonPlusSerializer基于ormsgpack JSON支持绝大多数 Python 原生类型。生产环境中如果 State 包含敏感数据可以用EncryptedSerializer包装任意序列化器做透明加密。能力checkpoint 不是存了就好它能做什么人工介入interrupt() 和 Command(resume)最常用的能力是Human-in-the-Loop人工介入循环。在节点里调用interrupt()暂停执行等待人类审核后继续# 在 StateGraph 的节点内使用 interrupt def review_node(state: State) - State: # 先做自动化检查 if state[risk_score] 0.8: # 暂停等待人工审批 decision interrupt(风险过高需要人工审批) state[approved] decision return state # 外部恢复 config {configurable: {thread_id: task-123}} graph.invoke(input_data, config) # 在 interrupt() 处暂停 # 人类审核后 graph.invoke(Command(resumeTrue), config) # 恢复执行Command(resumeTrue)恢复后review_node会从头重新执行而不是从interrupt()调用之后的那一行继续。也就是说interrupt()之前的所有代码会再跑一遍。这意味着两点。第一interrupt()之前的代码如果有副作用比如发了一条消息、写了一次数据库恢复时会再执行一次需要你自己做幂等处理。第二interrupt()可以用在循环里——每次循环命中interrupt()都会产生一个新的暂停点人类逐一审批。Time Travel回溯和分叉get_state_history返回一个 thread 的所有 checkpoint 列表最新在前。你可以回退到历史状态graph.invoke(None, history[-1].config)——从旧 checkpoint 重新执行分叉新分支graph.update_state(old_config, new_values)——在旧 checkpoint 基础上修改状态值LangGraph 会创建一个source update的新 checkpointparent_checkpoint_id指向旧 checkpoint分叉产生的实际上是一条新的 checkpoint 链。list()方法返回的 checkpoint 列表包含所有分支你可以通过parent_checkpoint_id重建执行树。容错pending writes 的真正价值回到前面提到的那张 writes 表。假设一个 super-step 里并行跑着 3 个节点node_a完成 →put_writes持久化node_b完成 →put_writes持久化node_c执行到一半进程崩溃恢复时PregelLoop 调用get_tuple()拿到最新 checkpoint发现pending_writes里已经有node_a和node_b的输出。它会把这两个输出应用到 channel 上更新versions_seen然后只调度node_c重新执行。已经成功的节点正常情况下不会重跑。不是靠记住哪些节点跑过了是直接靠 writes 表里的记录驱动调度决策。进行完整的故障恢复测试的方法很简单启动 Agent → 在中间步骤kill -9进程 → 用同一个thread_id重新invoke→ 验证 Agent 从崩溃点继续而非重头开始。长期记忆Store API 与 checkpoint 的分工checkpoint 解决的是单线程内的状态持久化——同一个thread_id的完整执行历史。但如果你的 Agent 需要在多个 thread 之间共享信息checkpoint 不是答案。Store API跨线程记忆解决这个问题Agent 可以在一个 thread 里写入记忆store.put在另一个 thread 里通过语义搜索读取store.search。checkpoint 管这次任务跑到了哪里Store 管之前所有任务学到了什么。两者互补不是替代关系。arXiv 2604.28138Crab提出了语义感知的 checkpoint/restore 运行时arXiv 2606.06090 则将记忆管理定义为执行状态管理的一种——两者都指向统一的框架。实践选对后端避开三个坑后端选择没有最佳后端只有匹配部署拓扑的后端InMemorySaver开发调试用。进程重启数据全丢SqliteSaver单机本地持久化。适合个人项目、单进程部署。注意如果跑在容器里但没有挂持久卷效果等同于 InMemorySaverAsyncPostgresSaver生产环境首选。支持连接池、并发读写。多容器部署k8s的默认选择langgraph-redis社区维护低延迟场景。配合 AOF 持久化避免数据丢失。适合对恢复速度要求极高的场景实时对话 Agent自定义 DynamoDB/Cloud StorageServerless 场景。LangGraph 的BaseCheckpointSaver接口足够简单实现一个适配层通常不到 200 行代码关键约束checkpointer实例必须是同一个对象传给compile()和invoke()。如果你在每次请求里新建一个SqliteSaver每次的 checkpoint 都写在内存里——换个请求就丢了。五个实践原则State 保持小而可序列化。State 里不要放数据库连接、文件句柄、大段二进制数据。序列化开销和存储膨胀会在长对话中放大。实在需要传递大对象时State 里只存引用如文件路径、数据库 ID在节点内部按需加载thread_id 使用 UUID 或业务 ID 哈希长度不超过 255 字符。Postgres 的 btree 索引对超长键的性能退化很明显——单键超过约 2704 字节会触发索引溢出同 thread_id 避免并发写入。checkpoint 链本质是单链表并发写会导致parent_checkpoint_id冲突。如果需要并行任务拆成不同 thread_id设置 checkpoint 清理策略。如果不设置清理策略checkpoint 会持续增长。每条消息至少产生一个 super-step一次对话轻松上百个 checkpoint。不清理的话几个月后存储成本和检索延迟都会明显上升。LangGraph 1.2 引入了 DeltaChannel 机制见下文大幅降低了存储压力但仍建议设置保留策略验证故障恢复流程。不是相信它能在崩溃后恢复而是用kill -9测试它真的能恢复。在 State 里嵌入retry_count在interrupt()之前的副作用做好幂等处理三个常见的坑坑一恢复时传了新的 input# ❌ 错误重新传了 input_data graph.invoke(input_data, config) # 从 input checkpoint 重新开始 # ✅ 正确传 None 或 Command graph.invoke(Command(resumeTrue), config) # 从 interrupt 处继续传了新的input_data会触发 PregelLoop 创建一个新的source inputcheckpoint——等同于从头开始。坑二InMemorySaver 上生产# ❌ 开发环境跑通了直接推到生产 graph builder.compile(checkpointerInMemorySaver()) # ✅ 生产环境用持久化后端 from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver graph builder.compile(checkpointerAsyncPostgresSaver(pool))InMemorySaver 在生产环境的表现是每次部署重启后所有对话状态丢失。用了 SqliteSaver 但没挂持久卷Docker/K8s 常见配置失误同理。坑三超长 thread_id# ❌ 把整个对话历史拼进 thread_id thread_id f{user_id}-{session_id}-{timestamp}-{random_suffix}-... # ✅ 用短哈希 import hashlib thread_id hashlib.sha256(f{user_id}:{session_id}.encode()).hexdigest()[:16]DeltaChannelLangGraph 1.2 的增量存储改进传统的 checkpoint 机制在每个 super-step 保存完整的channel_values——一条 500 轮对话的 Agent 会话中第 500 个 checkpoint 存储的是全部 500 轮的状态累积。存储复杂度是 O(N²)。DeltaChannelLangGraph 1.2beta改变了这个逻辑每个 checkpoint 只存储增量——本 super-step 新增的 channel 值。重建完整状态时沿着 checkpoint 链从基快照开始逐增量还原。效果很直接。实测数据2026 年 6 月 LangChain 官方博客DeltaChannel RFC500 轮对话的完整快照模式占用约 4GBDeltaChannel 配合snapshot_frequency50每 50 个增量 checkpoint 保存一个完整快照做锚点降到约 110MB——约 41 倍的存储缩减。DeltaChannel 的 reducerreducer增量合并器在读取时运行而非写入时。这意味着写入 checkpoints 的开销极低——不计算合并只做追加。读取时的合并操作沿着 checkpoint 链向前查找最近的完整快照然后逐增量 apply复杂度 O(增量数量)而非 O(总 state 大小)。