LangGraphの使い方｜checkpointerとinterruptで再開・承認

この記事で実際に動かしたもの：langgraph 1.2.19 のグラフ（LLM の代わりに固定応答のチャットモデル）、SqliteSaver と PostgresSaver、kill -9 からの再開、interrupt と Command(resume)、get_state_history・update_state、7種類のストリーミングのモード。

flowchart TD
    S([START]) --> A[agent<br>モデルを呼ぶ]
    A -->|tool_calls あり| P[approval<br>interrupt で止まる]
    A -->|tool_calls なし| E([END])
    P -->|承認・修正| T[tools<br>ToolNode]
    P -->|却下| A
    T --> A

この記事で作る「送信の前に人が承認するエージェント」の形です。よく調べられることの答えを先にまとめます。

知りたいこと	答え	詳しい節
LangGraph とは・LangChain との違い	LangChain はモデル・ツールの部品、LangGraph は状態を持って動かす実行基盤	LangGraphとは
LangGraph の使い方（最小の形）	State を TypedDict で書き、add_node・add_edge・compile・invoke	最小のエージェント
ループが止まらない	recursion_limit の既定は 10,007 ステップ。トップレベルのキーで小さく指定する	ループの上限
langgraph checkpointer の選び方	開発は InMemorySaver か SqliteSaver、本番は PostgresSaver（setup() を先に）	checkpointer と thread_id
落ちたプロセスの続きから再開	同じ thread_id で invoke(None, config)。落ちたノードは最初からやり直し	kill -9 から再開
langgraph interrupt で人の承認	interrupt(値) で止め、Command(resume=値) で再開。ノードは先頭から再実行	interrupt で人の承認
過去の状態に戻してやり直す	get_state_history で checkpoint_id を選び、invoke(None, その config)	タイムトラベル
stream_mode はどれを使う	画面の文字は messages、進捗は custom、状態の差分は updates	ストリーミングのモード比較

確認環境：Ubuntu 24.04.5 LTS（2 CPU・メモリ 8GB、ほかの作業と共有のマシン）の上で、Python 3.13.18 の venv に langgraph 1.2.19・langgraph-checkpoint 4.3.1・langgraph-checkpoint-sqlite 3.1.3・langgraph-checkpoint-postgres 3.2.0・langgraph-prebuilt 1.1.1・langchain-core 1.7.4・psycopg 3.3.7 を PyPI から入れて実行しました。PostgreSQL 16.15 は作業場所に initdb したクラスタを 127.0.0.1 だけで起動し、終わったら止めて消しています。LLM と LangSmith には接続していません（API キー無し・LANGSMITH_TRACING=false）。モデルは langchain-core の BaseChatModel を継承した固定応答のモデルで、最後のメッセージを見てツール呼び出しか最終回答を返します。そのため、この記事は LLM の回答の質には触れず、グラフの状態遷移・保存・再開だけを実物で示します。interrupt・durability・ストリーミング・recursion limit の仕様は LangChain 公式ドキュメント（docs.langchain.com）、LangSmith Deployment の名称は langchain.com で確かめました。図は mermaid 11.16.0 で描画を確かめています（確認日 2026年11月4日）。

LangGraphとは（LangChainとの関係と向く場面）

この節では、LangGraph が何を受け持つ部品で、LangChain や LangSmith とどう分かれているかが分かります。

LangGraph は、長く動き、状態を持つエージェントのための低レベルのオーケストレーション（処理の段取り）の基盤です。 公式の概要ページは、LangChain を「エージェントのフレームワーク（モデル・ツール・エージェントループの抽象化と連携）」、LangGraph を「オーケストレーションのランタイム」と位置づけ、LangGraph は LangChain が無くても単独で動くと説明しています。プロンプトやエージェントの構成は決めず、決まった処理と LLM に任せる処理を同じグラフに混ぜられるのが特徴です。

LangChain・LangSmith・LangSmith Deployment の分け方

名前	受け持つこと	この記事での扱い
LangChain	モデル・ツール・メッセージの共通の形、create_agent などの高レベルの部品	langchain-core のメッセージとツールだけ使用
LangGraph	状態・分岐・ループ・保存・再開・中断の実行基盤（MIT ライセンスの OSS）	全部の実験の本体
LangSmith（Observability）	トレース・評価・監視の SaaS	接続していない
LangSmith Deployment	ステートフルなエージェントのデプロイ。Studio（画面で見る IDE）を含む	公式の説明の範囲だけ

langchain.com の FAQ には「LangGraph Platform は 2025年10月に LangSmith Deployment に名称が変わった」とあります。以前の記事で「LangGraph Platform」「LangGraph Cloud」と書かれているものは、いまの LangSmith Deployment のことです。公式のチェックポイントのページは、Agent Server を使う場合は checkpointer を自分で設定しなくても保存が自動で行われる、と説明しています。この記事は、自分のプロセスの中で LangGraph を動かす形（OSS の部分）だけを扱います。

エージェントの型（ReAct・計画と実行・ルーター・評価ループなど）をどう選ぶかは、関連記事「AIエージェントの設計パターン一覧｜5つの型の違いと選び方」で扱っています。この記事は、その型を LangGraph のグラフに落とし、止めて再開できる形にする部分を受け持ちます。

LangGraphの使い方：最小のエージェント（StateGraph・ToolNode・偽のモデル）

