⬡ API & SDK/2026-04-07上級

Claude API × LangGraph 本番品質ステートフルエージェント構築ガイド — グラフベース設計・永続化・ヒューマンインザループの実装

LangGraphとClaude APIでステートフルエージェントを本番運用するための実装パターン集。from_conn_stringがcompile()で落ちる原因、recursion_limitの既定値変更、interrupt()再開の落とし穴まで、手元で検証した挙動と共に整理します。

LangGraphClaude API¹²⁸ エージェント¹⁴ステートフルマルチエージェント⁶ Python²² 本番運用³⁶

✦ プレミアム記事

取り組みの背景 — なぜ「ステートフル」なエージェントが必要なのか

AIエージェントの開発において、最大の課題のひとつが「状態管理」です。シンプルな質問応答システムであれば、各リクエストを独立したものとして扱えます。しかし現実のビジネスワークフローは、それほど単純ではありません。

たとえば、見積作成から上長承認、最終発注まで複数ステップにわたる承認フローを考えてみてください。あるいは、データ収集・分析・レポート作成と段階を踏む長時間実行の研究タスク。エラー時に途中から再開できる大規模バッチ処理、または人間の判断を適切なタイミングで挟む意思決定プロセス——これらのシナリオでは、エージェントが「今どこまで進んでいるか」「前回の実行でどんなデータを収集したか」を記憶し続ける必要があります。このような要件を満たすのがステートフルエージェントです。

LangGraphは、LangChainチームが開発したグラフベースのエージェントフレームワークです。Pythonの状態機械（ステートマシン）のようにエージェントのワークフローを定義でき、永続化・再開・ヒューマンインザループなどを標準サポートしています。単発のチェーン実行を前提とした従来型の構成や、Cloudflare Durable Objectsでインフラ側に状態を持たせる方法と比べて、LangGraphは「循環・分岐・長期記憶」をアプリケーション層のグラフ定義として扱える点に特徴があります。

Claude APIの高度な推論能力とLangGraphの堅牢な状態管理を組み合わせることで、エンタープライズレベルのAIエージェントが構築できます。ここではこの組み合わせの実装パターンを実際のコードを交えながら深堀りします。

私自身、個人開発で運営しているアプリ群の問い合わせ対応を支援する内部ワークフローに、この構成を小さく導入しております。返金や規約に関わる返信の下書きだけは必ず人間の確認を挟みたい、という要件がまさに interrupt() の出番でした。承認待ちのまま一晩置いてもチェックポイントから正確に再開できる。cronとJSONファイルで状態管理を自作していた頃には得られなかった安心感です。

環境構築とセットアップ

まず必要なパッケージをインストールします。Python 3.11以上を推奨します。

pip install anthropic langgraph langchain-anthropic langgraph-checkpoint-sqlite tenacity

各パッケージの役割を整理します。

anthropic — Claude API クライアント（最新SDK）
langgraph — グラフベースエージェントフレームワーク
langchain-anthropic — LangChain向けClaude統合レイヤー
langgraph-checkpoint-sqlite — SQLiteを使ったチェックポイント永続化
tenacity — リトライロジックのライブラリ

基本的なセットアップが正しく機能するか確認します。

import os
import sqlite3
from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
 
# Claude モデルの初期化（temperature=0 で再現性を高める）
model = ChatAnthropic(
    model="claude-sonnet-4-6",
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    temperature=0,
    max_tokens=4096,
)
 
# SQLiteチェックポインターの設定
#
# ⚠️ SqliteSaver.from_conn_string() は「コンテキストマネージャ」です。
#    memory = SqliteSaver.from_conn_string(":memory:") と代入すると、
#    受け取れるのは _GeneratorContextManager であって SqliteSaver ではありません。
#    この後の compile(checkpointer=memory) が TypeError で落ちます（詳細は後述）。
#
# 本記事では、以降のノード定義・グラフ構築でも同じ checkpointer を
# 使い回すため、接続を自分で保持するコンストラクタ形式を採用します。
conn = sqlite3.connect(":memory:", check_same_thread=False)
memory = SqliteSaver(conn)
 
print("✅ LangGraph × Claude API セットアップ完了")
print(type(memory).__name__)
# 出力:
# ✅ LangGraph × Claude API セットアップ完了
# SqliteSaver

check_same_thread=False は必須です。LangGraph はノードをワーカースレッドで実行することがあり、既定の SQLite 接続はスレッドをまたぐと ProgrammingError を投げます。

短命なスクリプトであれば with SqliteSaver.from_conn_string(":memory:") as memory: の形が最も安全です。ただし with を抜けた時点で接続が閉じられるため、長時間動き続けるWebアプリではコンストラクタ形式のほうが扱いやすくなります。

環境変数 ANTHROPIC_API_KEY に有効なAPIキーを設定しておく必要があります。本番環境では .env ファイルや AWS Secrets Manager などのシークレット管理サービスを使ってください。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦interrupt()承認待ちスレッドがデプロイ後に再開できなくなるステートスキーマ進化問題と、total=False・バージョンキーによる回避パターン

