17. 状態の永続化とチェックポイント

第15章ではエージェントループを StateGraph として再構築し、第16章ではプリビルトコンポーネントとマルチブランチのルーティングを使ってより手の込んだものを構築しました。これまでに書いてきたすべてのグラフには、1つの制約が共通しています。それは、グラフが自身の状態(state)を保持しないということです。

グラフは、1回の invoke() 呼び出しのあいだだけ状態を保持・管理し、それ以上は保持しません。呼び出すと、LangGraphは新しい状態を作成し、ノードを実行し、リデューサーのルールに従って各戻り値を状態にマージし、最終的な状態を呼び出し元に返します。その状態を渡してしまえば、グラフはもうそれを覚えていません。次の invoke() は、前回の呼び出しとは何のつながりもない、まっさらな状態から始まります。

ここから2つの問題が生じます。まず、messages も状態の一部なので、エージェントは以前あなたが言ったことを何も覚えていられません。次に、実行が途中で失敗すると、その時点までに達成したことがすべて消えてしまいます。3番目のノードが例外を発生させたとしましょう。最初の2つのノードが生成した結果もそれと一緒に失われ、最初からやり直すことになります。15.1で「中断からの回復」をLangGraphに手を伸ばす理由の1つとして挙げましたが、まさにここで考えていたのがこの問題です。

LangGraphはこれをフレームワークのレベルで扱います。グラフにチェックポインター(checkpointer)を取り付けると、LangGraphは実行の各ステップで状態のスナップショットを自動的に保存します。これらの保存されたスナップショットは invoke() 呼び出しよりも長く生き残るので、次の呼び出しは前回が終わった箇所から再開できます。この性質——1回の実行を超えて状態が生き残ること——を永続化(persistence)と呼びます。

実はあなたはこれまでにもチェックポインターを使ったことがあります。第11章で会話型RAGエージェントにマルチターンのメモリを与えたとき、create_agent(..., checkpointer=InMemorySaver()) と thread_id を渡しました。そのときは、チェックポインターが thread_id ごとに会話履歴を保持することさえ知っていればよく、その仕組みは説明しませんでした。また第16章で checkpointer パラメータを紹介したとき、「これがどう機能するかは第17章で扱います」と述べました。それがこの章です。

この章は3つのパートから成ります。17.1では、グラフにチェックポインターを取り付け、thread_id を使ってマルチターンの会話を行います。17.2では、保存されたチェックポイントを開いて、エージェントが任意の時点で何を知っていたかを確認します——チェックポイントは、エージェントがなぜ誤動作したのかを突き止めるための主要なツールです。17.3では、実行の途中で失敗したグラフを、最初からではなく、止まった箇所から再開します。

17.1) 呼び出しをまたいで状態を持ち運ぶ

17.1.1) 忘れてしまうグラフ

この章の冒頭で、グラフは自身の状態を保持しないと述べました。それをコードで確認しましょう。

以下のグラフは、15.2の say_hello グラフと同じ形をしています。唯一の違いは、ノードが固定の文字列ではなくLLMの応答を返すことです。

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
graph = builder.compile()   # チェックポインターなし

では、2ターンの会話をしてみましょう。最初の呼び出しでモデルに自分の名前を伝え、2回目の呼び出しで名前が何かを尋ねます。

# 最初の呼び出し — 名前を伝える
graph.invoke({"messages": [{"role": "user", "content": "Hi, my name is Bob."}]})
 
# 2回目の呼び出し — 名前を聞き返す
result = graph.invoke({"messages": [{"role": "user", "content": "What's my name?"}]})
print(result["messages"][-1].content)

出力:

申し訳ありませんが、あなたの名前は分かりません。名前を教えていただけますか?

2回目の呼び出しでモデルが受け取ったメッセージは "What's my name?" だけでした。最初の呼び出しの状態はもうグラフに残っていないので、以前のメッセージ——名前を運んでいたもの——はモデルには届かなかったのです。

もちろん、これを自分たちで解決することもできます。最初の呼び出しが返した messages を保持しておき、2回目と一緒に渡すのです。それはまさに第8章で会話履歴を管理していた方法です。しかしそうすると、すべての会話とすべてのユーザーについて、履歴を保存・取得するコードを自分で書くことになります。それこそが、チェックポインターがあなたから引き受けてくれる作業です。

17.1.2) チェックポイントとチェックポインター

