Checkpoint 系统与 Durable Execution

Checkpoint 系统与 Durable Execution

学习目标

本章要解决什么问题：

• 如何使用 Checkpoint 系统实现 Agent 的持久化执行
• 如何从任意检查点恢复或 fork 新的执行分支
• 如何选择合适的 Checkpoint 后端（内存、SQLite、PostgreSQL）
• Durable Execution 如何在故障后自动恢复

前置知识

本章涉及 Agent 状态序列化的通用原理，建议先阅读：

• Agent 状态序列化与持久化 — 状态与配置的区别、Pydantic 序列化策略

下文假设你已理解上述概念，直接聚焦 LangGraph 的具体实现。

---

项目实践

Checkpoint 的接入方式

Checkpoint 通过 checkpointer 参数注入到编译后的图：

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "session-1"}}
result = app.invoke({"messages": ["你好"]}, config)

关键：每次 invoke() 必须传入包含 thread_id 的 config，否则 checkpoint 无法关联。

三种 Checkpoint 后端

后端	类	适用场景
内存	InMemorySaver	开发/测试，进程重启后丢失
SQLite	SqliteSaver	单机部署，轻量级持久化
PostgreSQL	PostgresSaver	生产部署，高并发、可扩展

切换后端无需修改业务代码，只需更换 checkpointer 实例：

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("checkpoints.db")
checkpointer = SqliteSaver(conn)

时间旅行：从历史快照恢复

Checkpoint 系统为每次状态变更生成唯一 checkpoint_id，可以回退到任意历史点：

# 获取历史 checkpoint 列表
history = app.get_state_history(config)
for checkpoint in history:
    print(checkpoint["id"], checkpoint["created_at"])

# 从特定 checkpoint 恢复
app.invoke(None, {
    "configurable": {
        "thread_id": "session-1",
        "checkpoint_id": "1f0e2d3c-..."
    }
})

这支持”分支”操作：从历史点 fork 新的执行路径，原路径保持不变。

Durable Execution：故障自动恢复

Durable Execution 指 Agent 在故障后可以从最后一个 checkpoint 自动恢复执行，而非从头开始。

关键机制：

• 每个节点执行前后自动 checkpoint
• 中间结果写入 checkpoint 的 pending_writes 字段
• 恢复时跳过已成功节点，从失败点重试

问题与规避

状态字段必须可序列化

问题：Checkpoint 需要序列化整个状态，如果状态中包含不可序列化对象（如数据库连接、文件句柄），保存会失败。

规避：

• 使用 UntrackedValue Channel 标记不可序列化的字段
• 使用 EphemeralValue 标记不需要持久化的字段
• 将外部连接放入 context_schema 而非 state_schema

大状态导致存储膨胀

问题：每次 checkpoint 保存完整状态，如果状态中包含大量数据（如长对话历史、文件内容），checkpoint 体积会快速增长。

规避：

• 定期清理状态中的过期数据
• 使用 PostgreSQL 后端，利用 JSONB 压缩
• 考虑只保存状态引用而非完整数据

thread_id 混淆

问题：使用错误的 thread_id 会加载到另一个会话的 checkpoint。

规避：在业务逻辑中明确管理 thread_id 的生命周期，避免复用已结束会话的 ID。

设计取舍

优势

• 可插拔：三种后端适配不同部署场景
• 时间旅行：完整的执行历史可回溯
• 故障恢复：自动从失败点恢复，不丢失已完成工作

代价

• 序列化开销：每次 checkpoint 需要序列化完整状态
• 存储成本：长期运行的 Agent 会产生大量 checkpoint 数据
• 一致性风险：并发写入同一 thread_id 可能导致 checkpoint 冲突

替代方案

• 应用层状态管理：自行管理状态保存和恢复，但需要处理序列化、版本迁移等细节
• 外部工作流引擎：如 Temporal、Camunda，提供更强的保证但引入额外依赖

参考来源

• 源码验证: libs/checkpoint/ — 基础接口
• 源码验证: libs/checkpoint-sqlite/ — SQLite 实现
• 源码验证: libs/checkpoint-postgres/ — PostgreSQL 实现
• 文档: LangGraph Persistence, https://docs.langchain.com/oss/python/langgraph/persistence