✦SqliteSaver/PostgresSaverのfrom_conn_stringがコンテキストマネージャであるためcompile()がTypeErrorで落ちる問題と、用途別の2つの正しい書き方

✦recursion_limitの既定値が25から10007へ変わったことで暴走ループのリスクが反転した点と、二重上限による打ち切り設計

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

LangGraph の基本概念 — グラフ・ノード・エッジ

LangGraphを使いこなすには、3つのコアコンセプトを理解する点が肝心です。

ステート（State）: エージェントが保持するすべての情報の型定義です。TypedDict を使って型安全に管理することで、実行中のデータ構造が明確になります。

ノード（Node）: ステートを受け取り、処理を行い、更新されたステートのサブセットを返す関数です。各ノードは一つの責任（情報収集、分析、判断など）を担います。

エッジ（Edge）: ノード間の接続を定義します。固定エッジ（常に同じノードへ）と条件付きエッジ（ステートの値によって分岐先が変わる）の2種類があります。

以下のコードで、分析→評価→分岐という基本的なサイクルを実装します。

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
import operator
 
# ステートの定義
class AgentState(TypedDict):
    messages: Annotated[List, operator.add]  # operator.add で追記型リストになる
    iteration: int                           # 反復回数
    final_result: str                        # 最終出力
 
# 分析ノード
def analyze_node(state: AgentState) -> dict:
    """メインの分析ステップ"""
    messages = state["messages"]
    response = model.invoke(messages)
 
    return {
        "messages": [response],
        "iteration": state.get("iteration", 0) + 1,
    }
 
# 評価ノード
def evaluate_node(state: AgentState) -> dict:
    """完了か継続かを判断するステップ"""
    last_message = state["messages"][-1]
 
    eval_prompt = HumanMessage(content=f"""
    前回の応答: {last_message.content[:500]}
 
    この応答は質問に対して十分な回答になっていますか？
    「complete」または「continue」とだけ答えてください。
    """)
 
    eval_response = model.invoke([eval_prompt])
    decision = "complete" if "complete" in eval_response.content.lower() else "continue"
 
    return {"final_result": decision}
 
# 条件分岐ロジック（最大3回でタイムアウト）
def should_continue(state: AgentState) -> str:
    if state.get("final_result") == "complete" or state.get("iteration", 0) >= 3:
        return "end"
    return "analyze"
 
# グラフの構築
workflow = StateGraph(AgentState)
 
# ノードの登録
workflow.add_node("analyze", analyze_node)
workflow.add_node("evaluate", evaluate_node)
 
# エントリーポイントと接続の定義
workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "evaluate")
workflow.add_conditional_edges(
    "evaluate",
    should_continue,
    {"analyze": "analyze", "end": END}
)
 
# グラフのコンパイル
app = workflow.compile()
 
# 実行テスト
initial_state = {
    "messages": [HumanMessage(content="量子コンピュータが現代の暗号技術に与える脅威と対策を説明してください")],
    "iteration": 0,
    "final_result": "",
}
 
final_state = app.invoke(initial_state)
print(f"完了ステータス: {final_state['final_result']}")
print(f"総反復回数: {final_state['iteration']}")
# 出力例:
# 完了ステータス: complete
# 反復回数: 2

このシンプルな例でも、analyze → evaluate → (analyze or END) というサイクルが自動的に制御されます。人間がループ制御を書く必要はありません。

永続化とチェックポイント — 中断・再開を実現する

LangGraphの最も強力な機能のひとつがチェックポイント機能です。実行中の任意の時点でステートを永続化でき、後から同じ状態に戻って実行を再開できます。これにより以下が実現します。

サーバー再起動後も会話セッションが継続する
長時間タスクの途中状態を安全に保存する
障害発生時に最後の成功ポイントから自動回復する
ユーザーごとの独立したセッション管理ができる

以下の実装では、SQLiteを使った永続化と、スレッドIDによるセッション管理を示します。

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
 
# 本番環境ではファイルパスを指定
# 開発: ":memory:" | 本番: "/var/data/agent_checkpoints.db"
conn = sqlite3.connect("agent_checkpoints.db", check_same_thread=False)
memory = SqliteSaver(conn)
 
# チェックポイント付きでグラフをコンパイル
app_with_memory = workflow.compile(checkpointer=memory)
 
# スレッドIDでユーザーセッションを識別
config_user_a = {"configurable": {"thread_id": "user-alice-research-001"}}
 
# セッション1: 最初のクエリを実行
first_query = {
    "messages": [HumanMessage(content="2026年の生成AI市場規模を調査してください")],
    "iteration": 0,
    "final_result": "",
}
 
print("セッション1 開始...")
for event in app_with_memory.stream(first_query, config_user_a):
    for node_name in event:
        print(f"  ノード '{node_name}' 実行完了")
 