チェックポインター(checkpointer)は、状態を保存することを役割とするオブジェクトです。インスタンスを作成し——たとえば InMemorySaver() ——それを builder.compile(checkpointer=...) に渡してグラフに取り付けます。

チェックポインターが取り付けられると、グラフは実行が進むにつれて状態全体をコピーして保存します。それらの保存されたコピーの1つ1つがチェックポイント(checkpoint)と呼ばれます。写真だと思ってください。その瞬間の状態全体が、そのままの姿で保存されるのです。

ビデオゲームのオートセーブが、ちょうど良いイメージです。ゲームは、あなたが意味のあるポイントを通過するたびに進行状況をこっそり記録するので、いったんやめて後で戻ってきたり、最初からやり直すことなく死んだりできます。チェックポインターはグラフに対してまさにこれを行います。

では「意味のあるポイント」とはいつでしょうか。LangGraphはグラフの実行をステージに分割し、各ステージはスーパーステップ(super-step)と呼ばれます。スーパーステップが終わるたびにチェックポイントが保存されます。

これが単なるステップではなくスーパーステップである理由は、1つのステージで複数のノードを同時に実行できるからです。17.1.1のグラフのように、ノードが一直線に並んでいる場合、1つのノードの実行が1つのスーパーステップです。しかし複数のノードが並列に実行されるグラフでは、それらのノードすべてがまとめて1つのスーパーステップを構成します。

つまり、invoke() を1回呼び出すだけでも、いくつものチェックポイントが残ります。それらを取り出して、それぞれに何が含まれているかを17.2で詳しく見ていきます。

例として使ってきた InMemorySaver は、存在する中で最もシンプルなチェックポインターです。名前が示すとおり、チェックポイントをプロセスのメモリ(RAM)に保存します。インストールするものも設定するものも何もないので、学習やローカル開発によく合います。トレードオフは、保存されたすべてのチェックポイントがプロセスの再起動で消えてしまうことです。本番向けの代替手段は17.1.5で見ていきます。

17.1.3) チェックポインターを追加する

チェックポインターの取り付けに必要なのは、たった2つのことだけです。

1. チェックポインターのインスタンスを作成し、compile() に渡す。
2. invoke() を呼び出すたびに、thread_id を含む config を渡す。

2番目がなぜ必要なのかはまもなく説明します。今のところは、それがチェックポインターに、保存された会話のどれを続けたいのかを伝えるものだと理解しておいてください。

両方を17.1.1のグラフに適用しましょう。

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
# 1. チェックポインターを作成し、comp()に渡す
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
 
# 2. thread_idを持つconfigをinvoke()に渡す
config = {"configurable": {"thread_id": "1"}}
 
graph.invoke(
    {"messages": [{"role": "user", "content": "Hi, my name is Bob."}]},
    config,
)
result = graph.invoke(
    {"messages": [{"role": "user", "content": "What's my name?"}]},
    config,
)
print(result["messages"][-1].content)

出力:

あなたの名前はBobです。

17.1.1と同じ2回の呼び出しですが、結果が違います。今回は名前がちゃんと残りました。

理由はこうです。チェックポインターが取り付けられたグラフは、llm_call ノードを実行する前に、保存された状態を読み込みます。その状態にはすでに最初のやり取りが含まれています。渡された新しいメッセージは、それにマージされます。15.2.2で見たように、messages フィールドには add_messages リデューサーが付いているので、新しいメッセージは既存のリストに追加されます。結果として、モデルは3つのメッセージを受け取ります。最初のあいさつ、それ自身の最初の返答、そして新しい質問です。

私たちは新しいメッセージだけを送り、LangGraphが最後のチェックポイントから以前の会話を読み込みます。 第8章で手作業で管理していた会話履歴が、今やLangGraphによって管理されているのです。

17.1.4) thread_id: 会話を分離する識別子

thread_id は、ある会話を別の会話から区別する識別子です。値はあなたが選びます。17.1.3では "1" を使いましたが、どんな文字列でも構いません。同じ thread_id でグラフを呼び出せばその会話を続けることになり、別のもので呼び出せば別の会話を始めることになります。

それを確認しましょう。AliceとBobに、同じグラフを通して別々の会話をしてもらいます。