この節では、LangGraph のインストールから、ツールを呼ぶ最小のエージェントを動かすまでと、LLM 無しで試すための偽のモデルの作り方が分かります。

インストールは pip install -U langgraph です。 チェックポイントを SQLite・PostgreSQL に保存するなら、別のパッケージを足します。

python3 -m venv .venv && . .venv/bin/activate
pip install -U "langgraph>=1.2,<1.3" langgraph-checkpoint-sqlite \
    langgraph-checkpoint-postgres "psycopg[binary]"
pip list | grep -E "langgraph|langchain-core"

langchain-core                1.7.4
langgraph                     1.2.19
langgraph-checkpoint          4.3.1
langgraph-checkpoint-postgres 3.2.0
langgraph-checkpoint-sqlite   3.1.3
langgraph-prebuilt            1.1.1
langgraph-sdk                 0.4.12

langgraph を入れると、langchain-core・langgraph-prebuilt（ToolNode など）・langgraph-checkpoint（InMemorySaver）が一緒に入りました。venv の大きさは pip のキャッシュを除いて 126MB でした。

LLM 無しで試す：固定応答のチャットモデル

langchain-core には GenericFakeChatModel があり、渡した AIMessage を順に返します。ツール呼び出しを含む AIMessage も返せましたが、手元では bind_tools() が NotImplementedError になり、用意した応答を使い切ると StopIteration になりました。再開やタイムトラベルで同じノードがもう一度動くと「次の応答」がずれるので、この記事では最後のメッセージを見て応答を決めるモデルを自作しました。

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatResult

class ScriptedToolModel(BaseChatModel):
    always_call_tool: bool = False   # True ならツールを呼び続ける（ループ確認用）

    @property
    def _llm_type(self) -> str:
        return "scripted-tool-model"

    def bind_tools(self, tools, **kwargs):
        return self

    def _decide(self, messages):
        last = messages[-1]
        n = sum(isinstance(m, AIMessage) for m in messages) + 1
        if self.always_call_tool or isinstance(last, HumanMessage):
            return AIMessage(content="", id=f"ai-{n}", tool_calls=[
                {"name": "get_weather", "id": f"call_{n}", "args": {"city": "東京"}}])
        if isinstance(last, ToolMessage):
            return AIMessage(content=f"結果：{last.content}", id=f"ai-{n}")
        return AIMessage(content="了解しました。", id=f"ai-{n}")

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        return ChatResult(generations=[ChatGeneration(message=self._decide(messages))])

ストリーミングを確かめるために、実際には _stream も書き、最終回答を3文字ずつ返すようにしています。本物のモデルに替えるときは、ScriptedToolModel() を各社のチャットモデルのクラスに置き換えるだけで、グラフ側は変わりません。

StateGraph・条件分岐・ToolNode

from typing import Annotated, TypedDict
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

@tool
def get_weather(city: str) -> str:
    """都市の天気を返す"""
    return f"{city}は晴れ、22度"

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 追記する reducer

model = ScriptedToolModel().bind_tools([get_weather])

def agent(state: State):
    return {"messages": [model.invoke(state["messages"])]}

b = StateGraph(State)
b.add_node("agent", agent)
b.add_node("tools", ToolNode([get_weather]))
b.add_edge(START, "agent")
b.add_conditional_edges("agent", tools_condition)  # tool_calls があれば tools、無ければ END
b.add_edge("tools", "agent")
graph = b.compile()

out = graph.invoke({"messages": [HumanMessage("東京の天気は？")]})

HumanMessage '東京の天気は？'
AIMessage '' [{'name': 'get_weather', 'args': {'city': '東京'}, 'id': 'call_1', 'type': 'tool_call'}]
ToolMessage '東京は晴れ、22度'
AIMessage '結果：東京は晴れ、22度'

graph.get_graph().draw_mermaid() は、点線（条件分岐）と実線（固定のエッジ）を分けた Mermaid の文字列を返しました。agent -.-> tools・agent -.-> __end__・tools --> agent の3本で、ToolNode には登録したツール名が添えられます。設計レビューでグラフの形を見せるときに使えます。

LangChain 側の部品（プロンプト・チェーン・出力パーサー）の書き方は、関連記事「LangChain入門｜PythonでLLMアプリを作る基本とコピペ集」で扱っています。

条件分岐とループの上限：recursion_limit の既定値とエラー原文

この節では、エージェントのループが止まらないときに何が起きるか、recursion_limit の数え方と既定値が分かります。

LangGraph 1.2.19 の既定の上限は 10,007 super-step でした。 公式の Graph API のページも、既定値は 1.1.4 以降 10,007、1.0.6〜1.1.3 は 10,000 と書いています。以前の版の解説で見かける「25」は、いまの既定値ではありません。モデルが毎回ツールを呼ぶ設定にして、上限を変えずに動かしました。

default: Recursion limit of 10007 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key. model calls=5004 89.4s

偽のモデルでも、メッセージが1万件まで増えて 89.4 秒かかりました。本物の LLM では、1回ごとに入力のトークンが増えるので、時間と料金はこれより大きくなります。本番のエージェントでは recursion_limit を必ず小さく指定します。

指定する場所と数え方

graph.invoke(inp, {"recursion_limit": 6})                    # 効く
graph.invoke(inp, {"configurable": {"recursion_limit": 3}})  # 効かない

limit=6: GraphRecursionError | Recursion limit of 6 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/GRAPH_RECURSION_LIMIT
configurable 内: Recursion limit of 10007 reached without hitting a stop condition. ...