# --- 後のリクエスト（別のHTTPリクエストや再起動後でも可） ---
# 同じ thread_id を使うだけで前回の会話履歴が自動的に引き継がれる
follow_up = {
    "messages": [HumanMessage(content="前回の調査を踏まえて、日本市場に特化した課題を教えてください")],
    "iteration": 0,
    "final_result": "",
}
 
print("\nセッション2（継続）開始...")
final = app_with_memory.invoke(follow_up, config_user_a)
print(f"累積メッセージ数: {len(final['messages'])}")
print(f"最終ステータス: {final['final_result']}")
# 出力例:
# セッション1 開始...
#   ノード 'analyze' 実行完了
#   ノード 'evaluate' 実行完了
# セッション2（継続）開始...
# 累積メッセージ数: 6
# 最終ステータス: complete

本番環境では SQLite の代わりに PostgresSaver を使用します。スケーラブルで複数インスタンスからの同時アクセスにも対応できます。

# 本番用: PostgreSQL チェックポインター
#
# ⚠️ 依存パッケージは psycopg2 ではなく psycopg 3 系です。
#    langgraph/checkpoint/postgres/__init__.py の冒頭が
#    `from psycopg import Capabilities, Connection, ...` なので、
#    psycopg2-binary を入れても ImportError のままです。
# pip install "langgraph-checkpoint-postgres" "psycopg[binary]"
 
from langgraph.checkpoint.postgres import PostgresSaver
 
# from_conn_string は SqliteSaver と同じくコンテキストマネージャです。
# 戻り値を変数に代入して .setup() を呼ぶと AttributeError になります。
with PostgresSaver.from_conn_string(
    os.environ["DATABASE_URL"]  # postgresql://user:pass@host:5432/dbname
) as pg_memory:
    pg_memory.setup()  # 必要なテーブルを作成（初回に一度、明示的な呼び出しが必要）
    production_app = workflow.compile(checkpointer=pg_memory)
    # グラフの実行は with ブロックの内側で行う（抜けると接続が閉じる）

Webアプリのようにプロセスが長く生き続ける構成では、with ではなく接続プールを自分で管理し PostgresSaver(pool) を組み立てるほうが素直です。setup() はマイグレーションを含むため、アプリ起動時ではなくデプロイ手順の一部として一度だけ走らせる運用をお勧めします。

ヒューマンインザループ — 重要な判断ポイントで人間が介入する

エンタープライズ用途では、AIが自律で動き続けるだけでなく、重要な判断ポイントで人間が介入できる設計が求められます。契約書への署名、本番データベースへの変更、外部への発注・決済など、取り消しのきかない操作の前には必ず人間の確認を挟むべきです。

LangGraphでは interrupt() 関数を使うことで、実行を一時停止して人間の入力を待つことができます。

from langgraph.types import interrupt, Command
from typing import TypedDict
 
class ApprovalState(TypedDict):
    task: str
    draft_proposal: str
    human_decision: str
    approved: bool
    final_document: str
 
def generate_draft_node(state: ApprovalState) -> dict:
    """提案ドラフトを自動生成するノード"""
    response = model.invoke([
        HumanMessage(content=f"""
        以下のタスクについて詳細な実施提案を作成してください:
 
        タスク: {state['task']}
 
        提案は具体的な手順、コスト見積もり、リスク分析を含めてください。
        """)
    ])
 
    return {"draft_proposal": response.content}
 
def request_human_approval_node(state: ApprovalState) -> dict:
    """人間の承認を求めるノード（実行を一時停止）"""
 
    # interrupt() を呼ぶと実行がここで停止する
    # チェックポインターに現在のステートが保存され、後から再開できる
    human_input = interrupt({
        "message": "生成された提案を確認してください",
        "draft": state["draft_proposal"],
        "options": {
            "approve": "承認して最終化する",
            "revise": "修正を加えて再生成する",
            "reject": "却下する"
        }
    })
 
    approved = human_input == "approve"
    return {
        "human_decision": human_input,
        "approved": approved,
    }
 
def finalize_document_node(state: ApprovalState) -> dict:
    """承認後の最終文書作成ノード"""
    if state["approved"]:
        response = model.invoke([
            HumanMessage(content=f"""
            以下の承認済み提案を、正式な業務文書として整形してください:
 
            {state['draft_proposal']}
 
            文書番号、日付、承認済みスタンプの記載も含めてください。
            """)
        ])
        return {"final_document": response.content}
    else:
        return {"final_document": f"提案は却下されました（理由: {state['human_decision']}）"}
 
def route_after_approval(state: ApprovalState) -> str:
    if state["approved"]:
        return "finalize"
    elif state["human_decision"] == "revise":
        return "generate"  # 再生成
    else:
        return "end"  # 却下
 
# 承認ワークフローのグラフ構築
approval_workflow = StateGraph(ApprovalState)
approval_workflow.add_node("generate", generate_draft_node)
approval_workflow.add_node("approval", request_human_approval_node)
approval_workflow.add_node("finalize", finalize_document_node)
 