def send(thread_id: str, text: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(
        {"messages": [{"role": "user", "content": text}]},
        config,
    )
    return result["messages"][-1].content
 
# Aliceの会話
send("alice", "My favorite color is teal.")
 
# Bobの会話 — 別のthread_id
send("bob", "My favorite color is orange.")
 
# それぞれにもう一度尋ねる
print("Alice:", send("alice", "What's my favorite color?"))
print("Bob:  ", send("bob", "What's my favorite color?"))

出力:

Alice: あなたの好きな色はティール(青緑)です。
Bob:   あなたの好きな色はオレンジです。

どちらの会話も同じ graph オブジェクトと同じチェックポインターを通りましたが、決して混ざりませんでした。thread_id は、チェックポインターが状態を保存し検索するために使う主キー(primary key)です。キーが違えば、ストレージは完全に別々です。

では、値を省略するとどうなるでしょうか。

graph.invoke({"messages": [{"role": "user", "content": "Hello"}]})

出力:

ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

グラフはまったく実行されません。チェックポインターが取り付けられると、thread_id はオプションではなく——必須になります。

いま見たことを踏まえれば、それも納得できます。ノードを実行する前に、チェックポインターは保存された状態を読み込まなければなりません——そして thread_id は、どの会話の状態を読み込むかを伝えるものなのです。

これがチャットボットサービスの基本的な形です。1つのグラフ、1つのチェックポインター、そしてユーザーごとまたはチャットルームごとに1つの thread_id。

17.1.5) InMemorySaver の限界と、本番向けの代替手段

先ほど、InMemorySaver はチェックポイントをメモリに保持すると述べました。その選択には2つの限界が伴います。

プロセスを再起動すればすべて消えます。 サービスを再デプロイしたりサーバーを立ち上げ直したりすると、それまでに蓄積されたすべての会話が消えてしまいます。

別々のプロセス同士では共有できません。 実際のサービスは、受け取ったリクエストを複数のプロセスに分散させます。各プロセスは自分自身のメモリを持っているので、プロセスAが保存した会話はプロセスBには見えません。ユーザーは毎回同じ thread_id を送っても、どのプロセスがたまたまリクエストを拾うかによって、会話が崩壊するのを目にすることになりかねません。

だからこそ、本番ではチェックポイントをデータベースに保存するチェックポインターを使います。

• SqliteSaver / AsyncSqliteSaver (langgraph-checkpoint-sqlite) — すべてを単一のファイルに保存します。1台のサーバーで動く小規模なサービスや、ローカルのプロトタイプによく合います。
• PostgresSaver / AsyncPostgresSaver (langgraph-checkpoint-postgres) — チェックポイントをデータベースサーバーに保存します。サーバーを増やしても、どのプロセスも同じチェックポイントを見られます。これはLangGraphのドキュメントが本番向けに推奨しているチェックポインターです。

これらはすべて、InMemorySaver と同じインターフェースを実装しています。あなたのグラフのコード、ノード、そして thread_id の使い方は、そのまま変わりません。変わるのは、チェックポインターをどう作成するかだけです。

第16章で出会った create_agent も、同じ方法でチェックポインターを使います。その checkpointer パラメータにチェックポインターを渡し、thread_id を持つ config を invoke() に渡します。これが、第11章の会話型RAGエージェントに以前のターンを覚えさせていたものです。

この章の残りでは引き続き InMemorySaver を使います。チェックポイントがどこに置かれることになろうと、それらを扱う方法は同じです。

17.2) 状態の確認とデバッグ

17.1.2で、チェックポインターはすべてのスーパーステップで状態を保存すると述べました。それらのチェックポイントを取り出して見てみましょう。

保存されたチェックポイントを開くのは、単なる好奇心からではありません。エージェントは誤動作します。終わりのないループでツールを呼び出したり、以前のターンを見失ったり、まったく予期していなかった分岐を進んだりします。その理由を突き止めるには、その瞬間に状態がどう見えていたかを知る必要があります——そしてチェックポイントがその答えを持っています。

LangGraphは2つのメソッドを提供します。

• graph.get_state(config) — その会話の最新のチェックポイントを返します。
• graph.get_state_history(config) — その会話のすべてのチェックポイントを、新しいものから順に返します。

どちらのメソッドも config に thread_id を必要とします。誰のチェックポイントを尋ねているのかを知らなければならないからです。

これら2つのメソッドが返すチェックポイントは、StateSnapshot オブジェクトとして表現されます。

17.2.1) StateSnapshot: チェックポイントの中身

チェックポイントに実際に何が含まれているのかを見てみましょう。この例ではLLMを使いません。LLMは毎回違うものを返すので、各フィールドを1つずつ見ていきたいときには役に立ちません。代わりに、固定の値を返す小さなグラフを作ります。

