InMemorySaver + SqliteSaver:进程内与文件级检查点

源码版本1.2.9

职责

BaseCheckpointSaver 给了接口形状,真正落地的两个最常用 saver 是 InMemorySaver 和 SqliteSaver——前者把整份状态塞进进程内存里的 defaultdict,后者把同样的字段拆成 checkpoints 和 writes 两张 SQLite 表。两者一起覆盖了开发调试和单机持久化两个场景,生产规模才换 PostgresSaver。

InMemorySaver 在 libs/checkpoint/langgraph/checkpoint/memory/__init__.py (InMemorySaver:33),内部用三个 defaultdict 装三样东西:storage 装 checkpoint 主体 + 元数据 + parent_id,writes 装 (thread_id, ns, checkpoint_id) → (task_id, idx) → 写记录,blobs 装 (thread_id, ns, channel, version) → 序列化后的二进制 (storage / writes / blobs:68-83)。这个分桶结构跟 Postgres/SQLite 的三表结构是一一对应的,只是用 dict 代替了表。

SqliteSaver 在 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py (SqliteSaver:45)。它直接继承 BaseCheckpointSaver[str],拿一个 sqlite3.Connection,在 setup() 里建两张表(<SrcLink path="libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py" lines="129-166" label="setup"/>),put / put_writes 走标准 SQL INSERT OR REPLACE / INSERT OR IGNORE。它自带 threading.Lock 把单连接串行化,所以 check_same_thread=False 是安全的。

另外 MemorySaver = InMemorySaver 是历史别名 (MemorySaver alias:631);老代码里常见的 from langgraph.checkpoint.memory import MemorySaver 拿到的就是同一个类。

设计动机

• InMemory 的三桶 dict 对应三表:storage / writes / blobs 这三个字段 (data members:68-83) 跟 SQLite 的 checkpoints / writes 表和 Postgres 的 checkpoints / checkpoint_writes / checkpoint_blobs 三表一一对应。InMemory 把 blob 也独立存,是为了让 channel value 的序列化只在版本变化时发生一次,而不是每次 put 整个 checkpoint 都重序列化所有通道。
• PersistentDict 做可选落盘:memory 模块里还有个 PersistentDict(defaultdict) (PersistentDict:634),用 pickle 把 dict dump 到文件,关机重启再 load。InMemorySaver 的 __init__ 接收 factory 参数 (__init__:85-99) 就是为了让 factory=PersistentDict 时自动挂上下文管理器,这是「调试用但想跨重启保状态」的折中路径。
• SQLite 走 WAL + 单 lock:setup() 第一条就是 PRAGMA journal_mode=WAL (WAL:141),让读不阻塞写;但写还是靠 self.lock 串行,因为 sqlite3.Connection 不是线程安全的。docstring 明说「不scale 到多线程」(note:48-54),要并发上 Postgres。
• 写表区分 REPLACE 和 IGNORE:put_writes 在 SQLite 里根据写是不是全在 WRITES_IDX_MAP 里选 INSERT OR REPLACE 还是 INSERT OR IGNORE (put_writes query:462-466):控制 channel 写 (ERROR / INTERRUPT / RESUME) 可以覆盖,普通业务写撞了就忽略,语义跟 InMemory 里 if inner_key[1] >= 0 ... continue 一致 (InMemorySaver.put_writes:499-509)。
• InMemory 的 max(checkpoints.keys()) 当 latest:没有显式 ORDER BY,靠 checkpoint_id 字符串单调递增 (latest:282-283),所以 checkpoint_id 必须是字典序可比较的 UUID6/UUID7,不能用纯随机 UUID。

关键文件

• InMemorySaver 类定义:33-83 — 类声明 + storage / writes / blobs 三个 defaultdict 字段。
• InMemorySaver.put:427-471 — 把 channel_values 拆到 blobs,checkpoint 主体进 storage,返回新 config。
• InMemorySaver.put_writes:473-509 — 跳过已有负 idx 控制写,普通写按 (task_id, idx) 索引。
• InMemorySaver.get_tuple:236-316 — 用 checkpoint_id 精确查或 max(keys()) 取最新,再拼回 CheckpointTuple。
• MemorySaver alias:631 — 历史别名,等价 InMemorySaver。
• SqliteSaver 类:45-95 — 接 sqlite3.Connection,带 threading.Lock 串行写。
• SqliteSaver.setup:129-166 — 建表 + 开 WAL。
• SqliteSaver.get_tuple:191-263 — 两条 prepared SELECT,按 checkpoint_id 精确查或 ORDER BY checkpoint_id DESC LIMIT 1 取最新,再 JOIN writes。
• SqliteSaver.put:387-443 — INSERT OR REPLACE INTO checkpoints,metadata 用 JSON 序列化成 bytes。
• AsyncSqliteSaver:38 — 异步版,aput / aget_tuple / alist 用 aiosqlite 重新走。