approval_workflow.set_entry_point("generate")
approval_workflow.add_edge("generate", "approval")
approval_workflow.add_conditional_edges(
    "approval",
    route_after_approval,
    {"finalize": "finalize", "generate": "generate", "end": END}
)
approval_workflow.add_edge("finalize", END)
 
approval_app = approval_workflow.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "approval-flow-project-alpha"}}
 
# フェーズ1: ドラフト生成まで実行（承認ポイントで自動停止）
print("フェーズ1: ドラフト生成...")
initial_state = {
    "task": "新しい顧客管理システムへの移行計画",
    "draft_proposal": "",
    "human_decision": "",
    "approved": False,
    "final_document": "",
}
 
for event in approval_app.stream(initial_state, config):
    for key in event:
        print(f"  ノード '{key}' 完了")
# approval ノードの interrupt() でここで停止する
 
# --- ここで人間が提案内容を確認 ---
print("\n提案が生成されました。人間の確認を待機中...")
 
# フェーズ2: 人間の承認を受けて実行を再開
human_decision = "approve"  # 実際はAPIやUIから受け取る
 
print(f"\nフェーズ2: '{human_decision}' の決定で実行を再開...")
final = approval_app.invoke(
    Command(resume=human_decision),
    config
)
print(f"最終文書:\n{final['final_document'][:200]}...")
# 出力例:
# フェーズ1: ドラフト生成...
#   ノード 'generate' 完了
#   ノード 'approval' 完了  ← ここで interrupt() により停止
# 提案が生成されました。人間の確認を待機中...
# フェーズ2: 'approve' の決定で実行を再開...
# 最終文書:
# 【承認済み業務文書】No.2026-001 ...

このパターンの重要なポイントは、interrupt()が発火してからresumeするまでの時間がどれだけ長くても（数時間、数日でも）問題ないことです。チェックポインターがステートを永続化しているため、システムを再起動してもセッションが復元されます。

マルチエージェント協調 — 役割分担で複雑タスクを処理する

大きく複雑なタスクを単一のエージェントで処理しようとすると、コンテキストウィンドウの限界やプロンプトの肥大化が問題になります。LangGraphではSupervisorパターンを使って、複数の専門エージェントを協調させる構造が推奨されます。

from typing import Annotated
import operator
 
class TeamState(TypedDict):
    messages: Annotated[list, operator.add]
    next_worker: str
    team_results: dict
    final_report: str
 
def supervisor_node(state: TeamState) -> dict:
    """タスクを分析して次のエージェントを指定するスーパーバイザー"""
 
    system_prompt = """あなたはプロジェクトマネージャーです。
 
    利用可能なチームメンバー:
    - researcher: Webリサーチ・情報収集の専門家
    - data_analyst: 統計分析・データ解釈の専門家
    - writer: レポート・ドキュメント作成の専門家
    - FINISH: 全タスク完了（最終レポートが揃ったとき）
 
    現在の会話とチームの成果物を見て、次に誰を動かすべきか1つだけ答えてください。
    名前のみ（researcher / data_analyst / writer / FINISH）を回答してください。
    """
 
    response = model.invoke([
        HumanMessage(content=system_prompt),
        *state["messages"]
    ])
 
    # 次のワーカーを抽出
    next_worker = "researcher"  # デフォルト
    response_text = response.content.strip().lower()
    for worker in ["researcher", "data_analyst", "writer", "FINISH"]:
        if worker.lower() in response_text:
            next_worker = worker
            break
 
    return {
        "next_worker": next_worker,
        "messages": [response]
    }
 
def researcher_agent(state: TeamState) -> dict:
    """リサーチ専門エージェント"""
    response = model.invoke([
        HumanMessage(content="""あなたは市場リサーチの専門家です。
        与えられたテーマについて、具体的な統計・事例・トレンドを収集・整理してください。
        情報源の種類と信頼性も示してください。"""),
        *state["messages"]
    ])
    results = state.get("team_results", {})
    results["research"] = response.content
    return {
        "messages": [response],
        "team_results": results
    }
 
def data_analyst_agent(state: TeamState) -> dict:
    """データ分析専門エージェント"""
    research_data = state.get("team_results", {}).get("research", "")
 
    response = model.invoke([
        HumanMessage(content=f"""あなたはデータアナリストです。
        以下のリサーチ結果を定量的に分析し、重要なパターンと洞察を抽出してください。
        数値化できる指標は必ず数値で表現してください。
 
        リサーチデータ:
        {research_data[:2000]}
        """)
    ])
    results = state.get("team_results", {})
    results["analysis"] = response.content
    return {
        "messages": [response],
        "team_results": results
    }
 