configurable の中に書いた 3 は無視され、既定の 10,007 まで回りました。公式も、上限は configurable ではなくトップレベルのキーで指定すると書いています。数え方も確かめました。最初の例（agent→tools→agent の3回のノード実行）は、recursion_limit=3 では GraphRecursionError、4 で通りました。「動かしたいノードの回数＋1」を目安に、余裕を持たせて決めます。

上限の手前で自分から止める（RemainingSteps）

エラーで終わると、ユーザーには何も返りません。State に RemainingSteps を置くと、残りのステップ数をノードの中で読めます。

from langgraph.managed import RemainingSteps

class State(TypedDict):
    messages: Annotated[list, add_messages]
    remaining_steps: RemainingSteps

def agent_guard(state):
    if state["remaining_steps"] <= 2:
        return {"messages": [AIMessage("ステップの上限に近いので打ち切ります。")]}
    return {"messages": [loop_model.invoke(state["messages"])]}

guard: 10 messages; last = ステップの上限に近いので打ち切ります。

recursion_limit=10 で、例外ではなく打ち切りのメッセージで終わりました。無限ループの防ぎ方は、ほかにも「ツール呼び出しの回数を State に数えて条件分岐で抜ける」方法があります。どこで打ち切るかは、タスクの粒度で決まります。粒度の決め方は関連記事「AIエージェントのタスク分解｜指示書テンプレートと粒度の判断表」で扱っています。

State と reducer：並列ノードで同じキーに書くとき（Send を含む）

この節では、2つのノードを同時に動かしたときに State がどう合わさるかと、reducer を付け忘れたときのエラーが分かります。

START から2本のエッジを出すと、2つのノードは同じ super-step で並列に動きます。 同じキーに書き込むとき、reducer が無いとエラーになりました。

class S(TypedDict):
    results: list[str]                                # reducer 無し
#   results: Annotated[list[str], operator.add]       # reducer あり

g.add_edge(START, "a"); g.add_edge(START, "b")
g.add_edge(["a", "b"], "join")                        # 両方が終わってから join

langgraph.errors InvalidUpdateError | At key 'results': Can receive only one value per step. Use an Annotated key to handle multiple values.
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/INVALID_CONCURRENT_GRAPH_UPDATE

書き方	結果	時間（各ノード 0.5 秒）
reducer 無し	InvalidUpdateError	―
Annotated[list, operator.add]	{'results': ['A', 'B']}	0.51 秒（並列に動いた）
Send で4件に分ける（各 0.3 秒）	['X', 'Y', 'Z', 'W']	0.32 秒

件数が実行時に決まるとき（検索結果ごとに要約する、など）は、条件付きエッジから Send("work", {"item": x}) のリストを返します。公式は「通常のエッジではこれはできない」と書いています。

def fan(s):
    return [Send("work", {"item": x}) for x in s["items"]]

g.add_conditional_edges(START, fan, ["work"])

公式も、operator.add は追記だけなので human-in-the-loop で中身を編集する用途には向かない、と書いています。後の「interrupt で人の承認」の節で、同じ id の AIMessage を送り直して引数を書き換える例を動かします。reducer を通さずに値を丸ごと入れ替えたいときは Overwrite を使う、と公式にあります。

langgraph checkpointer の使い方：thread_id とチェックポイントの中身

この節では、checkpointer を付けると何がどこに保存されるのか、種類ごとの違いと、PostgreSQL で最初に要る準備が分かります。

checkpointer は compile(checkpointer=…) に渡し、実行のたびに {"configurable": {"thread_id": "…"}} を指定します。 thread_id が「どの会話・どのジョブの続きか」を表します。公式は thread_id を「永続的なカーソル」と表現しています。

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

cp = SqliteSaver(sqlite3.connect("agent.db", check_same_thread=False))
g = b.compile(checkpointer=cp)
cfg = {"configurable": {"thread_id": "h1"}}
g.invoke({"messages": [HumanMessage("東京の天気は？")]}, cfg)

1回の実行（agent→tools→agent）で、SQLite には次のものが残りました。

tables: ['checkpoints', 'writes']
件数 5
step= 3 source=loop   next=()             msgs=4
step= 2 source=loop   next=('agent',)     msgs=3
step= 1 source=loop   next=('tools',)     msgs=2
step= 0 source=loop   next=('agent',)     msgs=1
step=-1 source=input  next=('__start__',) msgs=0

入力を受け取った時点（step -1）と、各 super-step の後に1件ずつ、合わせて5件です。next は「次に動くノード」で、再開するときはここから続きます。get_state() が返す StateSnapshot のフィールドは values・next・config・metadata・created_at・parent_config・tasks・interrupts でした。

checkpointer の種類

クラス	パッケージ	保存先	向く用途
InMemorySaver	langgraph-checkpoint（同梱）	プロセスのメモリ	テスト・試作。再起動で消える
SqliteSaver / AsyncSqliteSaver	langgraph-checkpoint-sqlite	SQLite のファイル	手元の開発・1台で動く小さなもの
PostgresSaver / AsyncPostgresSaver	langgraph-checkpoint-postgres	PostgreSQL	本番・複数プロセス
MongoDB・Azure Cosmos DB 向けなど	それぞれ別パッケージ	各 DB	公式の一覧の範囲

PostgresSaver は setup() を先に呼ぶ

