【2025年完全版】LangGraph完全マスターガイド：マルチエージェント構築からトラブル解決まで

🆕 2025年11月最新版！
LangGraph v0.5.4対応、エラー10選追加、パフォーマンス最適化を反映。

5分でわかる：LangGraph完全マスターガイド

LangChainでは実現できない複雑なAIエージェントを構築したい。そんなあなたのために、LangGraphの全てを実践的に解説します。

なぜこのガイドが必要なのか

本記事で学べること

1. 基礎理解（第1章）：LangGraphとは、LangChainとの違い
2. 実装パターン（第2章）：5つの実践パターン
3. マルチエージェント（第3章）：協調システム構築
4. Human-in-the-Loop（第4章）：人間介入の実装
5. トラブル解決（第5章）：よくあるエラー10選
6. 最適化（第6章）：パフォーマンスチューニング

第1章：LangGraph完全理解

1.1 LangGraphとは

LangGraphの核心:
├── ステートフルAIシステム
│   └── グラフ構造で状態管理
│
├── 循環グラフサポート
│   └── ループ・動的分岐可能
│
├── Human-in-the-Loop
│   └── 人間介入ネイティブサポート
│
└── マルチエージェント
    └── 協調・分業が容易

LangGraphが必要な理由:
✅ 複雑なワークフロー
✅ 動的な意思決定
✅ 長期記憶の実現
✅ エージェント協調
✅ 人間フィードバック統合

1.2 LangChain vs LangGraph

アーキテクチャ比較:
┌────────────────────────────┐
│ LangChain（線形DAG）       │
│ Input → Chain → Output     │
│ ✅ シンプル                │
│ ❌ ループ不可              │
│ ❌ 動的分岐限定的           │
└────────────────────────────┘

┌────────────────────────────┐
│ LangGraph（循環グラフ）     │
│ Input ⇄ Graph ⇄ Output     │
│ ✅ ループ可能              │
│ ✅ 動的分岐                │
│ ✅ ステート永続化            │
└────────────────────────────┘

機能比較:
| 項目 | LangChain | LangGraph |
|------|-----------|-----------|
| 構造 | 線形 | 循環 |
| ステート | 限定的 | 永続化可能 |
| ループ | 困難 | ネイティブ |
| Human-in-loop | 限定的 | フルサポート |
| マルチエージェント | 基本的 | 高度な協調 |

使い分け:
- シンプルなチェーン → LangChain
- 複雑なワークフロー → LangGraph
- ReAct単体 → LangChain
- マルチエージェント → LangGraph
- ループ不要 → LangChain
- ループ必要 → LangGraph

第2章：5つの実装パターン

パターン1：基本グラフ（最小構成）

from langgraph.graph import StateGraph
from typing import TypedDict

# 1. ステート定義
class State(TypedDict):
    messages: list
    count: int

# 2. ノード定義
def chatbot(state: State):
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],
        "count": state.get("count", 0) + 1
    }

# 3. グラフ構築
graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.set_entry_point("chatbot")
graph.add_edge("chatbot", "__end__")

# 4. コンパイル & 実行
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "Hello"}]})

パターン2：条件分岐（動的ルーティング）

class AnalysisState(TypedDict):
    input: str
    analysis: str
    confidence: float
    decision: str

def analyze(state: AnalysisState):
    result = llm.invoke(f"Analyze: {state['input']}")
    return {
        "analysis": result.content,
        "confidence": extract_confidence(result.content)
    }

def decide(state: AnalysisState):
    if state["confidence"] > 0.8:
        return {"decision": "EXECUTE"}
    elif state["confidence"] > 0.5:
        return {"decision": "REVIEW"}
    else:
        return {"decision": "RETRY"}

def route_decision(state: AnalysisState) -> str:
    """動的ルーティング"""
    decision = state.get("decision", "")
    
    if decision == "EXECUTE":
        return "execute"
    elif decision == "REVIEW":
        return "human_review"
    else:
        return "analyze"  # ループバック

# グラフ構築
graph = StateGraph(AnalysisState)
graph.add_node("analyze", analyze)
graph.add_node("decide", decide)
graph.add_node("execute", execute_action)
graph.add_node("human_review", human_review)

# 条件分岐
graph.add_edge("analyze", "decide")
graph.add_conditional_edges(
    "decide",
    route_decision,
    {
        "analyze": "analyze",  # ループ
        "execute": "execute",
        "human_review": "human_review"
    }
)
graph.add_edge("execute", "__end__")
graph.add_edge("human_review", "analyze")

graph.set_entry_point("analyze")
app = graph.compile()

パターン3：ReActエージェント

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def search_web(query: str) -> str:
    """Web検索を実行"""
    # 実装
    return f"検索結果: {query}"

@tool
def calculate(expression: str) -> str:
    """数式を計算"""
    try:
        return str(eval(expression))
    except:
        return "Error"