数据流

InMemorySaver.put 这一段把「拆 channel_values 到 blobs」和「主体进 storage」两件事一次完成,可以很清楚地看到 saver 怎么把一个 Checkpoint TypedDict 拆成三层结构:

c = checkpoint.copy()
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"]["checkpoint_ns"]
values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
for k, v in new_versions.items():
    self.blobs[(thread_id, checkpoint_ns, k, v)] = (
        self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
    )
self.storage[thread_id][checkpoint_ns].update(
    {
        checkpoint["id"]: (
            self.serde.dumps_typed(c),
            self.serde.dumps_typed(get_checkpoint_metadata(config, metadata)),
            config["configurable"].get("checkpoint_id"),  # parent
        )
    }
)

(put body:448-464) 注意 blobs 用 (thread_id, checkpoint_ns, channel, version) 当 key,而 storage 用 (thread_id, checkpoint_ns, checkpoint_id) 当 key——两套索引不重合,channel value 跟 checkpoint 主体解耦,版本没变的通道不需要重序列化。读回来时 get_tuple 调 _load_blobs 把 channel_versions 里每个 (channel, version) 对应的 blob 反序列化拼回 channel_values (_load_blobs:259-265),再装进 CheckpointTuple。

SqliteSaver 走同样的逻辑,只是 dict 换成 SQL,blob 桶换成主表里的 checkpoint BLOB 列 + writes 表里的 value BLOB 列——没有独立的 blob 表,所有 channel value 直接 serde 进 checkpoint blob 字段里:

cur.execute(
    "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
    (
        str(config["configurable"]["thread_id"]),
        checkpoint_ns,
        checkpoint["id"],
        config["configurable"].get("checkpoint_id"),
        type_,
        serialized_checkpoint,
        serialized_metadata,
    ),
)

(SqliteSaver.put INSERT:425-436)

边界与失败

• InMemorySaver 进程一挂全没:docstring 明说「只用 InMemorySaver 做调试或测试」(warning:40-44),要持久化要么换 Sqlite 要么 Postgres。
• InMemorySaver 取 latest 靠 checkpoint_id 字典序:max(checkpoints.keys()) (max:282-283) 假设 ID 单调递增;如果自定义 saver 用非单调 ID,get_tuple 不带 checkpoint_id 时会拿错行。
• SqliteSaver 单连接单锁:cursor() 上下文管理器拿 self.lock (cursor:181-189) 并发写会被串行化,长事务会阻塞其他读。要并发只能上 Postgres 的 pipeline 模式。
• SqliteSaver 用 ? 占位符列名顺序硬编码:列名顺序必须跟 INSERT INTO checkpoints (...) 列表一致 (INSERT columns:426),迁移加列要同步改两处;task_path 列就是后加的,见 Postgres MIGRATIONS 末尾 (task_path migration:90)。
• AsyncSqliteSaver 不只是包 to_thread:AsyncSqliteSaver:38 是独立类,aput / aget_tuple / alist 全部用 aiosqlite 重写 (aput:509),不是基类默认的 asyncio.to_thread 转发——所以异步路径能真并发,不会被全局 lock 串行。
• factory= 用 PersistentDict 时退出要 close():PersistentDict.sync() 把 pickle dump 到文件 (sync:657),不调 __exit__ 不会自动落盘;with InMemorySaver(factory=PersistentDict, ...) 是惯用法。

小结

InMemorySaver 是「三桶 dict」的参考实现,SqliteSaver 把同样的三桶展开成 SQLite 的两张表 + WAL,两者一起覆盖了调试和单机持久化;主键设计、blob 拆分、控制 channel 负 idx 这些约定都在这两份实现里见底。生产规模见 PostgresSaver,接口形状见 BaseCheckpointSaver。

对照官方资料:LangGraph 文档 · README。