StateSnapshot:每個超步結束後的狀態快照

源码版本1.2.9

職責

StateSnapshot 是 LangGraph 在每個超步 (superstep) 邊界上對外暴露的「這一步剛跑完時整張圖長什麼樣」的不可變快照。它把五件事打包成一個 NamedTuple:當前各「通道 (channel)」的值 (values)、下一步要跑哪些節點 (next)、本次用的 config、對應的 checkpoint metadata (metadata + created_at)、以及父檢查點的 config (StateSnapshot 定義:643-661)。你呼叫 graph.get_state(config)、graph.aget_state(config),或遍歷 graph.get_state_history(config),拿回來的都是這個型別。

它在架構裡處在「檢查點 (checkpoint) 之上、使用者視圖之下」。檢查點 (CheckpointTuple) 是引擎內部的序列化格式,直接存的是 channel_values / channel_versions / pending_writes 這些低層欄位,看不懂業務狀態長什麼樣;StateSnapshot 則把這層翻譯成「業務能直接讀的 dict + 下一步要跑什麼 + 任務列表」(_prepare_state_snapshot:1145-1162)。子圖狀態也走它——task.state 欄位在遞迴 subgraphs=True 時會塞一個子圖的 StateSnapshot(PregelTask.state:200-201)。

StateSnapshot 本身是 NamedTuple,不可變;要「改狀態」必須透過 graph.update_state(config, values) 寫新值,引擎會把它變成 NULL_TASK_ID 的 pending write 落到下一個檢查點裡,然後再生成一份新的 StateSnapshot。

設計動機

為什麼不直接把 CheckpointTuple 暴露給使用者?

• 解耦內部表示:檢查點格式為了高效序列化,存的是 channel_versions 這種帶版本號的扁平 dict;而使用者關心的是「我的業務 state dict」,中間需要 read_channels 把通道值讀出來 (read_channels:1258)。多一層 StateSnapshot 讓檢查點格式可以自由演進,不破壞對外 API。
• 帶上「下一步」語義:引擎知道下一步要跑哪些節點,但 CheckpointTuple 本身只存寫過的值;StateSnapshot 在生成時呼叫 prepare_next_tasks(prepare_next_tasks:1178-1195),算出 next=(node_name, ...),直接回答「接下來會發生什麼」。
• 任務列表可見:每個 StateSnapshot.tasks 是當前步要執行的 PregelTask 元組(tasks:658),包含 writes / interrupts / 子圖狀態——這是人機協作裡判斷「現在卡在哪個節點、等什麼輸入」的關鍵。
• 不可變 = 可對比:歷史快照用 parent_config 串成鏈結串列(parent_config:656),get_state_history 拿到的就是按時間倒序的快照列表,任意一份都可以用 config 重新指向它做時間旅行。
• pending writes 透明應用:get_state 預設會把還沒落檢查點的 pending_writes 臨時應用進通道再讀 (apply pending writes:1239-1249),所以你在節點剛寫完但還沒落 checkpoint 時呼叫 get_state,看到的就是最新值,而不是上一步的舊值。

關鍵檔案

• StateSnapshot NamedTuple:643-661 — 7 個欄位:values / next / config / metadata / created_at / parent_config / tasks / interrupts。
• PregelTask:200-217 — 任務物件,state 欄位在遞迴時存子圖的 StateSnapshot 或定位它的 RunnableConfig。
• _prepare_state_snapshot:1145-1162 — 空 saved 時的兜底:回傳 values={} / next=() / tasks=() 的空快照。
• step 計算 + prepare_next_tasks:1167-1195 — 從 saved.metadata["step"] 推當前步號,呼叫 prepare_next_tasks 算出本步要跑的任務字典。
• taskstates 遞迴:1199-1227 — 對每個 task.name in subgraphs 的任務,組裝 checkpoint_ns 後遞迴呼叫子圖的 get_state。
• apply pending writes:1229-1255 — 把 NULL_TASK_ID 的寫和常規 pending writes 臨時應用到通道,再算 tasks_with_writes。
• 組裝 StateSnapshot:1257-1266 — 把通道值、next 節點元組、config、metadata、ts、parent_config、tasks、interrupts 拼成最終 StateSnapshot。
• get_state:1392-1434 — 同步入口:解析 checkpointer、處理 checkpoint_ns 路由到子圖、決定是否 apply_pending_writes。
• get_state_history:1480 — 遍歷同一 thread_id 下所有歷史檢查點,每個都生成 StateSnapshot。
• _aprepare_state_snapshot:1268 — 非同步版,邏輯跟同步版對稱,只是 channels_from_checkpoint 換成 achannels_from_checkpoint。