自分で initdb した PostgreSQL 16.15 で、setup() を呼ぶ前に使うと次のエラーになりました。

setup() 前: UndefinedTable relation "checkpoints" does not exist

from psycopg import Connection
from langgraph.checkpoint.postgres import PostgresSaver

DSN = "host=127.0.0.1 port=55247 user=t247 dbname=t247db"
with Connection.connect(DSN, autocommit=True) as conn:
    PostgresSaver(conn).setup()     # テーブルとマイグレーションを作る（何度呼んでもよい）

tables: ['checkpoint_blobs', 'checkpoint_migrations', 'checkpoint_writes', 'checkpoints']
migrations: 10

接続は autocommit=True にします。50ステップのグラフを1回動かしたスレッドでは、checkpoints が52行、checkpoint_blobs が52行、checkpoint_writes が152行になりました。公式は、長い会話ではチェックポイントが増え続けるので、以前のものを定期的に消すか保持期間を決めるよう勧めています。 SQLite では1件あたり平均 1,364 バイト（メッセージ4件までの短い会話）でした。thread_id は 255 文字以内にする、という注意も公式の Postgres の説明にあります。

途中で kill -9 しても続きから再開できるか（SQLite・PostgreSQL）

この節では、プロセスが落ちたときにどこから再開されるのか、落ちたノードはどう扱われるのかが、実際に kill -9 した記録で分かります。

plan→fetch（3秒かかる）→write の3ノードのグラフを別プロセスで動かし、fetch の途中（開始から 2.2 秒）で kill -9 しました。 新しいプロセスで同じ thread_id を指定し、入力に None を渡すと続きから動きます。

cfg = {"configurable": {"thread_id": "job-1"}}
g.invoke({"done": []}, cfg, durability="sync")   # 1つ目のプロセス（途中で kill -9）
g.invoke(None, cfg)                              # 2つ目のプロセス：続きから

[sync] kill -9 9605
exit=137
[sync] state:
values= {'done': ['plan']} next= ('fetch',) step= 1
[sync] resume:
{'done': ['plan', 'fetch', 'write']}
[sync] log:
pid=9605 --- run durability=sync
pid=9605 plan 実行
pid=9605 fetch 開始（3秒かかる）
pid=9748 --- resume
pid=9748 fetch 開始（3秒かかる）
pid=9748 fetch 終了
pid=9748 write 実行

再開の単位はノードです。 fetch の中で外部 API に書き込み（決済・メール送信・チケット作成）をしていたら、再開でもう一度書き込まれます。ノードの中の副作用は、冪等キー（同じ要求を2回受けても1回分だけ処理する仕組み）で守るか、ノードを細かく分けて「書き込みだけのノード」にします。冪等キーの作り方は、関連記事「冪等性とは？二重課金を防ぐ冪等キーとリトライ設計」で扱っています。

同じ super-step の片方が失敗したとき（pending writes）

ok と flaky の2ノードを並列に動かし、flaky だけ例外を出しました。

1回目 例外: 外部APIが 503 を返した {'ok': 1, 'flaky': 1}
next: ('flaky',) tasks: [('ok', None), ('flaky', "RuntimeError('外部APIが 503 を返した')")]
再開: {'results': ['flaky', 'ok']} {'ok': 1, 'flaky': 2}

再開で動いたのは flaky だけで、ok の呼び出し回数は1回のままでした。公式が pending writes と呼ぶ仕組みで、同じ super-step で成功したノードの書き込みは保存され、再開で繰り返されません。ただし、結果のリストの順番は ['flaky', 'ok'] で、通常の実行と違いました。operator.add で集めた並列の結果に、順番の意味を持たせないようにします。

一時的な失敗は RetryPolicy で

from langgraph.types import RetryPolicy
g.add_node("f", flaky2, retry_policy=RetryPolicy(max_attempts=3, initial_interval=0.1))

RetryPolicy: {'results': ['3回目で成功']} 0.88s

2回 ConnectionError を出して3回目で成功するノードが、例外を外に出さずに終わりました。再試行はノードの中で完結し、checkpointer が無くても動きます。プロセスごと落ちる失敗は checkpointer、ノードの中の一時的な失敗は RetryPolicy、と分けて考えます。

durability（exit・async・sync）の違いと速度の実測

この節では、チェックポイントを書き込むタイミングの3つのモードで、落ちたときの結果と速さがどう変わるかが分かります。

公式によると、exit は実行の終わりにだけ保存（最も速いが途中のクラッシュからは戻れない）、async は次のステップと並行して保存（落ちると保存が失われる小さなリスクがある）、sync は次のステップの前に保存（最も確実だが遅い）です。invoke(…, durability="sync") のように指定します。

kill -9 したときの違い

durability	kill 後の get_state()	invoke(None) で再開
sync	values={'done': ['plan']} next=('fetch',)	fetch から再開して完了
async	values={'done': ['plan']} next=('fetch',)	fetch から再開して完了
exit	values={} next=()（何も残っていない）	EmptyInputError: Received no input for __start__

exit では途中の状態が1件も残らず、invoke(None) は「入力が無い」というエラーになりました。最初からやり直すしかありません。async は今回の条件（fetch に 3 秒かかり、その間に保存が終わっている）では sync と同じ結果でしたが、保存の途中で落ちたときは失われ得る、というのが公式の説明です。

速さ（50ステップのグラフ・3回測った中央値）

