StateSnapshot:各スーパーステップ終了時の状態スナップショット

源码版本1.2.9

役割

StateSnapshot は LangGraph が各スーパーステップ (superstep) 境界で外部に晒す「このステップが終わった瞬間のグラフ全体の姿」の不変スナップショットです。5 つの情報を 1 つの NamedTuple に詰めます:現在の各チャネル (channel) の値 (values)、次ステップで走るノード (next)、今回使った config、対応するチェックポイント (checkpoint) のメタデータ (metadata + created_at)、そして親チェックポイントの config (StateSnapshot 定义:643-661)。graph.get_state(config)、graph.aget_state(config)、あるいは graph.get_state_history(config) のイテレーションで返ってくるのがすべてこの型です。

アーキテクチャ上は「チェックポイントの上、ユーザー画面の下」に位置します。チェックポイント (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 の 1 層挟むことで、チェックポイント形式を自由に進化させても外部 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)。そのためノードが書き終わった直後でまだチェックポイントに落ちていない時に get_state を呼んでも、1 つ前のステップの古い値ではなく最新値が見えます。

主要ファイル

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 を呼んでこのステップで走るタスク辞書を算出します。
task_states 递归: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 でこのステップのタスクを取得し、サブグラフ再帰を処理します。

python

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 書き込み済みのタスク名をフィルタして、不変スナップショットを組み立てます:

python

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 が空タプルかどうかで「グラフがまだ走っていない」と「グラフが走り終わった」を区別する必要があります。
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

StateSnapshot:各スーパーステップ終了時の状態スナップショット ​

役割 ​

設計動機 ​

主要ファイル ​

データフロー ​

境界と失敗 ​

まとめ ​

StateSnapshot:各スーパーステップ終了時の状態スナップショット

役割

設計動機

主要ファイル

データフロー

境界と失敗

まとめ