壊れるなら1箇所で壊せ — LangGraph多段AIエージェントのステート肥大化とエラー伝播を防ぐ3原則

TL;DR

• Human-in-the-Loop（HITL）: 自動化の暴走を防ぐため、各段に停止・承認・再開のポイントを設ける
• 3段以内の連鎖: AIエージェントを「3段以上」連鎖させると、状態爆発と推論精度の劣化（Lost in the Middle）が発生する
• エラーポイントの単一化: 例外（Exception）で落とすのはアンチパターン。「Stateが真実」の原則に従い、エラー情報はStateに乗せて1箇所に集約する
• LangGraph 0.2.x以降（Python 3.10+）および n8n（Slack Block Kit）を用いた、運用可能な実装アーキテクチャを示す

---

なぜ「動くエージェント」が「運用できるエージェント」にならないのか

AIエージェントの開発において、「動くPoC（概念実証）」を作るのは簡単になりました。
しかし、それを本番環境にデプロイし、安定稼働させる段階になると、多くのプロジェクトが壁にぶつかります。

単一のLLM呼び出しであれば、制御は比較的容易です。
問題は「多段エージェント（エージェントが次のエージェントを呼ぶアーキテクチャ）」にあります。
Agent Aの出力がAgent Bの入力になり、Agent Bの微小なハルシネーションがAgent Cで致命的なエラーに拡大する「ハルシネーションの伝播」が発生します。

さらに、各ステップでState（状態）に情報が蓄積されていくため、エラーログを見ても「どの段の、どの変数が原因で、何が壊れたか」をトレースするためのパターンが組み合わせ爆発的に増加し、人間の認知限界を超えてしまいます。

本記事では、この「ステート肥大化」と「エラー伝播」の問題を構造的に解決し、多段エージェントを本番運用に乗せるための3つの設計原則を示します。

---

前提：対象読者とスコープ

この記事は以下の読者を想定しています：

• LangGraphを利用してマルチエージェントを構築中のエンジニア
• エージェントのR&D成果を本番運用に移行しようとしているテックリード
• 「たまに謎のエラーで止まるが、原因が追いきれない」と悩んでいる方

利用技術：LangGraph 0.2.x以降（Python 3.10+）、n8n、Slack API

---

設計原則①：Human-in-the-Loop（HITL）停止ポイントを各段に置く

問題

自律型エージェントは「止まらない」ように設計されがちです。しかし、ハルシネーションや不適切なフォーマットが発生したとき、エージェントは止まらずに壊れた出力を次の処理に渡し続けてしまいます。
また、無限ループ（再生成の連続）に陥るリスクもあります。

解決策

エージェントグラフの要所に 「停止→人間承認→再開」のポイント を設けます。
以下のコード例は1段のシンプルなデモですが、実運用ではこれを各段（サブグラフの入口や出口など、ビジネスロジックの境界）に配置します。

LangGraphでは interrupt() で停止させ、外部から Command(resume=...) を注入することで再開させます。
この際、LangGraph特有の「Resume時のReplay（ノード再実行）仕様」 を正しく理解し、副作用に備える必要があります。

import logging
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

# LangGraphのStateは部分更新が前提のため、初期呼び出し（graph.invoke）時に
# 全キーを持たないStateを渡せるよう total=False を指定します。
class AgentState(TypedDict, total=False):
    input: str
    draft_output: str
    approved: bool
    final_output: str
    retry_count: int  # 無限ループ防止用

def dummy_call_llm(text: str) -> str:
    return f"【生成されたドラフト】: {text}に基づく回答"

def generate_node(state: AgentState) -> AgentState:
    """LLMでドラフトを生成するノード"""
    draft = dummy_call_llm(state.get("input", ""))
    current_retry = state.get("retry_count", 0)
    return {"draft_output": draft, "retry_count": current_retry + 1}