保存先	exit	async	sync
checkpointer 無し	13.4ms	―	―
InMemorySaver	14.8ms	33.4ms	28.6ms
SqliteSaver（tmpfs 上のファイル）	34.1ms	37.3ms	44.2ms
SqliteSaver（ディスク上のファイル）	40.3ms	246ms	133ms
PostgresSaver（127.0.0.1）	40.7ms	89.7ms	91.7ms

1ステップで State に約 200 バイトの文字列を1件足すグラフを50回まわしました。ほかの作業と共有の2コアのマシンなので値は回ごとに揺れ（例：SQLite のディスク上・async は 201〜335ms）、順位の目安として読みます。ディスク上の SQLite では async が sync より遅くなりました。 1本の SQLite 接続を背景のスレッドと取り合う形になるためと考えられ、async が常に速いとは限りません。1ステップあたりに直すと、PostgreSQL の sync でも約 1.8ms で、LLM の1回の呼び出し（数百ミリ秒〜数秒）に比べれば小さい値です。LLM を呼ぶエージェントでは sync を基本にし、速さが要る短いバッチ処理だけ exit を検討します。

langgraph interrupt で人の承認を挟む（Command(resume=…)）

この節では、ツールを実行する前に人の承認を待ち、承認・却下・引数の修正のどれでも続きを動かす実装と、別のプロセスから再開した記録が分かります。

承認用のノードで interrupt(値) を呼ぶと、その値が呼び出し元に返り、グラフは止まります。 人の判断を受け取ったら、同じ thread_id に Command(resume=判断) を渡すと、interrupt() の戻り値としてノードに届きます。

from langgraph.types import interrupt, Command

def approval(state) -> Command:
    call = state["messages"][-1].tool_calls[0]
    decision = interrupt({"question": "このツールを実行してよいですか",
                          "tool": call["name"], "args": call["args"]})
    if decision["action"] == "approve":
        return Command(goto="tools")
    if decision["action"] == "edit":
        msg = state["messages"][-1].model_copy(deep=True)
        msg.tool_calls[0]["args"].update(decision["args"])
        return Command(goto="tools", update={"messages": [msg]})  # 同じ id なので置き換わる
    return Command(goto="agent", update={"messages": [ToolMessage(
        content=f"人が却下しました：{decision.get('reason', '')}", tool_call_id=call["id"])]})

def route(state):
    return "approval" if state["messages"][-1].tool_calls else END

b.add_node("approval", approval, destinations=("tools", "agent"))
b.add_conditional_edges("agent", route, ["approval", END])

destinations は、Command で移る先を図に描くための指定です。止めるプロセスと再開するプロセスは別にしました（SqliteSaver）。止まっている間、プロセスは残しておかなくてよい作りです。

$ python ex4_interrupt.py start hitl.db
keys: ['messages', '__interrupt__']
__interrupt__: [Interrupt(value={'question': 'このツールを実行してよいですか', 'tool': 'send_email', 'args': {'to': 'team@example.com', 'subject': '週報', 'body': '今週の進捗です'}}, id='db4355271fbee55aa2d3bfa58eb46b9f', response_schema=None)]

$ python ex4_interrupt.py show hitl.db
next: ('approval',)
tasks: [('approval', [{'question': 'このツールを実行してよいですか', 'tool': 'send_email', ...}])]

resume に渡した値	最後のメッセージ	送信の記録
{"action": "approve"}	結果：team@example.com に「週報」を送信しました	team@example.com に1件
{"action": "reject", "reason": "宛先が違う"}	結果：人が却下しました：宛先が違う	0件
{"action": "edit", "args": {"to": "me@example.com"}}	結果：me@example.com に「週報」を送信しました	me@example.com に1件

edit では、AIMessage の履歴そのものが 'to': 'me@example.com' に書き換わっていました。add_messages が同じ id のメッセージを置き換えたためです。却下のときは、tool_call_id を合わせた ToolMessage を返して agent に戻しています。tool_calls を持つ AIMessage の後に対応する ToolMessage が欠けた履歴は、モデルの API でエラーになることがあるので、却下も ToolMessage で返します。

承認の流れを組む順番

どこまでを決まった手順（ワークフロー）にし、どこからをモデルの判断（エージェント）に任せるかの分け方は、関連記事「AIエージェントとは｜仕組みとワークフローとの違い・作り方入門」で扱っています。承認ノードは、その境目に置くのが基本です。静的な止め方（interrupt_before・interrupt_after）もありますが、公式は「人の入力を待つ用途には勧めない（デバッグ向け）」と書いています。

interrupt の落とし穴：再実行・try/except・同時再開

この節では、interrupt を使うときに手元で実際に起きた5つのつまずきと、その避け方が分かります。

1. 再開するとノードは先頭からもう一度動く

承認ノードの先頭でファイルに1行書くようにしたところ、止めたときと再開したときで2回書かれました。

pid=11461 approval ノードに入った
pid=11540 approval ノードに入った

公式も、再開したときは interrupt を呼んだノードの先頭から動き直すと書いています。interrupt より前にある副作用（DB への書き込み・通知）は2回動きます。 冪等にするか、interrupt の後ろへ移します。

2. try/except Exception で囲むと止まらない

try:
    ans = interrupt("承認しますか")
except Exception as e:
    return {"v": f"except で握りつぶした: {type(e).__name__}"}