def writer_agent(state: TeamState) -> dict:
    """レポート作成専門エージェント"""
    team_results = state.get("team_results", {})
 
    response = model.invoke([
        HumanMessage(content=f"""あなたはプロのテクニカルライターです。
        以下のリサーチ・分析結果を、経営幹部向けの構造化されたレポートにまとめてください。
 
        エグゼクティブサマリー（3行以内）→ 詳細分析 → 推奨アクション の順で記述してください。
 
        リサーチ結果: {team_results.get('research', '')[:1000]}
 
        分析結果: {team_results.get('analysis', '')[:1000]}
        """)
    ])
    results = state.get("team_results", {})
    results["report"] = response.content
    return {
        "messages": [response],
        "team_results": results,
        "final_report": response.content
    }
 
def route_from_supervisor(state: TeamState) -> str:
    next_worker = state.get("next_worker", "researcher")
    if next_worker == "FINISH":
        return END
    return next_worker
 
# マルチエージェントグラフの構築
team_graph = StateGraph(TeamState)
team_graph.add_node("supervisor", supervisor_node)
team_graph.add_node("researcher", researcher_agent)
team_graph.add_node("data_analyst", data_analyst_agent)
team_graph.add_node("writer", writer_agent)
 
team_graph.set_entry_point("supervisor")
team_graph.add_conditional_edges(
    "supervisor",
    route_from_supervisor,
    {
        "researcher": "researcher",
        "data_analyst": "data_analyst",
        "writer": "writer",
        END: END
    }
)
# 各エージェントはスーパーバイザーに戻る
for worker in ["researcher", "data_analyst", "writer"]:
    team_graph.add_edge(worker, "supervisor")
 
team_app = team_graph.compile(checkpointer=memory)
 
# チームワークフロー実行
config = {"configurable": {"thread_id": "team-report-2026-q2"}}
 
initial = {
    "messages": [HumanMessage(content="2026年第2四半期の日本のSaaS市場動向を分析し、経営判断に使えるレポートを作成してください")],
    "next_worker": "",
    "team_results": {},
    "final_report": "",
}
 
print("マルチエージェントチーム始動...")
for event in team_app.stream(initial, config, stream_mode="updates"):
    for node in event:
        print(f"  [{node}] 作業完了")
 
final = team_app.get_state(config).values
print(f"\n最終レポート（冒頭）:\n{final.get('final_report', '')[:300]}...")
# 出力例:
# マルチエージェントチーム始動...
#   [supervisor] 作業完了
#   [researcher] 作業完了
#   [supervisor] 作業完了
#   [data_analyst] 作業完了
#   [supervisor] 作業完了
#   [writer] 作業完了
#   [supervisor] 作業完了
# 最終レポート（冒頭）:
# 【エグゼクティブサマリー】2026年Q2の日本SaaS市場は前年比23%成長...

エラー回復と本番品質の実装パターン

本番環境ではAPIのタイムアウト、レート制限、ネットワークエラーが避けられません。Claude APIの本番回復パターンで紹介している考え方を、LangGraphと組み合わせて実装します。

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from anthropic import RateLimitError, APITimeoutError
import traceback
import time
 
class ResilientAgentState(TypedDict):
    task: str
    partial_results: list
    error_log: list
    retry_count: int
    status: str
 
def resilient_task_node(state: ResilientAgentState) -> dict:
    """リトライロジック付きのタスク実行ノード"""
 
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=2, min=4, max=30),
        retry=retry_if_exception_type((RateLimitError, APITimeoutError)),
        reraise=True
    )
    def call_api_with_retry():
        """レート制限・タイムアウトに対するリトライ"""
        return model.invoke([
            HumanMessage(content=f"""
            タスク: {state['task']}
 
            これまでの部分的な結果:
            {state['partial_results'][-3:] if state['partial_results'] else '（なし）'}
 
            上記を踏まえて次のステップを実行してください。
            """)
        ])
 
    try:
        response = call_api_with_retry()
        return {
            "partial_results": state["partial_results"] + [response.content],
            "status": "in_progress",
            "retry_count": state.get("retry_count", 0),
        }
 
    except RateLimitError as e:
        # レート制限は深刻ではないが、記録する
        error_entry = {
            "type": "rate_limit",
            "message": str(e),
            "timestamp": time.time(),
            "attempt": state.get("retry_count", 0)
        }
        return {
            "error_log": state.get("error_log", []) + [error_entry],
            "status": "rate_limited",
            "retry_count": state.get("retry_count", 0) + 1,
        }
 
    except Exception as e:
        error_entry = {
            "type": type(e).__name__,
            "message": str(e)[:200],
            "traceback": traceback.format_exc()[:500],
            "timestamp": time.time(),
        }
        return {
            "error_log": state.get("error_log", []) + [error_entry],
            "status": "error",
            "retry_count": state.get("retry_count", 0) + 1,
        }
 
def error_recovery_node(state: ResilientAgentState) -> dict:
    """エラーから学習して代替アプローチで再試行するノード"""
    recent_error = state.get("error_log", [{}])[-1]
 
    recovery_response = model.invoke([
        HumanMessage(content=f"""
        実行中に以下のエラーが発生しました:
        エラー種別: {recent_error.get('type', 'Unknown')}
        エラー詳細: {recent_error.get('message', '')}
 
        元のタスク: {state['task']}
        これまでの結果: {state['partial_results']}
 
        エラーを踏まえて、より堅牢な方法で再試行してください。
        エラーの原因を回避した代替アプローチを取ってください。
        """)
    ])
 
    return {
        "partial_results": state["partial_results"] + [f"[回復] {recovery_response.content}"],
        "status": "recovered",
    }
 
