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