TL;DR
LangGraph 1.2.11 で interrupt() の挙動を条件別に実測したところ、中断と再開で checkpointer の要求が非対称 でした。
| 条件 |
interrupt() を含むノードの実行 |
結果 |
|---|---|---|
checkpointer なしで invoke()
|
例外なし |
__interrupt__ を含む dict が返り、後続ノードは実行されない |
checkpointer なしで Command(resume=...)
|
例外 | RuntimeError: Cannot use Command(resume=...) without checkpointer |
checkpointer ありで resume せず再 invoke()
|
例外なし | 前段ノードと中断ノードが もう一度 実行される |
checkpointer ありで Command(resume=...)
|
例外なし | 中断ノードだけ再実行され、前段ノードは再実行されない |
| SqliteSaver で別プロセスから resume | 例外なし | 前段ノードを再実行せずに完了する |
つまり interrupt() を checkpointer なしで呼んでも、その場では何も壊れません。壊れるのは再開しようとした瞬間です。無人実行のパイプラインでは、この「壊れない中断」が完了と区別されずに素通りします。
はじめに
LangGraph で人間の承認を挟むグラフを書くとき、承認ポイントには interrupt() を置きます。公式ドキュメントは checkpointer が必須だと明記していますが、手元で checkpointer を渡し忘れたグラフを走らせても例外は出ませんでした。返ってきたのは中身の詰まった辞書で、一見すると正常終了と区別が付きません。
対象読者は、LangGraph で human-in-the-loop を組み込んでいて、そのグラフをスケジュール実行や CI のような無人環境でも動かす予定のある開発者の方です。
承認待ちで止まったグラフを「完了した」と扱ってしまうと、後続ノードが走らないまま次の処理へ進みます。この記事では、どの条件で何が起き、無人実行側は何を見て中断を検知すればよいのかを実測で整理します。
検証環境
クラウド実行環境（Linux x64・Python 3.11.15）に、uv で作った仮想環境へ以下を導入しました。
| パッケージ | バージョン |
|---|---|
| langgraph | 1.2.11 |
| langgraph-checkpoint | 4.2.0 |
| langgraph-checkpoint-sqlite | 3.1.1 |
検証に使ったグラフは before → gate → after の3ノード直列で、gate の中で interrupt() を呼びます。LLM は呼ばず、各ノードは呼び出し回数をカウンタに刻むだけの純粋な関数にしました。ノードが何回走ったかを数値で見たかったためです。
CALLS = {"before": 0, "gate": 0, "after": 0}
def before(state):
CALLS["before"] += 1
return {"log": state["log"] + ["before"]}
def gate(state):
CALLS["gate"] += 1
ans = interrupt({"question": "approve?"})
return {"log": state["log"] + ["gate"], "approved": ans}
def after(state):
CALLS["after"] += 1
return {"log": state["log"] + ["after"]}
checkpointer なしでも中断そのものは通る
compile() に checkpointer を渡さず invoke() した結果です。
app = build(checkpointer=None)
out = app.invoke({"log": [], "approved": ""})
--- returned: {'log': ['before'], 'approved': '', '__interrupt__': [Interrupt(value={'question': 'approve?'}, id='2efe450d1248a1e70324411b1d0a3405')]}
--- node calls: {'before': 1, 'gate': 1, 'after': 0}
例外は送出されませんでした。戻り値は通常の状態辞書に __interrupt__ キーが加わったもので、after の呼び出し回数は 0 のままです。グラフは gate で止まっており、log にも gate と after が入っていません。
公式ドキュメントは interrupt を使う条件として「A checkpointer to persist the graph state」を挙げています（Interrupts）。要件を満たさない呼び出しなのに、実行時にはそれを知らせるものが何も出ないという状態です。
例外が出るのは再開しようとした瞬間
同じ checkpointer なしのグラフへ Command(resume=...) を渡すと、今度は落ちます。
app = build(checkpointer=None)
app.invoke(Command(resume="yes"))
--- raised: RuntimeError: Cannot use Command(resume=...) without checkpointer
中断は静かに成立し、再開だけがエラーになります。checkpointer の不足が表面化するのは、中断から再開までのラウンドトリップを実際に一周させたときだけです。動作確認で invoke() を1回叩いて「返ってきたから動いている」と判断すると、この非対称性は見えません。
resume を忘れて再実行すると前段ノードが二重に走る
ここからは InMemorySaver を付けた状態です。thread_id を固定して1回目の invoke() を実行し、中断させます。
--- return type: dict
--- keys: ['log', 'approved', '__interrupt__']
--- state values: {'log': ['before'], 'approved': ''}
--- node calls: {'before': 1, 'gate': 1, 'after': 0}
--- truthy? (無人実行が完了と誤判定しうるか): True
戻り値は truthy な辞書です。if result: のような素朴な判定は通過してしまいます。
続けて、resume を渡さずに 同じ入力 で同じ thread_id へもう一度 invoke() してみました。
--- keys: ['log', 'approved', '__interrupt__']
--- node calls (累計): {'before': 2, 'gate': 2, 'after': 0}
--- 再実行された回数の差分: {'before': 1, 'gate': 1, 'after': 0}
before と gate が1回ずつ余分に走りました。中断は解除されず、同じ地点でまた止まります。前段ノードが外部 API を叩いたりレコードを作ったりしていれば、その副作用が実行回数ぶん積み上がります。
正しく Command(resume="yes") を渡した場合はこうなります。
--- keys: ['log', 'approved']
--- final state: {'log': ['before', 'gate', 'after'], 'approved': 'yes'}
--- node calls (累計): {'before': 2, 'gate': 3, 'after': 1}
before の累計は 2 のまま増えず、中断ノードの gate が 1、その後続の after が 1 増えました。前段ノードだけが再実行を免れています。公式ドキュメントは再開時の挙動について「the runtime restarts the entire node from the beginning」と書いており、interrupt() を呼んだ行から続きが走るのではなく ノード単位で先頭から やり直されます。したがって interrupt() より前に置いた副作用は冪等である必要があります。この点は仕様どおりの動きでした。
差が出るのは resume を渡すかどうかです。渡せば再実行は中断ノード1つで済み、渡さなければグラフの入口から数え直しになります。
中断中のスレッドに別の入力を渡すと状態がマージされる
無人実行のリトライでよくあるのは、入力を作り直して同じ thread_id へ投げ直すパターンです。試してみました。
app.invoke({"log": [], "approved": ""}, cfg) # 1回目・中断する
out = app.invoke({"log": ["injected"], "approved": ""}, cfg) # 別の入力を渡す
--- keys: ['log', 'approved', '__interrupt__']
--- log : ['injected', 'before']
--- next: ('gate',)
新しい入力が state に書き込まれ、そこから before が再実行されています。log の中身が ['injected', 'before'] になっているとおり、中断前の履歴は上書きされました。それでいて next は ('gate',) のままで、中断は解除されていません。
入力を差し替えたリトライは、中断を解除しないまま状態だけを書き換えます。リトライのつもりが状態の破壊になり得るということです。
無人実行から中断を検知する3つの手段
例外が飛ばない以上、呼び出し側が明示的に中断を見に行く必要があります。実測できた手段は3つです。
--- '__interrupt__' in out : True
--- state.next : ('gate',)
--- len(state.interrupts) : 1
--- state.tasks[0].interrupts: [{'question': 'approve?'}]
-
戻り値の
__interrupt__キー:invoke()の返り値に含まれます。checkpointer の有無にかかわらず付くので、checkpointer を渡し忘れたグラフでも検知できます -
get_state(config).next: 次に実行予定のノード名のタプルです。完了していれば空になります。checkpointer が必要です -
get_state(config).interrupts: 中断のペイロードが取れます。tasks[].interruptsからも同じ値が読めます
stream() で流した場合は、中断がチャンクとして観測できました。
--- stream chunk keys: [['before'], ['__interrupt__']]
before のチャンクの次が __interrupt__ で、after のチャンクは流れません。ストリーミングで進捗を監視している場合は、最後のチャンクのキーを見れば中断かどうかが分かります。
interrupt の id は冪等キーに使えない
中断を外部のキューやデータベースに記録するなら、同じ中断を二重登録しないための識別子が欲しくなります。Interrupt オブジェクトは id を持っているので、これが使えるか確かめました。
--- id 1: d71940c0cb45b05bc5a2d5ad3e665129
--- id 2: 9848be9b3b92e02ce8dcad1de9ad00b8
--- same: False
同じ thread_id・同じ中断ポイントであっても、再実行のたびに別の id が振られました。この id を冪等キーにすると、同一の承認待ちがリトライ回数ぶん重複して登録されます。重複排除には thread_id と中断ノード名の組み合わせなど、こちら側で決めたキーを使うほうが安全です。
SqliteSaver ならプロセスを跨いで再開できる
スケジュール実行のように、中断したプロセスと再開するプロセスが別になるケースを SqliteSaver で確かめました。1回目のプロセスで中断させ、SQLite ファイルの中身を覗いています。
--- c1 node calls: {'before': 1, 'gate': 1, 'after': 0}
--- sqlite tables: ['checkpoints', 'writes']
--- checkpoints rows: 3
--- writes rows: 6
--- db size bytes: 20480
3ノードのグラフ1回の中断で、checkpoints が3行・writes が6行、ファイルサイズは 20,480 バイトでした。テーブルは checkpoints と writes の2つだけという素直な構成です。
同じ SQLite ファイルを開き直した2回目では、中断が復元されていました。
--- c2 next: ('gate',)
--- c2 interrupts: [{'question': 'approve?'}]
--- c2 final: {'log': ['before', 'gate', 'after'], 'approved': 'yes-from-new-process'}
--- c2 node calls (このプロセスでの実行回数): {'before': 0, 'gate': 1, 'after': 1}
このプロセスでの before の実行回数は 0 です。前段ノードを再実行することなく、gate から続きが走って完了しました。中断のペイロードも get_state() から読めているので、承認を求める内容を別プロセス（通知を送るワーカーなど）から取り出せます。
著者視点の発見ポイント
筆者が今回いちばん引っかかったのは、checkpointer なしの invoke() が「成功でも失敗でもない第三の返り方」をする点です。例外でも None でもなく、状態辞書に1つキーが増えるだけで返ってきます。
無人実行のコードは、たいてい例外か戻り値の真偽で成否を分けます。今回の中断はそのどちらにも引っかかりません。実際に走らせるまで、筆者もグラフが最後まで走ったものと読み違えていました。カウンタを仕込んで after の実行回数が 0 だと分かって初めて、途中で止まっていたと気づいた形です。
もう1つは、resume を渡さない再 invoke() が「同じ場所で止まり直す」だけでなく、前段ノードを走らせ直すことです。中断中のスレッドに新しい入力を渡した実験（log が ['injected', 'before'] になった件）と合わせると、中断中のグラフに対する invoke() は再開ではなく 新しい実行の開始 として扱われていると理解するのが実態に近いと感じました。リトライを素朴に実装すると、承認待ちのグラフを叩くたびに前段の副作用が増えていきます。
実装するときの整理
実測から導ける方針を並べます。
-
compile(checkpointer=...)は human-in-the-loop を含むグラフでは省略しない。省略しても起動時に気づけないため、グラフの生成箇所で checkpointer がNoneでないことを自前で確認しておく -
invoke()の戻り値は必ず"__interrupt__" in resultで判定する。真偽値や例外の有無では中断を検知できない - 中断を検知したら、resume 用の入力が揃うまで 同じ
thread_idへinvoke()を投げ直さない。投げ直すと前段ノードが再実行される -
interrupt()より前に置く副作用は冪等にする。これは公式ドキュメントが明記している要件で、実測でも再開時に中断ノードが先頭から走り直しました - 中断を外部に記録するときの冪等キーに
Interrupt.idを使わない。再実行のたびに変わります - プロセスを跨ぐ運用では
SqliteSaverなどファイルに落ちる checkpointer を使う。別プロセスからget_state()で中断内容を読み、Command(resume=...)で前段を再実行せずに続きを走らせられます
まとめ
LangGraph 1.2.11 の interrupt() は、checkpointer がなくても中断だけは成立し、再開しようとしたときに初めて RuntimeError になります。中断そのものは例外にならず、__interrupt__ キーを持つ truthy な辞書として返るため、無人実行のパイプラインでは正常完了と区別が付きません。
検知は戻り値の __interrupt__ キーを見るのが最も確実で、これは checkpointer の有無にかかわらず機能します。resume を渡さない再実行は前段ノードの再実行を招き、中断中のスレッドへ別の入力を渡すと状態がマージされて履歴が書き換わります。承認待ちを跨ぐ処理を組むなら、ファイルに落ちる checkpointer と Command(resume=...) の組み合わせが前提になります。
関連記事
- Rust製MCPサーバーをcargoに繋いだら、ツール定義だけで66KB消えた
- Go製EinoでReActエージェントを動かし、MaxIterations超過で最終回答が消えるのを実測した
- Goose reviewを試したら、チェック定義ファイルまで診断対象に入っていた