def review_node(state: AgentState) -> AgentState:
    """Human-in-the-Loopの停止ポイント"""
    # 【重要】LangGraphはresume時にこのノードを先頭から再実行(Replay)する。
    # 2回目の到達時、interrupt()は停止せずresumeされた値を即時返却するため
    # 無限ループにはならないが、ノード内に外部API呼び出し等の副作用がある場合は注意。
    
    # human_review には Command(resume=...) に渡した値がそのまま入る
    human_review = interrupt({
        "question": "この出力を承認しますか？",
        "draft": state.get("draft_output", "（ドラフト未生成）")
    })
    return {"approved": human_review.get("approved", False)}

def abort_node(state: AgentState) -> AgentState:
    """エスカレーション処理（無音終了を防ぐ）"""
    logging.error(f"[HITL Abort] 再生成上限に達しました。input='{state.get('input', '')[:100]}'")
    # 実際の運用ではここでSlack通知等のアラートを発火させる
    return state

def route_after_review(state: AgentState) -> Literal["finalize", "regenerate", "abort"]:
    """承認/拒否とリトライ上限でルーティング"""
    if state.get("approved"):
        return "finalize"
    if state.get("retry_count", 0) >= 3:
        # 初回生成(1) + 否認による再生成(2) の計3回失敗でエスカレーション
        return "abort"
    return "regenerate"

def finalize_node(state: AgentState) -> AgentState:
    # 型レベルの整合性を保つため、None時のフォールバックを添える
    return {"final_output": state.get("draft_output") or ""}

# グラフの構築
builder = StateGraph(AgentState)
builder.add_node("generate", generate_node)
builder.add_node("review", review_node)
builder.add_node("abort", abort_node)
builder.add_node("finalize", finalize_node)

builder.add_edge(START, "generate")
builder.add_edge("generate", "review")
builder.add_conditional_edges("review", route_after_review, {
    "finalize": "finalize",
    "regenerate": "generate",
    "abort": "abort",  # END直行ではなくエラーハンドリングを経由させる
})
builder.add_edge("abort", END)
builder.add_edge("finalize", END)

memory = InMemorySaver()
graph = builder.compile(checkpointer=memory)

Note: abort_node（HITLの上限超過）と、後述する原則③の error_handler_node（ノード内例外）は発生源が異なるため分離しています。
ただし、どちらも通知・ログの処理は同じWebhookに集約することで、アラートの一元管理は維持できます。

外部システム（n8n/Slack）との連携と停止の検知

グラフが停止したことを検知し、外部システムに承認を求める場合、スレッドID（thread_id）をWebhookのペイロードに含めることが不可欠です。
invoke()メソッドは停止時に例外を投げず、戻り値に__interrupt__リストを含める仕様になっています。

# --- LangGraph側から n8n Webhook へのデータ送信イメージ ---
config = {"configurable": {"thread_id": "thread_abc123"}}

# 1. グラフを実行し、停止を検知する
result = graph.invoke({"input": "要件定義書を書いて"}, config=config)

if "__interrupt__" in result:
    # 逐次グラフでは [0] のみ。並列ノードが複数 interrupt() を発火させる場合は
    # リストをループして全Interruptを処理する必要があります。
    interrupt_payload = result["__interrupt__"][0].value
    
    # 停止を検知し、n8n等のWebhookへPOSTする
    webhook_payload = {
        "thread_id": config["configurable"]["thread_id"],
        "draft_output": interrupt_payload.get("draft")
    }
    # requests.post(N8N_WEBHOOK_URL, json=webhook_payload)

# 2. ユーザーがSlackで承認後、n8nから再開する
# ※ 実運用では FastAPI 等で自作のエンドポイント（/resume）を構築し、
#    内部で以下の Command 発火を行うのが一般的です。
# graph.invoke(Command(resume={"approved": True}), config=config)

これをn8nで受け取り、Slackの Block Kit（現在の推奨仕様）で通知するJSONイメージは以下のようになります。

