03. State と Checkpoint¶

ゴール¶

LangGraph の state 更新ルール（reducer）を理解する
Checkpointer で会話を永続化し、プロセスを落としても続きから再開する
SQLite と Postgres の両方で動かす

State の reducer¶

State は TypedDict で定義するが、フィールドごとに どう更新するか（reducer）を Annotated で指定できる。

from typing import Annotated, TypedDict
from operator import add

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 追加（同じIDなら上書き）
    counter: Annotated[int, add]              # 加算
    user_name: str                             # 上書き（デフォルト）

デフォルト: 上書き（最新の値で置換）
add_messages: messages 用の特殊 reducer。message ID で上書き or 追加
add (operator.add): 加算 or list 連結
自作 reducer: def my_reducer(left, right) -> X を渡せる

ノードは部分的な state を返せばよい。reducer がマージしてくれる。

def node(state: State) -> dict:
    return {"counter": 1, "messages": [new_msg]}  # counter は+1、messages は追加

Checkpointer：state を永続化する¶

graph.compile(checkpointer=...) で permissioned。各 super-step ごとに state snapshot を保存する。

SQLite で動かす¶

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
    graph = builder.compile(checkpointer=memory)

    config = {"configurable": {"thread_id": "user-42"}}
    graph.invoke({"messages": [("user", "私の名前はケン")]}, config)
    graph.invoke({"messages": [("user", "私の名前は何でしたか？")]}, config)
    # → 「ケンです」と答える

thread_id が 会話の識別子。同じ thread_id なら state が引き継がれる
別 thread_id にすれば独立した会話になる（マルチユーザー対応の基本形）

Postgres で動かす¶

uv add langgraph-checkpoint-postgres psycopg2-binary

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/langgraph"

with PostgresSaver.from_conn_string(DB_URI) as memory:
    memory.setup()  # 初回のみ。テーブル作成
    graph = builder.compile(checkpointer=memory)
    # ...以下 SQLite と同じ

本番では Postgres 一択。Aurora、Cloud SQL、Supabase どれでも可。

state を覗く¶

state = graph.get_state(config)
print(state.values)              # 現在の state
print(state.next)                # 次に実行されるノード
print(state.tasks)               # 保留中のタスク

履歴も取れる:

for snapshot in graph.get_state_history(config):
    print(snapshot.config["configurable"]["checkpoint_id"], snapshot.next)

Time travel（過去に戻る）¶

# 過去の checkpoint_id を取って、その地点から再実行
target = list(graph.get_state_history(config))[3]  # 3つ前
graph.invoke(None, target.config)

LLM の応答を後で変えて「もしあの時こう答えていたら」を試せる。デバッグに強力。

State 更新（人間が割り込む）¶

graph.update_state(config, {"messages": [HumanMessage("やっぱり違う、こっちで")]})

これで state を上書きできる。次章の human-in-the-loop で使う。

Try¶

SQLite checkpointer で会話を3往復させ、プロセスを Ctrl+C で落としてから再起動して続きを話せるか確認
get_state_history で時系列を表示
counter フィールドを add reducer で追加し、各ノードで +1 して step 数を数える
Postgres でも同じことを試す（Docker で postgres:16 立てれば早い）

学んだこと¶

State は TypedDict + reducer。ノードは部分更新を返す
Checkpointer は thread 単位で state snapshot を保存
SQLite はローカル/プロトタイプ、Postgres は本番
get_state / get_state_history / update_state で state を観察・操作できる
これが durable execution の正体（Temporal の event history 方式とは別アプローチ）

Temporal と比較¶

	LangGraph Checkpoint	Temporal Event History
単位	super-step（ノード境界）	Activity 実行
復帰方法	最後の checkpoint から該当ノードを再実行	Workflow コードを history から replay
状態の所在	DB の checkpoints テーブル	Temporal Server の event history
プロセス障害時	OK（次回 invoke で続きから）	OK（worker が拾い直す）
「数日寝かせる」	できるが scheduler は自作	1級サポート（durable timer）

次は 04_human_in_the_loop。途中で人間レビューを入れる方法を扱う。

作成: 2026-05-16 ／最終更新: 2026-05-16