def completion_check(state: ResilientAgentState) -> dict:
    """タスク完了の判定ノード"""
    if len(state.get("partial_results", [])) >= 3:
        return {"status": "complete"}
    return {"status": "in_progress"}
 
def route_resilient(state: ResilientAgentState) -> str:
    status = state.get("status", "in_progress")
    retry_count = state.get("retry_count", 0)
 
    if status == "complete":
        return "end"
    elif status in ["error", "rate_limited"] and retry_count < 3:
        return "recovery"
    elif retry_count >= 3:
        return "end"  # 最大リトライ超過、強制終了
    else:
        return "task"
 
resilient_graph = StateGraph(ResilientAgentState)
resilient_graph.add_node("task", resilient_task_node)
resilient_graph.add_node("recovery", error_recovery_node)
resilient_graph.add_node("check", completion_check)
 
resilient_graph.set_entry_point("task")
resilient_graph.add_edge("task", "check")
resilient_graph.add_conditional_edges(
    "check",
    route_resilient,
    {"task": "task", "recovery": "recovery", "end": END}
)
resilient_graph.add_edge("recovery", "task")
 
resilient_app = resilient_graph.compile(checkpointer=memory)

デバッグとモニタリング — 本番運用のための可視化

本番環境でのLangGraphエージェントの問題調査には、構造化されたログとモニタリングが不可欠です。

import json
import time
from datetime import datetime, timezone
 
class AgentMonitor:
    """エージェント実行の監視・ログ記録クラス"""
 
    def __init__(self, agent_id: str, environment: str = "production"):
        self.agent_id = agent_id
        self.environment = environment
        self.execution_log = []
        self.start_time = None
 
    def start_session(self, thread_id: str):
        """セッション開始を記録"""
        self.start_time = time.time()
        self._log("session_start", {"thread_id": thread_id})
 
    def log_node_execution(self, node_name: str, state_summary: dict, duration_ms: int):
        """ノード実行を記録（Datadog/CloudWatchに送信可能）"""
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "agent_id": self.agent_id,
            "environment": self.environment,
            "node": node_name,
            "duration_ms": duration_ms,
            "state_keys": list(state_summary.keys()),
            "iteration": state_summary.get("iteration", 0),
            "status": state_summary.get("status", "unknown"),
        }
        self.execution_log.append(log_entry)
        # 構造化ログ出力（本番では logging ライブラリや外部サービスに送信）
        print(f"[TRACE] {json.dumps(log_entry, ensure_ascii=False)}")
 
    def log_error(self, node_name: str, error: Exception, retry_count: int):
        """エラーを記録しアラートを送信"""
        error_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "agent_id": self.agent_id,
            "node": node_name,
            "error_type": type(error).__name__,
            "error_message": str(error)[:500],
            "retry_count": retry_count,
            "severity": "WARNING" if retry_count < 2 else "ERROR",
        }
        print(f"[{error_entry['severity']}] {json.dumps(error_entry, ensure_ascii=False)}")
 
    def get_execution_summary(self) -> dict:
        """実行全体のサマリーを生成"""
        if not self.execution_log:
            return {"status": "no_executions"}
 
        total_ms = sum(e["duration_ms"] for e in self.execution_log)
        node_counts = {}
        for entry in self.execution_log:
            node_counts[entry["node"]] = node_counts.get(entry["node"], 0) + 1
 
        return {
            "agent_id": self.agent_id,
            "total_nodes_executed": len(self.execution_log),
            "total_duration_ms": total_ms,
            "avg_duration_ms": total_ms // len(self.execution_log),
            "node_execution_counts": node_counts,
            "session_duration_sec": round(time.time() - self.start_time, 2) if self.start_time else 0,
        }
 
# 実際のノードにモニタリングを組み込む例
monitor = AgentMonitor("market-research-agent-v2", environment="production")
 
def monitored_analyze_node(state: AgentState) -> dict:
    start = time.time()
    result = analyze_node(state)
    duration_ms = int((time.time() - start) * 1000)
    monitor.log_node_execution("analyze", state, duration_ms)
    return result
 
summary = monitor.get_execution_summary()
print(f"\n📊 実行サマリー:\n{json.dumps(summary, ensure_ascii=False, indent=2)}")
# 出力例:
# 📊 実行サマリー:
# {
#   "agent_id": "market-research-agent-v2",
#   "total_nodes_executed": 8,
#   "total_duration_ms": 5840,
#   "avg_duration_ms": 730,
#   "node_execution_counts": {"analyze": 3, "evaluate": 3, "recovery": 2},
#   "session_duration_sec": 6.12
# }