{
  "blocks": [
    {
      "type": "section",
      "text": {
        "type": "mrkdwn", 
        "text": "*AIエージェントが承認待ちです*\n{{ $json.draft_output }}"
      }
    },
    {
      "type": "actions",
      "block_id": "{{ $json.thread_id }}",
      "elements": [
        {
          "type": "button", 
          "text": {"type": "plain_text", "text": "✅ 承認"}, 
          "value": "approve", 
          "action_id": "btn_approve"
        },
        {
          "type": "button", 
          "text": {"type": "plain_text", "text": "❌ 否認"}, 
          "style": "danger", 
          "value": "reject", 
          "action_id": "btn_reject"
        }
      ]
    }
  ]
}

---

設計原則②：連鎖は3段以内に収める

「複雑なタスクを5段・6段のエージェント連鎖で解決したい」という要望はよくありますが、連鎖の長さは2つの致命的な問題を引き起こします。

1. 状態の組み合わせ爆発: 各段で「正常終了」か「エラー」の2択が発生する保守的なモデルでも、$k$段の連鎖では最大$2^k$通りの経路が生まれます。$k=6$ では64通りのパスを追跡する必要があり、運用時の認知限界を超えます。

2. 推論精度の劣化（Lost in the Middle）: 各ノードがMessagesにコンテキストを積み上げる実装では、LLMが参照する実効コンテキストが肥大化します。コンテキストが長くなるほど中間部分のトークンが参照されにくくなる傾向があるため、最新モデルで改善されつつあるとはいえ、サブグラフ単位で不要な中間Stateを切り捨てるアーキテクチャ設計が依然として有効です。

3段を超えるなら「サブグラフ化」せよ

複雑なステップが必要な場合は、処理を「1段として扱えるカプセル化されたモジュール」に分割します。
親グラフと子グラフのStateスキーマを分離し、ラッパー関数で変換を行うことで、予期せぬ名前空間の衝突を防止できます。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# --- 1. サブグラフの定義（内部のState空間） ---
class SubState(TypedDict, total=False):
    sub_input: str  # 親グラフとは独立した名前空間
    sub_result: str