# ReActエージェント作成
llm = ChatOpenAI(model="gpt-4")
tools = [search_web, calculate]

agent = create_react_agent(
    model=llm,
    tools=tools,
    state_modifier="あなたは有能なアシスタントです"
)

# 実行
config = {"configurable": {"thread_id": "1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "東京の人口を調べて、1人あたりのGDPを計算して"}]},
    config=config
)

パターン4：マルチエージェント協調

class TeamState(TypedDict):
    task: str
    research: List[str]
    analysis: str
    recommendations: List[str]

def research_agent(state: TeamState):
    """研究エージェント"""
    results = llm.invoke(f"Research: {state['task']}")
    return {"research": results.content.split('\n')}

def analysis_agent(state: TeamState):
    """分析エージェント"""
    analysis = llm.invoke(f"Analyze: {state['research']}")
    return {"analysis": analysis.content}

def strategy_agent(state: TeamState):
    """戦略エージェント"""
    recs = llm.invoke(f"Recommend: {state['analysis']}")
    return {"recommendations": recs.content.split('\n')}

# マルチエージェントグラフ
graph = StateGraph(TeamState)
graph.add_node("research", research_agent)
graph.add_node("analysis", analysis_agent)
graph.add_node("strategy", strategy_agent)

# フロー
graph.set_entry_point("research")
graph.add_edge("research", "analysis")
graph.add_edge("analysis", "strategy")
graph.add_edge("strategy", "__end__")

app = graph.compile()

パターン5：Human-in-the-Loop

from langgraph.checkpoint.memory import MemorySaver

class ReviewState(TypedDict):
    draft: str
    feedback: str
    final: str

def ai_writer(state: ReviewState):
    """AI執筆"""
    draft = llm.invoke(f"Write: {state.get('feedback', 'Initial')}")
    return {"draft": draft.content}

def human_reviewer(state: ReviewState):
    """人間レビュー"""
    print(f"Draft: {state['draft']}")
    feedback = input("Feedback (or 'approve'): ")
    return {"feedback": feedback}

def finalize(state: ReviewState):
    """最終化"""
    if state["feedback"] == "approve":
        return {"final": state["draft"]}
    else:
        return {}

def should_continue(state: ReviewState) -> str:
    if state.get("final"):
        return "end"
    elif state.get("feedback") and state["feedback"] != "approve":
        return "ai_writer"
    else:
        return "human_reviewer"

# Human-in-the-Loopグラフ
graph = StateGraph(ReviewState)
graph.add_node("ai_writer", ai_writer)
graph.add_node("human_reviewer", human_reviewer)  
graph.add_node("finalize", finalize)

graph.set_entry_point("ai_writer")
graph.add_edge("ai_writer", "human_reviewer")
graph.add_edge("human_reviewer", "finalize")
graph.add_conditional_edges(
    "finalize",
    should_continue,
    {
        "ai_writer": "ai_writer",
        "human_reviewer": "human_reviewer",
        "end": "__end__"
    }
)

# 中断ポイント設定
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["human_reviewer"]
)

# 実行
config = {"configurable": {"thread_id": "review_1"}}
result = app.invoke({"draft": ""}, config=config)

第3章：マルチエージェントシステム

3.1 エージェント間通信パターン

# パターン1：順次実行
def sequential_pattern():
    """A → B → C の順次実行"""
    graph = StateGraph(State)
    graph.add_edge("agent_a", "agent_b")
    graph.add_edge("agent_b", "agent_c")
    graph.add_edge("agent_c", "__end__")

# パターン2：並列実行
def parallel_pattern():
    """A ⇄ B, C（並列）"""
    graph = StateGraph(State)
    # 同時に複数エージェントを起動
    graph.add_edge("__start__", "agent_b")
    graph.add_edge("__start__", "agent_c")

# パターン3：階層型
def hierarchical_pattern():
    """Manager → Worker1, Worker2"""
    def manager_routes(state):
        if state["complexity"] > 5:
            return "specialist"
        else:
            return "generalist"
    
    graph.add_conditional_edges(
        "manager",
        manager_routes,
        {"specialist": "agent_specialist", "generalist": "agent_generalist"}
    )

第4章：よくあるエラー10選と解決策

エラー1：StateKeyError

エラー:
KeyError: 'messages' not in state

原因:
ステート定義とアクセスの不一致

