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

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

LangGraphとClaude APIでステートフルエージェントを本番運用するための実装パターン集。チェックポイント容量の実測値、interrupt()再開の落とし穴、状態肥大によるトークンコストまで、運用で得た知見と共に整理します。

LangGraph Claude 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
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チェックポインターの設定（開発環境では :memory: を使用）
memory = SqliteSaver.from_conn_string(":memory:")
 
print("✅ LangGraph × Claude API セットアップ完了")
# 出力: ✅ LangGraph × Claude API セットアップ完了

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

✦

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

この記事の続きを読む

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

この記事で得られること

✦SqliteSaver/PostgresSaverのチェックポイント容量が実運用でどう増えるかの実測値と、delete_threadによる週次棚卸し運用の設計

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

✦messages累積で入力トークンが初回の約9倍まで肥大した実測例と、トリミングノード導入によるコスト抑制の考え方

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によるセッション管理を示します。

from langgraph.checkpoint.sqlite import SqliteSaver
 
# 本番環境ではファイルパスを指定
# 開発: ":memory:" | 本番: "/var/data/agent_checkpoints.db"
memory = SqliteSaver.from_conn_string("agent_checkpoints.db")
 
# チェックポイント付きでグラフをコンパイル
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 チェックポインター
# pip install langgraph-checkpoint-postgres psycopg2-binary
 
from langgraph.checkpoint.postgres import PostgresSaver
 
pg_memory = PostgresSaver.from_conn_string(
    os.environ["DATABASE_URL"]  # postgresql://user:pass@host:5432/dbname
)
pg_memory.setup()  # 必要なテーブルを自動作成
 
production_app = workflow.compile(checkpointer=pg_memory)

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

エンタープライズ用途では、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() で読む
再帰上限への到達	ループ構造で GraphRecursionError	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ワークフローの考え方とも通じるものがあります。

recursion_limit の既定値25は、ループ構造だと意外に近い

LangGraphは無限ループ防止のため、既定でスーパーステップ25回を超えると GraphRecursionError を投げます。リサーチ系のループ（収集→評価→不足なら再収集）を組むと、1周で3〜4ステップ消費するため、7〜8周で上限に届いてしまいます。invoke 時に config={"recursion_limit": 80} のように明示しつつ、ステート側にも試行回数カウンタを持たせて「グラフの都合ではなく業務の都合」で打ち切る二重の上限を設けるのが私の落としどころです。

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

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

まとめ

LangGraph × Claude API の組み合わせは、2026年現在において本番品質のステートフルAIエージェントを構築する最も実践的なアプローチのひとつです。本記事で解説したパターンをまとめます。

グラフベース設計では、TypedDictでステートを型安全に定義し、ノードとエッジでワークフローを宣言的に構造化します。永続化・再開では、SqliteSaverやPostgresSaverによるチェックポイントで、長時間タスクと障害回復を実現します。ヒューマンインザループでは、interrupt()を使って重要な判断ポイントで実行を一時停止し、人間の承認を組み込みます。マルチエージェント協調では、Supervisorパターンで複数の専門エージェントを指揮し、タスクを分担処理します。本番品質対応では、tenacityによるリトライ・構造化ログ・LangGraph Studioによるデバッグを組み合わせます。

次のステップとして、サブエージェントのファンアウト/ファンイン設計と部分失敗の扱いやClaude API本番回復パターンも合わせてご覧ください。

LangGraphとClaudeの組み合わせをさらに深く