資料流

下面是 _prepare_state_snapshot 把檢查點翻譯成 StateSnapshot 的關鍵段:從 saved.metadata 推步號、呼叫 prepare_next_tasks 拿到本步任務,再處理子圖遞迴。

step = saved.metadata.get("step", -1) + 1
stop = step + 2
channels, managed = channels_from_checkpoint(
    self.channels,
    saved.checkpoint,
    saver=self.checkpointer
    if isinstance(self.checkpointer, BaseCheckpointSaver) else None,
    config=saved.config,
)
next_tasks = prepare_next_tasks(
    saved.checkpoint,
    saved.pending_writes or [],
    self.nodes,
    channels,
    managed,
    saved.config,
    step,
    stop,
    for_execution=True,
    store=self.store,
    checkpointer=(self.checkpointer
                  if isinstance(self.checkpointer, BaseCheckpointSaver) else None),
    manager=None,
)

(prepare_next_tasks 呼叫:1167-1195)

最後用讀出來的通道值、過濾掉已寫了 pending 的任務名,組裝成不可變快照:

return StateSnapshot(
    read_channels(channels, self.stream_channels_asis),
    tuple(t.name for t in next_tasks.values() if not t.writes),
    patch_checkpoint_map(saved.config, saved.metadata),
    saved.metadata,
    saved.checkpoint["ts"],
    patch_checkpoint_map(saved.parent_config, saved.metadata),
    tasks_with_writes,
    tuple([i for task in tasks_with_writes for i in task.interrupts]),
)

(組裝:1257-1266)

邊界與失敗

• 沒有檢查點時回傳空快照而非報錯(空 saved:1152-1162),values={} / next=(),呼叫方需要自己判斷 snapshot.next 是不是空 tuple 來區分「圖還沒跑」和「圖跑完了」。
• 沒設 checkpointer 直接呼叫 get_state 拋 ValueError(no checkpointer:1401-1402),No checkpointer set——純記憶體跑的圖沒有歷史快照。
• checkpoint_ns 找不到匹配子圖時報錯(subgraph not found:1415-1416),Subgraph {recast} not found,常見於 namespace 拼接錯了 NS_SEP(|) 或 NS_END(:)。
• apply_pending_writes 僅在沒顯式給 checkpoint_id 時啟用(apply 條件:1433),也就是說你拿歷史快照看的就是當時落盤的值,不會被後續 pending write 污染;拿「最新」快照才會臨時應用 pending。
• next 欄位過濾掉已經有 writes 的任務(next 過濾:1259),t.writes 非空的任務不算「next」——因為它們的輸入已經被本步消費了,所以 next 反映的是「還沒跑的任務」。
• tasks 包含 interrupts(interrupts 聚合:1265),把所有任務的 Interrupt 物件扁平化拼成元組,使用者呼叫 Command(resume=...) 時按這個順序回填。

小結

StateSnapshot 是 LangGraph 把內部檢查點格式翻譯成業務可讀視圖的邊界型別——它不存資料,只把檢查點 + pending writes 加上「下一步」語義組裝成不可變快照。所有斷點續跑、時間旅行、人機協作的 API 都建立在它之上。

它跟 RunnableConfig 是一對:config 是「怎麼找到這個快照」的座標,StateSnapshot 是「找到之後看到什麼」。繼續看 StateGraph 怎麼定義出被快照的狀態形狀,或看 Pregel 引擎 裡 _prepare_state_snapshot 是怎麼被 get_state 呼叫到的。

對照官方資料:LangGraph 文件 · README