6 try/except Exception で囲む -> {'v': 'except で握りつぶした: GraphInterrupt'}
GraphInterrupt の継承: ['GraphInterrupt', 'GraphBubbleUp', 'Exception', 'BaseException', 'object']

interrupt は内部で GraphInterrupt という例外を投げて止めます。これは Exception の子なので、except Exception で捕まえると止まらずに次へ進みました。公式も「interrupt を try/except で囲まない」と書いています。ノード全体を広い except で囲む書き方は、承認の抜け道になります。

3. 同じ thread を2か所から同時に再開すると、ツールが2回動いた

止まっている thread に、2つのプロセスからほぼ同時に Command(resume={"action": "approve"}) を送りました（送信のツールに2秒かかる設定）。

outbox:
{"to": "team@example.com", "subject": "週報"}
{"to": "team@example.com", "subject": "週報"}

承認は1回なのに、メールの送信は2回記録されました。 手元の SqliteSaver では thread ごとの排他は行われず、承認ボタンの二度押しや、ワーカーの再配送で同じことが起きます。承認の受付を DB の行ロックや「承認済み」フラグで1回に絞るか、送信のツール自体を冪等キーで守ります。

4. Command(resume) は待ちが無くてもエラーにならない

3 別の thread_id で resume -> {'__interrupt__': [Interrupt(value='承認しますか', id='45c7…', ...)]}
5 もう一度 resume（待ちが無い） -> {'v': 'answer=yes'}

thread_id を打ち間違えて resume すると、エラーにならず、新しい thread として最初から動いて止まりました。承認済みの thread にもう一度 resume しても、何も起きずに最後の状態が返りました。resume の前に get_state(cfg).interrupts で、待ちがあることと、その interrupt の id を確かめると取り違えを防げます。

5. checkpointer が無いと、止まるが再開できない

1 checkpointer なしで interrupt -> {'v': '', '__interrupt__': [Interrupt(value='承認しますか', ...)]}
resume without cp: RuntimeError Cannot use Command(resume=...) without checkpointer
1b thread_id なし -> 例外 ValueError : Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

checkpointer を付け忘れても、止めるところまでは例外が出ませんでした。再開しようとして初めて RuntimeError になります。テストで「止まった」ことだけを確かめていると気づきにくいので、テストには再開まで含めます。

• 1interrupt より前に副作用を置いていない再開でノードが先頭から動くため
• 2interrupt を except Exception で囲んでいないGraphInterrupt が捕まり、止まらずに進む
• 31つのノードの interrupt の順番が毎回同じ公式は、再開の値を呼び出しの順番で対応させると書いている
• 4同じ thread の再開を1か所に絞っている同時の resume でツールが2回動いた
• 5テストで resume まで動かしているcheckpointer 無しでも止まるところまでは通る

状態の履歴 get_state_history とタイムトラベル（replay・update_state）

この節では、保存された履歴を見る方法と、過去の時点からやり直す・値を変えて分岐させる方法が、チェックポイントの親子関係の出力で分かります。

get_state_history(cfg) は、その thread のチェックポイントを新しい順に返します。 各チェックポイントには自分の id と親の id があり、木の形でつながっています。

replay：ツールを実行する直前からやり直す

hist = list(g.get_state_history(cfg))
before_tools = next(s for s in hist if s.next == ("tools",))
out = g.invoke(None, before_tools.config)     # config に checkpoint_id が入っている

replay 後のメッセージ数 4 モデル呼び出し 1

指定した時点より前のノードは動かず、tools と、その後の agent（モデル呼び出し1回）がもう一度動きました。公式も、再生ではそれ以降のノードが再実行され、LLM の呼び出しや interrupt ももう一度起きると書いています。replay はキャッシュの読み直しではなく再実行です。 外部への書き込みのツールがある区間を再生すると、もう一度書き込まれます。

update_state：値を変えて分岐させる

ai = before_tools.values["messages"][-1].model_copy(deep=True)
ai.tool_calls[0]["args"]["city"] = "大阪"
new_cfg = g.update_state(before_tools.config, {"messages": [ai]})
out2 = g.invoke(None, new_cfg)

fork の結果: 結果：大阪は晴れ、22度
履歴の件数（分岐後） 11
  step= 4 source=loop   next=()           id=…8c462789 parent=…abd2b403
  step= 3 source=loop   next=('agent',)   id=…abd2b403 parent=…033b557d
  step= 2 source=update next=('tools',)   id=…033b557d parent=…e87c3555
  step= 4 source=loop   next=()           id=…ffb81282 parent=…b6992f07
  step= 3 source=loop   next=('agent',)   id=…b6992f07 parent=…7e765ed7
  step= 2 source=fork   next=('tools',)   id=…7e765ed7 parent=…e87c3555

元の5件は消えず、replay の分岐（source=fork）と update_state の分岐（source=update）が、どちらも同じ親（…e87c3555、ツール実行の直前）から生えました。get_state(cfg) は最後に書かれた枝（大阪）を返します。公式も、update_state は元のチェックポイントを変えずに新しいチェックポイントを作り、reducer も通すと書いています。

使いどころは、障害の調査（どの step で State がおかしくなったかを next と values で追う）、同じ時点からのプロンプトの比較、人による引数の訂正の3つです。

ストリーミングのモード比較（values・updates・messages・custom ほか）

この節では、stream_mode の7種類で何が流れてくるか、同じ実行で件数とバイト数がどう違うかが分かります。