LangGraph Studio を使ったビジュアルデバッグ

LangGraph Studioは、グラフの実行フローをリアルタイムで可視化できるローカルデバッグツールです。

# LangGraph CLI のインストール
pip install langgraph-cli
 
# プロジェクト設定ファイル（langgraph.json）を作成
cat > langgraph.json << 'EOF'
{
    "dependencies": ["./"],
    "graphs": {
        "agent": "./src/agent.py:app",
        "team": "./src/agent.py:team_app"
    },
    "env": ".env"
}
EOF
 
# Studio の起動
langgraph dev --port 2024
# → ブラウザで http://localhost:2024 を開く

Studioでは以下のデバッグ機能が利用できます。

グラフの構造とノード間の接続を視覚的に確認できる
各ノードへの入力・出力（ステートの全フィールド）を確認できる
チェックポイントの一覧と各時点のステートを確認できる
タイムライン形式での実行履歴とボトルネックの特定ができる
任意のチェックポイントから実行を再開して挙動を検証できる

本番デプロイ前に必ずStudioでグラフの動作を確認することを強く推奨します。

公式ドキュメントには書かれていない、運用で気づいたこと

チュートリアルどおりに動かすところまでは順調でした。つまずいたのは、動き始めてから数週間後です。ここでは私の環境で実際に起きた4つの問題と、その対処を共有します。

運用課題	典型的な症状	初手の対処
チェックポイントの肥大	SQLiteファイルが数週間で数百MBに成長	delete_thread による週次棚卸しジョブ
ステートスキーマの進化	デプロイ後、承認待ちスレッドの再開時に KeyError	TypedDict を total=False にし .get() で読む
再帰上限の既定値が変わった	暴走ループが止まらずトークンを消費し続ける	recursion_limit を業務都合の値で明示＋ステートに試行回数カウンタ
入力トークンの肥大	ステップが進むほど1呼び出しの単価が上がる	トリミングノードで messages を要約・間引き

チェックポイントは「スーパーステップごと」に積まれる

チェックポインタはグラフのスーパーステップ単位でステートのスナップショットを保存します。つまり1回の実行で、ノード数ぶんの履歴が積まれます。私の環境では、12ノード構成のグラフを1回実行するごとにSQLiteファイルが約40〜80KB増え、1日100実行ペースの検証を3週間続けた時点で約210MBに達しました。放置するとバックアップも読み込みも遅くなる一方です。

対処として、完了済みスレッドを1週間保持したのちに checkpointer.delete_thread(thread_id) で削除する棚卸しジョブを入れました。導入後はファイルサイズが約30MB前後で安定しております。PostgresSaverでも事情は同じで、checkpoints テーブルの行数監視と定期削除は最初から設計に含めておくことをお勧めします。

ステートスキーマを変えると、古いチェックポイントが再開できない

もっとも痛かったのはこれです。TypedDictに新しいキーを追加してデプロイしたところ、デプロイ前から interrupt() で承認待ちになっていたスレッドが、再開時に新キーを参照するノードで KeyError を投げました。チェックポイントには「古い形のステート」が保存されているためです。

以来、私は次の3点を守るようにしています。第一に、ステート定義は total=False にして、ノード側は必ず state.get("key", default) で読む。第二に、ステートに schema_version キーを持たせ、再開直後のノードで旧バージョンを検出したら初期値を補完する。第三に、承認待ちスレッドが残っている間の破壊的なスキーマ変更は避け、キーの追加のみに留める。地味な規約ですが、長寿命スレッドを扱う以上は避けて通れません。この種の「長時間実行と再開」の設計は、Temporalで組む耐久AIワークフローの考え方とも通じるものがあります。

from_conn_string を変数に代入すると、compile() の時点で落ちる

最初にぶつかったのはここでした。チュートリアルの断片を写して memory = SqliteSaver.from_conn_string("agent.db") と書き、そのまま workflow.compile(checkpointer=memory) を呼んだところ、次のエラーで停止します。

TypeError: Invalid checkpointer provided. Expected an instance of
`BaseCheckpointSaver`, `True`, `False`, or `None`.
Received _GeneratorContextManager.
Pass a proper saver (e.g., InMemorySaver, AsyncPostgresSaver).

from_conn_string は @contextmanager で装飾されており、シグネチャも Iterator[SqliteSaver] を返します。つまり with を伴わない限り、手元に来るのはジェネレータのラッパーであって saver ではありません。PostgresSaver も同じ実装なので、pg_memory.setup() は AttributeError になります。

紛らわしいのは、この行自体は例外を投げないことです。失敗が表面化するのは数十行あとの compile() で、しかもメッセージが InMemorySaver を勧めてくるため、原因がインポート先や依存関係にあると誤解しやすくなっています。私はここで半日ほど溶かしました。

短命なスクリプトなら with、長く生きるプロセスなら接続を自分で持ってコンストラクタに渡す——この2択で覚えておくと迷いません。

