Checkpoint 接口抽象与后端可插拔

Checkpoint 接口抽象与后端可插拔

学习目标

本章要解决什么问题：

• BaseCheckpointSaver 接口定义的核心操作
• Checkpoint 数据结构的字段及其含义
• SQLite 和 PostgreSQL 后端的实现差异
• 序列化格式的选择（msgpack + xxhash）

项目实践

BaseCheckpointSaver 的四个核心操作

class BaseCheckpointSaver(Generic[Value]):
    def put(self, config, checkpoint, metadata):
        """保存一个 checkpoint。"""

    def get(self, config):
        """获取最新或指定 checkpoint_id 的 checkpoint。"""

    def list(self, config, *, filter, before):
        """列出历史 checkpoint。"""

    def put_writes(self, config, writes, task_id):
        """保存中间写入（节点执行过程中）。"""

每个操作接受 config 字典，包含 thread_id 和可选的 checkpoint_id。

Checkpoint 数据结构

class Checkpoint(TypedDict):
    v: int                    # 版本号
    id: str                   # checkpoint 唯一 ID（xxhash）
    ts: str                   # 时间戳
    channel_values: dict      # 各 Channel 的当前值
    channel_versions: dict    # 各 Channel 的版本号
    pending_sends: list       # 待分发的 Send 任务
    pending_writes: list      # 待完成的写入
    tasks: dict               # 当前待执行的任务
    versions: dict            # 版本向量（用于并发控制）

channel_values 是核心：存储每个 Channel 通过 checkpoint() 返回的序列化快照。

SQLite 实现

SqliteSaver（libs/checkpoint-sqlite/）：

• 使用单一 SQLite 表存储 checkpoint
• 索引：(thread_id, checkpoint_id, checkpoint_ns)
• 序列化：msgpack 编码 checkpoint 数据
• 原子写入：利用 SQLite 的事务机制

PostgreSQL 实现

PostgresSaver（libs/checkpoint-postgres/）：

• 使用多表存储：checkpoint 表、writes 表、blobs 表
• JSONB 存储结构化数据，支持查询
• 利用 PostgreSQL 的并发控制（行级锁）
• 支持大规模并发读写

序列化格式

LangGraph 使用 msgpack 而非 JSON：

维度	msgpack	JSON
体积	紧凑（二进制）	较大（文本）
速度	快（直接序列化）	慢（文本编解码）
类型支持	支持 bytes、datetime	仅字符串/数字/布尔
可读性	不可读	人类可读

Checkpoint ID 使用 xxhash 生成：

from xxhash import xxh3_128_hexdigest
checkpoint_id = xxh3_128_hexdigest(serialized_data)

xxhash 比 MD5/SHA 快 5-10 倍，适合高频 checkpoint 场景。

陷阱与对策

自定义 checkpointer 的原子性

问题：如果 put() 操作不是原子的（如先写 checkpoint 再写 writes），并发写入可能导致不一致。

规避：使用事务机制，确保 put() 和 put_writes() 在同一事务中。

大 checkpoint 的存储膨胀

问题：长期运行的 Agent 会产生大量 checkpoint，占用过多存储空间。

规避：

• 定期清理过期 checkpoint
• 使用差异存储（只保存变化部分）
• 考虑只保存关键步骤的 checkpoint

设计取舍

优势

• 可插拔：三种后端适配不同部署场景
• 一致性测试：checkpoint-conformance 包确保所有后端行为一致
• 高性能：msgpack + xxhash 降低序列化开销

代价

• 序列化限制：状态必须是 msgpack 可序列化的
• 存储成本：完整 checkpoint 体积较大

替代方案

• JSON 序列化：人类可读，但性能和体积较差
• 差异存储：只保存变化，但恢复时需要回放差异链

参考来源

• 源码验证: libs/checkpoint/langgraph/checkpoint/base/__init__.py — 接口定义
• 源码验证: libs/checkpoint-sqlite/ — SQLite 实现
• 源码验证: libs/checkpoint-postgres/ — PostgreSQL 实现