agent→tools→agent の同じ実行を、モードを変えて流しました（バイト数は各チャンクを JSON 文字列にした長さの合計）。

stream_mode	流れるもの（公式の説明）	件数	バイト数
values	各ステップの後の State 全体	4	1,360
updates	各ステップでノードが返した差分	3	481
messages	LLM のトークンとメタデータの組	8	4,637
custom	ノードが get_stream_writer で送った値	0（書くノードが無い）	0
checkpoints	チェックポイント（checkpointer が要る）	5	3,467
tasks	ノードの開始・終了と結果（checkpointer が要る）	6	1,928
debug	checkpoints と tasks をまとめたもの	11	6,426

messages モードでは、最終回答が3文字ずつ届き、どのノードのどの step から来たかがメタデータで分かりました。ツールの結果（ToolMessage）も流れてきます。

'' AIMessageChunk agent 1
'東京は晴れ、22度' ToolMessage tools 2
'結果：' AIMessageChunk agent 3
'東京は' AIMessageChunk agent 3
'晴れ、' AIMessageChunk agent 3
'22度' AIMessageChunk agent 3

進捗を送る custom と、形がそろう version="v2"

from langgraph.config import get_stream_writer

def tools_with_progress(state):
    w = get_stream_writer()
    w({"progress": "天気APIに問い合わせ中", "pct": 50})
    out = ToolNode([get_weather]).invoke(state)
    w({"progress": "完了", "pct": 100})
    return out

for part in g.stream(inp, stream_mode=["updates", "custom"], version="v2"):
    print(part["type"], part["ns"], part["data"])