def step_a(state: SubState) -> SubState:
    return {"sub_result": state.get("sub_input", "") + " [A完了]"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("step_a", step_a)
sub_builder.add_edge(START, "step_a")
sub_builder.add_edge("step_a", END)
sub_graph = sub_builder.compile()

# --- 2. メイングラフの定義（親のState空間） ---
class MainState(TypedDict, total=False):
    text: str

def run_subgraph(state: MainState) -> MainState:
    """親のStateと子のStateをマッピングするラッパー関数"""
    sub_output = sub_graph.invoke({"sub_input": state.get("text", "")})
    # サブグラフ内でも原則③（例外で落とさない）が守られている前提で安全にアクセス
    return {"text": sub_output.get("sub_result", "")}

main_builder = StateGraph(MainState)
main_builder.add_node("phase_1_subgraph", run_subgraph)
main_builder.add_edge(START, "phase_1_subgraph")
main_builder.add_edge("phase_1_subgraph", END)

# ⚠️ 注意（LangGraph 0.2.xでの挙動）: 
# このラッパー関数方式（invoke経由）では、サブグラフ内での interrupt() は 
# resume 時に正しく動作しません（サブグラフが先頭から再実行されてしまうため）。
# サブグラフ内でHITLを使いたい場合は add_node に直接グラフオブジェクトを渡し、
# 親子でStateスキーマを共有する設計を選択してください。
main_graph = main_builder.compile()

---

設計原則③：壊れるポイントを1箇所に集める

問題：例外（Exception）でプロセスを落とすな

多段エージェントで各ノードにバラバラな try-except を書き、エラー時にそのまま raise Exception してしまうのはアンチパターンです。
ステートマシンベースのフレームワークでは、「State（状態）が絶対的な真実」 であり、例外でクラッシュさせると現在の状況が失われます。

解決策：エラー情報をStateに乗せ、中央ノードに集約する

エラー情報を明示的にStateの一部として包含させ、一箇所でハンドリングします。

import os
import requests
import logging
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class ErrorInfo(TypedDict):
    node: str
    type: str
    message: str
    input_snapshot: str

class PipelineState(TypedDict, total=False):
    input: str
    a_output: str | None
    error: ErrorInfo | None

def node_a(state: PipelineState) -> PipelineState:
    try:
        # 実際に失敗しうる外部API呼び出し等のロジック
        response = requests.get("https://api.example.com/data", timeout=5)
        response.raise_for_status()
        return {"a_output": "success", "error": None}
    except Exception as e:
        # 例外で落とさず、エラー内容をStateに乗せて正常終了(return)する
        return {
            "a_output": None,
            "error": {
                "node": "node_a",
                "type": type(e).__name__,
                "message": str(e),
                "input_snapshot": state.get("input", "")[:200],
            }
        }

def node_b(state: PipelineState) -> PipelineState:
    """後続の正常系ノード"""
    # 文字列汚染（"None -> B完了"）を防ぐ安全なアクセス
    current_output = state.get("a_output") or ""
    return {"a_output": current_output + " -> B完了", "error": None}

def error_handler_node(state: PipelineState) -> PipelineState:
    """エラーハンドリングはここだけ。通知・リトライ・ログはすべてここに集約"""
    err = state.get("error")
    
    # 本番環境（最適化オプション起動時）で無効化される assert は避け、明示的にチェック
    if err is None:
        logging.warning("[error_handler_node] errorがNoneで呼ばれました。")
        return state
        
    logging.error(f"[Pipeline Error] Node: {err['node']}, Msg: {err['message']}")
    
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    if webhook_url:
        # 実運用ではコメントを外して有効化（本例ではURLがダミーのためスキップ）
        # requests.post(webhook_url, json={"error": err})
        pass
    
    return state

def route_on_error(state: PipelineState) -> Literal["continue", "error_handler"]:
    if state.get("error"):
        return "error_handler"
    return "continue"

# グラフ構築
builder = StateGraph(PipelineState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("error_handler", error_handler_node)

builder.add_edge(START, "node_a")
builder.add_conditional_edges("node_a", route_on_error, {
    "continue": "node_b",
    "error_handler": "error_handler"
})
builder.add_edge("node_b", END)
builder.add_edge("error_handler", END)

# 原則①の graph 変数との衝突を避けるため pipeline_graph と命名
pipeline_graph = builder.compile()

---

チェックリスト：本番リリース前に確認すべき7項目

• 各境界にHITL停止ポイントが配置されているか（thread_id が外部に渡っているか）
• n8nのWaitノードにタイムアウト（例: 60分）が設定されているか（無限ハング防止）
• 連鎖段数が最大3段に収まっているか（4段以上はサブグラフにカプセル化されているか）
• 再生成ループに上限と、上限超過時のエスカレーション処理が設定されているか
• 例外でプロセスを落としていないか（エラーがStateに格納されているか）
• エラーハンドリング（通知・ログ）が単一のノードに集約されているか
• StateがDBに永続化されているか（InMemorySaverのままデプロイしていないか。PostgreSQLの場合: pip install langgraph-checkpoint-postgres）

---

まとめ

多段AIエージェントの安定稼働は、プロンプトの問題ではなくシステムアーキテクチャの問題です。

暴走を防ぐ防波堤としてHITLを置き、
認知限界を超えないよう連鎖を3段以内に絞り、
例外をStateの一部として中央で管理する。

この3原則を徹底することで、予測不可能なLLMを内包しながらも、予測可能で堅牢なパイプラインを構築することができます。

従来のシステム設計で常識となっている「壊れるなら1箇所で壊せ」という原則は、AIエージェント開発においても最も確実な成功への近道となります。

この記事を書いた人✏️@YushiYamamoto
ITPRODX.com代表 / AIアーキテクト
Next.js / TypeScript / n8nを活用した自律型アーキテクチャ設計を専門としています。
日々の自動化の検証結果や、ビジネス側の視点（ROI等）に関するより深い考察は、以下の公式サイトおよびnoteで発信しています。