解決策:
```python
# ❌ 間違い
class State(TypedDict):
    input: str

def node(state: State):
    return {"messages": []}  # messagesはState未定義

# ✅ 正しい
class State(TypedDict):
    input: str
    messages: list  # 定義に追加

def node(state: State):
    return {"messages": []}  # OK

### エラー2-10のクイックリファレンス

| エラー | 原因 | 解決策 |
|--------|------|--------|
| **2. Missing checkpointer** | ステート永続化未設定 | MemorySaver追加 |
| **3. Cycle detected** | 無限ループ設定 | 終了条件追加 |
| **4. Node not found** | ノード名typo | 正しい名前確認 |
| **5. Invalid edge** | 存在しないノードへのエッジ | ノード存在確認 |
| **6. Interrupt not working** | interrupt_before設定忘れ | compile()に追加 |
| **7. State not persistent** | thread_id未設定 | config設定 |
| **8. Tool call failed** | Tool定義エラー | @tool使用確認 |
| **9. Memory overflow** | 大量ステート蓄積 | autoDispose使用 |
| **10. Slow execution** | 同期処理 | async/await使用 |

## 第5章：パフォーマンス最適化

### 5.1 最適化テクニック

```python
# テクニック1：ノードキャッシング
def expensive_node(state: State):
    """重い処理をキャッシュ"""
    # キャッシュキー生成
    cache_key = f"{state['task_id']}_{state['params']}"
    
    # キャッシュチェック
    if cache_key in cache:
        return cache[cache_key]
    
    # 実行
    result = heavy_computation(state)
    cache[cache_key] = result
    return result

# テクニック2：並列実行
async def parallel_execution():
    """複数ノードを並列実行"""
    tasks = [
        agent1.ainvoke(state),
        agent2.ainvoke(state),
        agent3.ainvoke(state)
    ]
    results = await asyncio.gather(*tasks)
    return merge_results(results)

# テクニック3：ストリーミング
async def streaming_response():
    """レスポンスをストリーミング"""
    async for chunk in app.astream(
        {"messages": [{"role": "user", "content": "Hello"}]},
        config=config,
        stream_mode="values"
    ):
        print(chunk, end="", flush=True)

# テクニック4：遅延ノード
graph.add_node("aggregator", aggregate, deferred=True)
# 上流の全パス完了まで延期

# テクニック5：メモリ管理
from langgraph.checkpoint.postgres import PostgresSaver

# メモリではなくDB永続化
checkpointer = PostgresSaver(connection_string)
app = graph.compile(checkpointer=checkpointer)

5.2 ベストプラクティス

# ベストプラクティス1：型安全
from typing import Annotated
from langgraph.graph.message import add_messages

class SafeState(TypedDict):
    # Annotatedで型安全に
    messages: Annotated[List[BaseMessage], add_messages]
    count: int
    metadata: Dict[str, Any]

# ベストプラクティス2：エラーハンドリング
def robust_node(state: State):
    try:
        result = llm.invoke(state["input"])
        return {"output": result.content}
    except Exception as e:
        logger.error(f"Node failed: {e}")
        return {"error": str(e), "status": "failed"}

# ベストプラクティス3：ログ・監視
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitored_node(state: State):
    logger.info(f"Node start: {state}")
    result = process(state)
    logger.info(f"Node end: {result}")
    return result

# ベストプラクティス4：テスト
def test_graph():
    """グラフのテスト"""
    test_state = {"input": "test"}
    result = app.invoke(test_state)
    
    assert "output" in result
    assert result["status"] == "success"

# ベストプラクティス5：バージョニング
class StateV1(TypedDict):
    input: str

class StateV2(TypedDict):
    input: str
    metadata: Dict  # 新規追加

# マイグレーション関数
def migrate_v1_to_v2(v1_state: StateV1) -> StateV2:
    return {**v1_state, "metadata": {}}

第6章：本番環境デプロイ

6.1 LangGraph Platform活用

# 本番用設定
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
import os

def create_production_app():
    """本番環境用アプリ"""
    # PostgreSQL接続
    pool = ConnectionPool(os.getenv("POSTGRES_URL"))
    checkpointer = PostgresSaver(pool)
    
    # グラフ構築
    graph = StateGraph(State)
    # ... ノード定義
    
    return graph.compile(
        checkpointer=checkpointer,
        debug=False
    )

# API化
from fastapi import FastAPI
from pydantic import BaseModel

app_api = FastAPI()

class Request(BaseModel):
    input: str
    config: dict = {}

@app_api.post("/process")
async def process(request: Request):
    app = create_production_app()
    result = await app.ainvoke(
        {"input": request.input},
        config=request.config
    )
    return {"result": result}

# 監視
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production"

まとめ：LangGraphで次世代AIエージェントを構築

LangGraphは、ステートフルで複雑なAIシステムの構築に最適です。

✅ 主要ポイント

LangGraphの強み:
1. 循環グラフ
   → ループ・動的分岐

2. ステート永続化
   → 長期記憶実現

3. Human-in-the-Loop
   → 人間介入を容易に

4. マルチエージェント
   → 協調システム構築

5. 型安全性
   → TypedDictで安全

使い分け指針:
- シンプル → LangChain
- 複雑 → LangGraph
- ループ不要 → LangChain
- ループ必要 → LangGraph
- 単一エージェント → LangChain
- マルエージェント → LangGraph

今すぐLangGraphで次世代AIエージェントを構築しよう。

※本記事の情報は2025年11月時点のものです。最新情報はLangGraph公式ドキュメントでご確認ください。