状態には2種類のフィールドを持たせます。foo にはリデューサーがないので上書きされ、bar にはリデューサーがあるので蓄積されます。15.2.2で messages に取り付けた add_messages リデューサーと同じ仕組みです——ここではPythonの operator.add をリデューサーとして使い、リストを連結します。

from operator import add
from typing_extensions import TypedDict, Annotated
 
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    foo: str                        # リデューサーなし → 上書き
    bar: Annotated[list[str], add]  # addリデューサー → 蓄積
 
def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}
 
def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}
 
builder = StateGraph(State)
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar": []}, config)
 
snapshot = graph.get_state(config)
print(snapshot)

出力:

StateSnapshot(
    values={'foo': 'b', 'bar': ['a', 'b']},
    next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                             'checkpoint_id': '1f17da03-9654-65d8-8002-9e59231bb481'}},
    metadata={'source': 'loop', 'step': 2, 'parents': {}},
    created_at='2026-07-12T03:17:33.637368+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                                    'checkpoint_id': '1f17da03-9653-6ca0-8001-7c24c8ec66f2'}},
    tasks=(),
    interrupts=()
)

8つのフィールドがあります。1つずつ見ていきましょう。

• values — このチェックポイント時点の状態です。bar が ['a', 'b'] なのは、add リデューサーが両方のノードが返したものを蓄積したからです。foo が 'b' なのは、リデューサーがなく、最後の書き込みが勝ったからです。これが「その瞬間に状態がどう見えていたか?」に答えるフィールドです。
• next — このチェックポイントの後に実行されるノード名のタプルです。空の () は実行すべきものが残っていない、つまりグラフが終了したことを意味します。('node_b',) は node_b がまだこの先にあることを意味します。
• config — このチェックポイントのアドレスです。thread_id は会話を識別し、checkpoint_id はその中のどの瞬間かを識別します。LangGraphはチェックポイントを保存するたびに checkpoint_id を自動的に割り当てます。
• metadata — 実行に関する記録情報です。source はチェックポイントがどこから来たかを教えてくれます。"input" は invoke() に渡した入力から構築されたことを、"loop" はグラフの実行中に生成されたことを意味します。step はスーパーステップの番号です。
• created_at — チェックポイントが保存された時刻です。ログと突き合わせるときに便利です。
• parent_config — このチェックポイントの直前のチェックポイントの config です。それをたどれば、実行を後ろ向きに歩いていけます。一番最初のチェックポイントでは None になります。
• tasks — next にリストされたノードの実行記録です。チェックポイントが保存される瞬間には、それらのノードはまだ実行されていません。実行されると、その結果がこのチェックポイントに付随します。成功したノードはその戻り値を result に残し、失敗したノードはその例外を error に残します。
• interrupts — グラフが人間に制御を戻すために一時停止した箇所です。LangGraphは実行の途中で止まって、誰かがステップを承認したり値を提供したりするのを待つことができ、このフィールドがそれらの一時停止を記録します。

フィールドは属性として読みます。metadata は辞書なので、キーを使って値を取り出します。

snapshot = graph.get_state(config)
 
print(snapshot.values)            # {'foo': 'b', 'bar': ['a', 'b']}
print(snapshot.next)              # ()
print(snapshot.metadata["step"])  # 2

この中で、デバッグ中に最もよく手を伸ばすのが next です。next が空でなければ、グラフは最後まで到達しなかった——途中のどこかで止まったということです。そして17.3で失敗したグラフを再開するとき、このフィールドが出発点になります。

17.2.2) チェックポイント履歴をたどる

get_state() は最新のチェックポイントだけを見せてくれます。しかしデバッグでは、しばしば「どうやってここに至ったのか?」を問うことになります——そのためには、実行の全軌跡が必要です。get_state_history() がそれを提供してくれます。

for snap in graph.get_state_history(config):
    print(f"step={snap.metadata['step']:>2}  "
          f"next={str(snap.next):<16}  values={snap.values}")

出力:

step= 2  next=()                values={'foo': 'b', 'bar': ['a', 'b']}
step= 1  next=('node_b',)       values={'foo': 'a', 'bar': ['a']}
step= 0  next=('node_a',)       values={'foo': '', 'bar': []}
step=-1  next=('__start__',)    values={'bar': []}

最新のチェックポイントが最初に来るので、実行を順番にたどるには下から上へ進みます。