updates () {'agent': {'messages': [AIMessage(content='', ...
custom () {'progress': '天気APIに問い合わせ中', 'pct': 50}
custom () {'progress': '完了', 'pct': 100}
updates () {'tools': {'messages': [ToolMessage(content='東京は晴れ、22度', ...
updates () {'agent': {'messages': [AIMessage(content='結果：東京は晴れ、22度', ...

既定の v1 で複数のモードを指定すると ('updates', {...}) のタプルが返り、モードの数やサブグラフの有無で形が変わります。version="v2"（1.1 以降）では、どの場合も type・ns・data の辞書でした。新しく書くコードは v2 にそろえると分岐が簡単になります。さらに 1.2 には stream_events(…, version="v3") がありますが、手元では LangChainBetaWarning: The v3 streaming protocol on Pregel is experimental. が出ました。実験的な扱いなので、この記事では v2 を基本にしています。

よくあるエラーとつまずき（原文つき）

この節では、LangGraph で手元に出たエラーの原文と、その原因・直し方が一覧で分かります。

出た文言（原文）	原因	直し方
GraphRecursionError: Recursion limit of 6 reached without hitting a stop condition.	ループが上限の super-step 数に達した	終わる条件を見直す。RemainingSteps で手前で止める
InvalidUpdateError: At key 'results': Can receive only one value per step.	並列のノードが reducer の無いキーに同時に書いた	Annotated[list, operator.add] などの reducer を付ける
ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, …	checkpointer 付きのグラフを thread_id 無しで実行した	{"configurable": {"thread_id": …}} を渡す
RuntimeError: Cannot use Command(resume=...) without checkpointer	checkpointer を付けずに compile した	compile(checkpointer=…)
EmptyInputError: Received no input for __start__	保存が無い thread に invoke(None) した（durability=exit で落ちた後など）	入力を渡して最初から動かす。再開したいなら sync か async
UndefinedTable relation "checkpoints" does not exist	PostgresSaver の setup() を呼んでいない	起動時に PostgresSaver(conn).setup()
UserWarning: `durability` has no effect when no checkpointer is present.	checkpointer 無しで durability を指定した	checkpointer を付けるか、指定を外す
NotImplementedError（GenericFakeChatModel の bind_tools）	偽のモデルがツールの登録に対応していない	bind_tools を持つ自作の偽モデルを使う
エラーは出ないが承認が飛ばされる	interrupt を except Exception で囲んだ	interrupt を try の外に出す

判断表：LangGraph を使うか、どの checkpointer・durability にするか

この節では、LangGraph を採用するかどうかと、保存先・保存のタイミングを決める順番が分かります。

次の順に問うと、構成が決まります。

1. 11回の呼び出しで終わり、状態も分岐も無い → LangGraph は使わないモデルの SDK か LangChain の部品だけで書く
2. 2ツールのループだけで、止めて再開する必要が無い → create_agent など高レベルの部品公式は LangChain の部品を LangGraph の上に作られたものと説明している
3. 3分岐・並列・人の承認・長い実行がある → StateGraph で自分で組むこの記事の形
4. 4保存先：テストは InMemorySaver、1台は SqliteSaver、複数プロセスは PostgresSaverPostgresSaver は setup() と autocommit
5. 5durability：LLM や外部 API を呼ぶなら sync、落ちてもやり直せる短い処理は exitexit は途中の状態が残らない
6. 6外部への書き込みがある → 冪等キーと、thread ごとの再開の排他再開・replay・同時 resume で2回動く
7. 7デプロイや画面での可視化まで任せたい → LangSmith Deployment を公式で検討料金・機能は公式ページで確かめる

場面	構成	理由（手元の実測）
社内のチャットで送信前に承認	PostgresSaver・sync・interrupt	承認待ちの間プロセスを持たない。別プロセスから再開できた
夜間の長いバッチ（数十ステップ）	SqliteSaver か PostgresSaver・sync	kill -9 の後、終わったノードを飛ばして再開できた
数秒で終わる分類の処理	checkpointer 無しか exit	保存の手間がかからない（50ステップで 13〜15ms）
単体テスト	InMemorySaver＋固定応答のモデル	LLM 無しで分岐・承認・再開を毎回同じ結果で試せる
プロンプトの比較・障害の調査	get_state_history と replay	同じ親から枝を作れた。ただし再生は再実行

2番目の段の説明は、公式の概要ページの「LangGraph の上に作られた連携とエージェントの抽象化（LangChain）」という記述に基づきます。langgraph の最新版は PyPI で 1.2.19（2026年10月30日公開、MIT ライセンス）でした（2026年11月4日時点）。

よくある質問

この節では、LangGraph について検索でよく調べられる疑問への短い答えが分かります。

Q. LangGraph は無料で使えますか？
LangGraph 本体は MIT ライセンスの OSS で、PyPI から入れて自分のプロセスで動かす分には料金はかかりません。モデルの API や LangSmith・LangSmith Deployment は、それぞれの料金になります（公式の料金ページで確かめます）。

Q. LangChain を使わずに LangGraph だけで書けますか？
書けます。公式も単独で動くと書いています。この記事でも、langchain-core のメッセージとツールの型だけを使い、LangChain のチェーンやエージェントは使っていません。ノードの中で各社の SDK を直接呼ぶ書き方もできます。

Q. recursion_limit はいくつにすればよいですか？
想定するノードの実行回数に、余裕を足した値にします。agent→tools→agent の3回の実行は、手元では limit 4 から通りました。既定の 10,007 のままだと、偽のモデルでも 5,004 回の呼び出しで 89 秒かかりました。

Q. interrupt と interrupt_before の違いは何ですか？
interrupt はノードの中で値を返して止め、再開の値を受け取れます。interrupt_before・interrupt_after はノードの前後で止めるだけで、公式はデバッグ向けとし、人の入力を待つ用途には勧めていません。

Q. 承認待ちのまま何日も置けますか？
チェックポイントが DB に残っている限り、プロセスを止めても再開できます。手元でも、止めたプロセスと別のプロセスから再開しました。ただし、ツールの引数（宛先・金額）が今も正しいかを再開の前に見直す仕組みは、アプリ側で用意します。

Q. SqliteSaver を本番で使えますか？
公式は SqliteSaver を開発向け、PostgresSaver を本番向けとしています。1台・1プロセスで動く小さな仕組みなら動きますが、複数のプロセスから同じ thread を扱うなら PostgreSQL にします。

Q. チェックポイントの中身は暗号化できますか？
公式は EncryptedSerializer を用意しており、鍵を LANGGRAPH_AES_KEY 環境変数から読めると説明しています。会話に個人情報が入るなら、暗号化と保持期間の両方を決めます（この記事では暗号化は動かしていません）。

Q. LLM 無しでテストするにはどうしますか？
BaseChatModel を継承し、最後のメッセージを見て応答を返すモデルを作ります。GenericFakeChatModel は順番に応答を返す形なので、再開や replay でノードがもう一度動くと応答がずれます。

Q. LangGraph Platform と LangGraph は別物ですか？
別物です。LangGraph は OSS のライブラリ、LangGraph Platform はデプロイの製品で、2025年10月に LangSmith Deployment へ名称が変わりました（langchain.com の FAQ）。

Q. Python 以外でも使えますか？
公式ドキュメントには JavaScript / TypeScript 版の LangGraph もあります。この記事の出力は Python 版（1.2.19）のものです。

まとめ

この節では、この記事で確かめたことの要点が分かります。

• LangGraph は状態・分岐・保存・再開の実行基盤：LangChain が無くても動き、固定応答のモデルで LLM 無しに試せる。
• recursion_limit の既定は 10,007：トップレベルのキーで小さく指定する。configurable の中に書くと効かなかった。
• checkpointer で kill -9 から再開できた：終わったノードは飛ばし、落ちたノードは最初から。durability=exit では何も残らなかった。
• interrupt と Command(resume) で承認・却下・修正：再開でノードは先頭から動き、except Exception で囲むと止まらず、同時の resume では送信が2回動いた。
• 履歴は木の形で残る：replay と update_state は元を消さずに枝を作るが、再生はツールの再実行になる。

参考・出典

この節では、この記事の根拠にした公式の資料が分かります（確認日 2026年11月4日）。

• LangChain Docs「LangGraph overview」https://docs.langchain.com/oss/python/langgraph/overview
• LangChain Docs「Graph API」（recursion limit・reducer・Send・Command）https://docs.langchain.com/oss/python/langgraph/graph-api
• LangChain Docs「Persistence」https://docs.langchain.com/oss/python/langgraph/persistence
• LangChain Docs「Checkpointers」（pending writes・durability・serializer）https://docs.langchain.com/oss/python/langgraph/checkpointers
• LangChain Docs「Interrupts」https://docs.langchain.com/oss/python/langgraph/interrupts
• LangChain Docs「Streaming」https://docs.langchain.com/oss/python/langgraph/streaming
• LangChain「LangGraph Platform」（LangSmith Deployment への名称変更の FAQ）https://www.langchain.com/langgraph-platform
• PyPI「langgraph」https://pypi.org/project/langgraph/

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。