recursion_limit の既定値は、もう25ではない

長く「LangGraphは25スーパーステップで GraphRecursionError を投げる」と説明されてきました。この記事も当初はその前提で書いていましたが、手元の LangGraph 1.2.11 で確かめたところ、既定値は 10007 に変わっていました（langgraph/_internal/_config.py の DEFAULT_RECURSION_LIMIT、環境変数 LANGGRAPH_DEFAULT_RECURSION_LIMIT で上書き可能）。60周するループを既定設定のまま流しても、素通りします。

	旧来の理解（〜0.x系）	手元で確認した挙動（1.2.11）
既定の recursion_limit	25	10007
リサーチループで起きること	7〜8周で早期に打ち切られる	止まらずに回り続ける
怖いこと	処理が途中で終わる	API 課金が静かに膨らむ

リスクの向きが反転している点が重要です。以前は「上限が近すぎて途中で落ちる」ことを心配していれば足りました。いまは「上限が遠すぎて、条件分岐のバグに気づかないまま回り続ける」ほうが怖いと感じております。GraphRecursionError は、実のところ安全装置として機能していたわけです。

そこで私は、既定値に頼るのをやめました。invoke 時に config={"recursion_limit": 40} のように業務都合の値を必ず明示し、加えてステート側にも試行回数カウンタを持たせて二重に打ち切ります。グラフの都合ではなく業務の都合で止める、という考え方です。

class ResearchState(TypedDict, total=False):
    messages: Annotated[list, operator.add]
    attempts: int          # 業務都合の上限（グラフの再帰上限とは別軸）
    schema_version: int
 
MAX_ATTEMPTS = 8
 
def should_continue(state: ResearchState) -> str:
    # 第1の上限: 業務ロジックとしての試行回数
    if state.get("attempts", 0) >= MAX_ATTEMPTS:
        return END
    if state.get("final_result"):
        return END
    return "collect"
 
# 第2の上限: グラフ全体の再帰上限を必ず明示する
result = app.invoke(
    initial_state,
    config={
        "configurable": {"thread_id": "research-001"},
        "recursion_limit": 40,   # 既定の 10007 に任せない
    },
)

バージョンによって既定値が動く以上、明示しておけば読む人にも意図が伝わります。上限値そのものよりも、「どちらの上限で止まったのか」がログから判別できることのほうが、運用では効いてきます。

ステートに積んだ messages は、そのまま毎回の入力トークンになる

ステートの messages に履歴を積む設計は素直ですが、各ノードがその全量をClaudeに渡すため、ステップが進むほど1呼び出しの入力トークンが増えます。20ステップの実行を計測したところ、最終盤のノードでは入力が初回の約1,800トークンから約16,000トークンまで、およそ9倍に膨らんでいました。プロンプトキャッシュである程度は吸収できるものの、根本対策として「直近6メッセージ＋それ以前の要約」に間引くトリミングノードを挟んだところ、実行全体の入力トークンを約4割削減できました。長いワークフローほど効いてきます。

まとめ

この記事を書き直すきっかけは、掲載していたチェックポインタのコードが一行も動かないと気づいたことでした。from_conn_string の戻り値をそのまま変数に入れる——公式のサンプル断片を切り貼りすると自然にそうなってしまう書き方で、しかも失敗はずっと後の compile() で表面化します。ドキュメントに書かれていない落とし穴ではなく、書かれ方の隙間に落ちる類のものでした。

もし手元のコードを一箇所だけ確かめるなら、まずここです。

# 自分の checkpointer が本物か、1行で判定できます
from langgraph.checkpoint.base import BaseCheckpointSaver
assert isinstance(memory, BaseCheckpointSaver), type(memory).__name__

これが通れば、永続化も interrupt() による再開も設計どおりに機能します。通らなければ、with で包むかコンストラクタに接続を渡すか、どちらかに寄せてください。

そのうえで、次に確認していただきたいのが recursion_limit です。既定値が 10007 に変わった以上、ループを含むグラフを既定のまま本番に置くのは、安全装置を外して走らせるのに近い状態です。invoke に一行足すだけで済みます。

長く動かすエージェントほど、派手な設計よりもこうした地味な前提の確認が効いてきます。私自身まだ手探りの部分は多いのですが、同じ場所で半日を溶かす方が一人でも減れば嬉しいです。

さらに踏み込む場合は、サブエージェントのファンアウト/ファンイン設計と部分失敗の扱いやClaude API本番回復パターンも合わせてご覧ください。

お読みいただきありがとうございます

Claude Lab は広告なしで運営しており、サーバー費用などの運営コストはメンバーシップのご支援で賄っています。実装コード・ベンチマーク・本番設計パターンなど、実務でお役立ていただける記事を毎日更新しています。もし読んでよかったと感じていただけましたら、ぜひご覧ください。

✦コピー&ペーストで使える実装コード付き
✦毎日新しい上級ガイドを追加
✦¥580/月または ¥1,480 の永久アクセス

メンバーシップを見る →