• step -1 — invoke() が入力を受け取った直後です。渡した {"foo": "", "bar": []} が values に現れないことに注目してください。入力を状態に取り込むこと自体が1つのステージであり、そのステージはまだ実行されていないのです。next の __start__ がそれを行う内部ノードです。

  bar が [] として現れていますが、それは私たちが渡した値ではありません。リデューサーを持つフィールドは、書き込みが蓄積されていくための空の値から始まります。foo にはリデューサーがないので、開始値がまったくありません——だからここには現れないのです。

• step 0 — __start__ が実行され、入力が状態に取り込まれました。foo='' と bar=[] が私たちが渡した値です。次は node_a です。
• step 1 — node_a を実行した結果です。foo は今 'a' で、bar は ['a'] になり、次は node_b です。
• step 2 — node_b を実行した結果です。next が空なので、グラフは完了しています。

step 0やstep 1で next が空でないことは、グラフがそこで止まったことを意味しません。グラフがまだ実行中に取られたチェックポイントは、当然のことながら次に実行されるノードが並んでいます。17.2.1で「next が空でなければグラフが止まった」と述べたのは、何かがうまくいかなかった時点でのチェックポイントについて話していたのです。履歴の途中では、next は単にグラフがどの経路をたどったかを見せてくれるだけです。

17.3) 失敗した箇所から再開する

状態はすべてのスーパーステップの終わりに保存されるので、実行途中の失敗が完了済みの作業を巻き添えにすることはありません——それはチェックポイントの中にまだ残っています。最初からやり直す理由はありません。止まった箇所から再開すればよいのです。

17.3.1) invoke(None, config) で再開する

再開は簡単です。入力が入る場所に None を渡します。

graph.invoke(None, config)

これは「新しい入力はない、保存された状態から続けよ」という意味です。もちろん config には thread_id がやはり必要です。LangGraphはどの会話を続けるかを知らなければならないからです。

失敗を仕込んで、そこから再開してみましょう。2つのノードから成るグラフを作り、その2番目のノードが最初の実行のときだけ失敗するようにします。リトライでは成功しなければなりません。そうでないと、再開が実際に機能しているところを目にすることができないからです。

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    step_1_done: bool
    step_2_done: bool
 
first_try = True   # 最初の実行を失敗させるためのフラグ
 
def step_1(state: State):
    print("step_1 running (expensive work)")
    return {"step_1_done": True}
 
def step_2(state: State):
    global first_try
    if first_try:
        first_try = False
        print("step_2 failed (API timeout)")
        raise RuntimeError("External API timed out")
    print("step_2 running")
    return {"step_2_done": True}
 
builder = StateGraph(State)
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "job-42"}}
 
try:
    graph.invoke({"step_1_done": False, "step_2_done": False}, config)
except RuntimeError as e:
    print("Failed:", e)

出力:

step_1 running (expensive work)
step_2 failed (API timeout)
Failed: External API timed out

step_1 は成功し、step_2 は例外を発生させました。17.2で学んだ get_state() を使って、グラフがどこで止まったかを確認しましょう。

snapshot = graph.get_state(config)
print("next   =", snapshot.next)
print("values =", snapshot.values)

出力:

next   = ('step_2',)
values = {'step_1_done': True, 'step_2_done': False}

next は ('step_2',) で、グラフが step_2 の実行途中で止まったことを教えてくれます。そして values では、step_1_done が True になっています——step_1 の結果はチェックポイントの中にまだ残っているのです。

では None で再開しましょう。

result = graph.invoke(None, config)
print("Final =", result)

出力:

step_2 running
Final = {'step_1_done': True, 'step_2_done': True}

step_1 running (expensive work) は一度も表示されませんでした。 step_1 は2度目の実行をしなかったのです。LangGraphは保存された状態を読み込み、step_2 から再開しました。あの高価な最初のステップの代金を2度払わずに済んだのです。

17.3.2) 再開するときの注意点

再開すると、失敗したノードが再び実行されます。そのノードがLLMを呼び出したり外部APIにアクセスしたりする場合、それらの呼び出しも再び発生します——そして違うものが返ってくるかもしれません。

そこに落とし穴があります。もし step_2 がメールを送信し、その後で失敗したのなら、再開すると2通目のメールが送られます。LangGraphが保証するのは、成功したノードを再実行しないことだけです。

ですから、再び実行される可能性のあるノードは冪等(idempotent)でなければなりません。同じことを2回行っても、同じ場所にいられるようにするのです。メールを送る前にすでに送信済みかどうかをチェックする、重複した挿入が起こらないようにデータベースのテーブルに一意なキーを付ける、といった具合です。