LangGraph 条件路由（Conditional Edges）実装完全ガイド：AI接客システムの実例で学ぶ

近年、ECサイトのAIカスタマーサービスは爆発的な増加を見せています。私は某大手ECプラットフォームでAIチャットボット基盤の構築を担当していますが、ユーザーの質問カテゴリは「配送状況確認」「注文キャンセル」「商品詳細問い合わせ」「クレーム対応」など多彩で、それぞれ全く異なる処理フローを必要とします。

このようなシナリオで威力を発揮するのがLangGraphの条件路由（Conditional Edges）です。本記事では、実際のAIカスタマーサービスシステムを例に、条件路由の実装方法を詳しく解説します。

条件路由とは：LangGraphにおける動的分岐の核心

LangGraphでは、グラフ内のノード間を移動するエッジに条件を付与できます。従来のif-else分岐とは異なり、条件路由はグラフの構造そのものを動的に変更できる点が革新的です。

基本的なアーキテクチャ

graph TD
    A[ユーザー入力] --> B{Intent Classification}
    B -->|配送確認| C[配送API呼び出し]
    B -->|注文操作| D[注文API呼び出し]
    B -->|商品検索| E[RAG検索]
    B -->|その他| F[一般回答生成]
    
    C --> G[応答生成]
    D --> G
    E --> G
    F --> G
    G --> H{解決?}
    H -->|いいえ| I[エージェントエスカレーション]
    H -->|はい| J[(End)]
    I --> J

上図のように、ユーザー意図（Intent）によって処理ルートが分岐し、最終的に応答生成ノードに集約される構造が典型的です。

実践的実装：EC向けAIカスタマーサービス

では具体的にHolySheep AIのAPIを活用した実装を見てみましょう。今すぐ登録하시면、¥1=$1の特恵レート（公式比85%節約）で本記事のコードを試せます。

プロジェクト構成

customer-service-bot/
├── main.py              # メインエントリーポイント
├── graph/
│   ├── __init__.py
│   ├── nodes.py         # ノード定義
│   ├── state.py         # 状態定義
│   └── router.py        # 条件路由ロジック
├── services/
│   ├── holysheep_client.py  # HolySheep API呼び出し
│   ├── shipping_api.py      # 配送API
│   └── order_api.py         # 注文API
└── requirements.txt

状態定義（state.py）

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage

class CustomerServiceState(TypedDict):
    """AIカスタマーサービスの状態定義"""
    messages: Annotated[Sequence[BaseMessage], add_messages]
    intent: str                              # 検出された意図
    intent_confidence: float                 # 信頼度スコア
    extracted_order_id: str | None           # 抽出された注文ID
    extracted_inquiry_type: str | None       # 抽出された問い合わせ種別
    requires_escalation: bool               # エスカレーション要否
    api_response: dict | None                # 外部API応答
    session_context: dict                    # セッションコンテキスト

Intent分類用の定数
INTENT_LABELS = {
    "shipping_inquiry": "配送状況確認",
    "order_cancel": "注文キャンセル",
    "order_change": "注文変更",
    "product_info": "商品詳細問い合わせ",
    "return_request": "返品・返金依頼",
    "payment_issue": "支払い関連問題",
    "complaint": "クレーム対応",
    "general": "一般的な質問"
}

HolySheep APIクライアント（services/holysheep_client.py）

import os
from openai import OpenAI

class HolySheepClient:
    """HolySheep AI APIクライアント（OpenAI互換）"""
    
    def __init__(self, api_key: str):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"  # HolySheep公式エンドポイント
        )
        # HolySheepの優位性：¥1=$1（公式¥7.3=$1比85%節約）
        # <50msレイテンシで遅延を感じさせない応答
        
    def classify_intent(self, user_message: str, conversation_history: str = "") -> dict:
        """
        ユーザー意図を分類
        
        Returns:
            dict: {"intent": str, "confidence": float, "entities": dict}
        """
        system_prompt = """あなたはECサイトのAIカスタマーサービスアシスタントです。
        ユーザーからのメッセージを分析し、以下のいずれかの意図を判定してください：
        
        - shipping_inquiry: 配送状況確認
        - order_cancel: 注文キャンセル
        - order_change: 注文変更
        - product_info: 商品詳細問い合わせ
        - return_request: 返品・返金依頼
        - payment_issue: 支払い関連問題
        - complaint: クレーム対応
        - general: 一般的な質問
        
        出力形式（JSON）：
        {
            "intent": "意图名",
            "confidence": 0.0-1.0,
            "entities": {
                "order_id": "注文番号（見つかった場合）",
                "product_id": "商品ID（見つかった場合）"
            }
        }"""
        
        response = self.client.chat.completions.create(
            model="gpt-4.1",  # 2026年価格: $8/MTok（HolySheep特恵）
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": f"会話履歴:\n{conversation_history}\n\n現在のメッセージ:\n{user_message}"}
            ],
            temperature=0.1,
            response_format={"type": "json_object"}
        )
        
        import json
        return json.loads(response.choices[0].message.content)
    
    def generate_response(self, intent: str, context: dict) -> str:
        """Intentに応じた応答を生成"""
        system_prompt = f"""あなたは{name}のAIカスタマーサービスの{name}です。
        以下の情報に基づいて、自然で親しみやすい応答を生成してください。
        
        Intent: {intent}
        Context: {context}
        
        応答は簡潔で要的を把握し、必要な場合は次のアクションを提案してください。"""
        
        response = self.client.chat.completions.create(
            model="deepseek-v3.2",  # 2026年価格: $0.42/MTok（最安値）
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": context.get("user_message", "")}
            ],
            temperature=0.7,
            max_tokens=500
        )
        
        return response.choices[0].message.content

ノード定義（graph/nodes.py）

from .state import CustomerServiceState, INTENT_LABELS
from .router import route_intent
from services.holysheep_client import HolySheepClient
from services.shipping_api import ShippingAPI
from services.order_api import OrderAPI

グローバルクライアント（実際のプロジェクトではDIを推奨）
HOLYSHEEP_CLIENT = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
SHIPPING_API = ShippingAPI()
ORDER_API = OrderAPI()

def classify_intent_node(state: CustomerServiceState) -> dict:
    """ユーザー意図を分類するノード"""
    messages = state["messages"]
    latest_message = messages[-1].content if messages else ""
    
    # 会話履歴を文字列に変換
    history_text = "\n".join([
        f"{msg.type}: {msg.content}" for msg in messages[:-1]
    ]) if len(messages) > 1 else "なし"
    
    # HolySheep APIでIntent分類
    result = HOLYSHEEP_CLIENT.classify_intent(
        user_message=latest_message,
        conversation_history=history_text
    )
    
    return {
        "intent": result["intent"],
        "intent_confidence": result["confidence"],
        "extracted_order_id": result["entities"].get("order_id"),
        "extracted_inquiry_type": INTENT_LABELS.get(result["intent"], "不明")
    }

def handle_shipping_node(state: CustomerServiceState) -> dict:
    """配送問い合わせを処理"""
    order_id = state.get("extracted_order_id")
    
    if not order_id:
        return {
            "api_response": {"error": "注文番号が見つかりませんでした"},
            "requires_escalation": False
        }
    
    # 配送API呼び出し（HolySheep APIではなく実際の配送API）
    shipping_info = SHIPPING_API.get_status(order_id)
    
    return {
        "api_response": shipping_info,
        "requires_escalation": False
    }

def handle_order_operation_node(state: CustomerServiceState) -> dict:
    """注文操作（キャンセル・変更）を処理"""
    order_id = state.get("extracted_order_id")
    intent = state["intent"]
    
    if intent == "order_cancel":
        result = ORDER_API.cancel_order(order_id)
    elif intent == "order_change":
        result = ORDER_API.get_change_options(order_id)
    else:
        result = {"error": "サポートされていない操作"}
    
    return {
        "api_response": result,
        "requires_escalation": result.get("requires_human", False)
    }

def handle_product_inquiry_node(state: CustomerServiceState) -> dict:
    """商品問い合わせを処理（RAG検索など）"""
    # RAG検索ロジック（省略）
    return {
        "api_response": {"type": "product_info", "data": "商品情報"},
        "requires_escalation": False
    }

def handle_general_inquiry_node(state: CustomerServiceState) -> dict:
    """一般的な問い合わせを処理"""
    messages = state["messages"]
    latest_message = messages[-1].content if messages else ""
    
    # HolySheep APIで応答生成
    response = HOLYSHEEP_CLIENT.generate_response(
        intent="general",
        context={
            "user_message": latest_message,
            "session_context": state.get("session_context", {})
        }
    )
    
    return {
        "api_response": {"response": response},
        "requires_escalation": False
    }

def escalate_to_human_node(state: CustomerServiceState) -> dict:
    """人間へのエスカレーション"""
    return {
        "api_response": {
            "type": "escalation",
            "message": "申し訳ございません。一時的にお待たせしてオペレーターに接続いたします。"
        },
        "requires_escalation": True
    }

条件路由の実装（graph/router.py）

from .state import CustomerServiceState

def route_intent(state: CustomerServiceState) -> str:
    """
    Intent分類結果に基づいて次に哪个ノードに遷移するかを決定
    
    Returns:
        str: 次のノード名
    """
    intent = state.get("intent", "general")
    confidence = state.get("intent_confidence", 0.0)
    
    # 信頼度が低い場合は人間にエスカレーション
    if confidence < 0.6:
        return "escalate_to_human"
    
    # Intentに応じた路由
    routing_table = {
        "shipping_inquiry": "handle_shipping",
        "order_cancel": "handle_order_operation",
        "order_change": "handle_order_operation",
        "product_info": "handle_product_inquiry",
        "return_request": "handle_order_operation",
        "payment_issue": "handle_order_operation",
        "complaint": "escalate_to_human",  # クレームは即エスカレーション
        "general": "handle_general_inquiry"
    }
    
    return routing_table.get(intent, "handle_general_inquiry")

def should_escalate(state: CustomerServiceState) -> bool:
    """
    エスカレーションが必要かどうかを判定
    
    Returns:
        bool: Trueならエスカレーションへ
    """
    # API応答に基づく判定
    api_response = state.get("api_response", {})
    
    if api_response.get("requires_human"):
        return True
    
    # Intentが解決不可能と判断された場合
    if api_response.get("error") == "不明"):
        return True
    
    return False

グラフ構築（main.py）

from langgraph.graph import StateGraph, END
from graph.state import CustomerServiceState
from graph.nodes import (
    classify_intent_node,
    handle_shipping_node,
    handle_order_operation_node,
    handle_product_inquiry_node,
    handle_general_inquiry_node,
    escalate_to_human_node
)
from graph.router import route_intent, should_escalate

def build_customer_service_graph():
    """AIカスタマーサービスグラフを構築"""
    
    # グラフの初期化
    workflow = StateGraph(CustomerServiceState)
    
    # ノードの追加
    workflow.add_node("classify_intent", classify_intent_node)
    workflow.add_node("handle_shipping", handle_shipping_node)
    workflow.add_node("handle_order_operation", handle_order_operation_node)
    workflow.add_node("handle_product_inquiry", handle_product_inquiry_node)
    workflow.add_node("handle_general_inquiry", handle_general_inquiry_node)
    workflow.add_node("escalate_to_human", escalate_to_human_node)
    
    # エントリーポイントの設定
    workflow.set_entry_point("classify_intent")
    
    # Intent分類後の条件路由
    workflow.add_conditional_edges(
        source="classify_intent",
        path=route_intent,  # router.pyで定義した路由関数
        path_map={
            "handle_shipping": "handle_shipping",
            "handle_order_operation": "handle_order_operation",
            "handle_product_inquiry": "handle_product_inquiry",
            "handle_general_inquiry": "handle_general_inquiry",
            "escalate_to_human": "escalate_to_human"
        }
    )
    
    # 処理ノードからの条件路由（エスカレーション判定）
    processing_nodes = [
        "handle_shipping",
        "handle_order_operation",
        "handle_product_inquiry",
        "handle_general_inquiry"
    ]
    
    for node_name in processing_nodes:
        workflow.add_conditional_edges(
            source=node_name,
            path=should_escalate,
            path_map={
                True: "escalate_to_human",
                False: END
            }
        )
    
    # エスカレーションノードは終了
    workflow.add_edge("escalate_to_human", END)
    
    return workflow.compile()

使用例
if __name__ == "__main__":
    from langchain_core.messages import HumanMessage
    
    # グラフのビルド
    graph = build_customer_service_graph()
    
    # 実行
    initial_state = {
        "messages": [HumanMessage(content="注文番号ORD-12345の配送状況を知りたいです")],
        "intent": "",
        "intent_confidence": 0.0,
        "extracted_order_id": None,
        "extracted_inquiry_type": None,
        "requires_escalation": False,
        "api_response": None,
        "session_context": {"user_id": "user_001"}
    }
    
    result = graph.invoke(initial_state)
    print(f"Final Intent: {result['intent']}")
    print(f"API Response: {result['api_response']}")

応用：複数条件の複合路由

実際のプロジェクトでは、単一のIntentだけでなく、複数の条件を組合せて判定する必要があります。以下に優先度と時間を考慮した複合路由の例を示します。

from typing import Literal

def complex_route(state: CustomerServiceState) -> Literal[
    "handle_urgent_shipping",
    "handle_standard_shipping",
    "handle_order_cancel",
    "handle_refund",
    "handle_general"
]:
    """
    複数条件を組合せた複合路由
    
    判定優先度:
    1. 緊急性（返金要求・長時間未配送）→ 最優先処理
    2. Intent分類結果
    3. セッション内的解決回数
    """
    intent = state.get("intent", "general")
    session_context = state.get("session_context", {})
    
    # 緊急性チェック
    is_urgent_refund = intent in ["return_request", "payment_issue"]
    is_long_waiting = session_context.get("waiting_hours", 0) > 24
    retry_count = session_context.get("auto_retry_count", 0)
    
    # 緊急ケースの処理
    if is_urgent_refund and is_long_waiting:
        return "handle_urgent_shipping"  # VIP対応フロー
    
    if is_urgent_refund:
        return "handle_refund"  # 返金優先処理
    
    if retry_count >= 3:
        return "escalate_to_human"  # リトライ過多でエスカレーション
    
    # 標準的なIntent路由
    intent_routing = {
        "shipping_inquiry": "handle_standard_shipping",
        "order_cancel": "handle_order_cancel",
        "order_change": "handle_order_change",
        "product_info": "handle_product_inquiry",
        "general": "handle_general"
    }
    
    return intent_routing.get(intent, "handle_general")

よくあるエラーと対処法

エラー1：APIキーが未設定导致的认证错误

# ❌ 誤った実装
client = HolySheepClient(api_key="")  # 空のAPIキー

錯誤: AuthenticationError: Invalid API key provided

✅ 正しい実装
import os

api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
    raise ValueError("HOLYSHEEP_API_KEY環境変数が設定されていません")

client = HolySheepClient(api_key=api_key)

環境変数の設定（.envファイルまたはexportコマンド）
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"

原因：APIキーが環境変数に設定されていない、または空の文字列になっている。
解決：HolySheep AIのダッシュボードからAPIキーを取得し、環境変数として正しく設定してください。

エラー2：Base URLの末尾にスラッシュがある导致的接続エラー

# ❌ 誤った実装
base_url="https://api.holysheep.ai/v1/"  # 末尾にスラッシュ

錯誤: NotFoundError: 404 page not found

✅ 正しい実装
base_url="https://api.holysheep.ai/v1"  # 末尾にスラッシュなし

完整的client設定
class HolySheepClient:
    def __init__(self, api_key: str):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"  # ← 重要：末尾に/なし
        )

原因：OpenAI互換クライアントでは、base_urlの末尾にスラッシュがあると正しくエンドポイントを解決できない。
解決：必ずhttps://api.holysheep.ai/v1（スラッシュなし）で設定してください。

エラー3：Conditional Edgesのpath_map不完整导致的実行時エラー

# ❌ 誤った実装：router返回值不在path_map中
def route_intent(state: CustomerServiceState) -> str:
    return "handle_new_intent"  # path_mapに未定義

workflow.add_conditional_edges(
    source="classify_intent",
    path=route_intent,
    path_map={
        "handle_shipping": "handle_shipping",
        "handle_order": "handle_order"
        # "handle_new_intent"が欠落している
    }
)

錯誤: ValueError: path must return a value in path_map

✅ 正しい実装：全戻り値をpath_mapで定義
def route_intent(state: CustomerServiceState) -> str:
    intent = state.get("intent", "general")
    
    routing = {
        "shipping_inquiry": "handle_shipping",
        "order_cancel": "handle_order_cancel",
        "order_change": "handle_order_change",
        "product_info": "handle_product_inquiry",
        "return_request": "handle_refund",
        "complaint": "escalate_to_human",
        "general": "handle_general"
    }
    
    return routing.get(intent, "handle_general")  # 必ずpath_map内の値を返す

workflow.add_conditional_edges(
    source="classify_intent",
    path=route_intent,
    path_map={
        "handle_shipping": "handle_shipping",
        "handle_order_cancel": "handle_order_cancel",
        "handle_order_change": "handle_order_change",
        "handle_product_inquiry": "handle_product_inquiry",
        "handle_refund": "handle_refund",
        "escalate_to_human": "escalate_to_human",
        "handle_general": "handle_general"
    }
)

原因：Router関数が返す値の中に、path_mapで定義されていないものが存在。
解決：Router関数の返す可能性のある全値をpath_mapに必ず含めてください。.get() 사용하여デフォルト値を返すことも有効です。

エラー4：型ヒントの不一致导致的グラフ構築エラー

# ❌ 誤った実装：router返回值类型不匹配
def route_intent(state: CustomerServiceState) -> str:  # 単一のstrを返す
    return "handle_shipping"

但しshould_escalateはboolを返す
def should_escalate(state: CustomerServiceState) -> bool:
    return True

✅ 正しい実装：pathに応じて返す型を統一
from typing import Literal

全ての場合を列挙型で定義
def route_intent(state: CustomerServiceState) -> Literal[
    "handle_shipping",
    "handle_order_cancel", 
    "handle_order_change",
    "handle_product_inquiry",
    "handle_refund",
    "escalate_to_human",
    "handle_general"
]:
    intent = state.get("intent", "general")
    
    routing = {
        "shipping_inquiry": "handle_shipping",
        "order_cancel": "handle_order_cancel",
        "order_change": "handle_order_change",
        "product_info": "handle_product_inquiry",
        "return_request": "handle_refund",
        "complaint": "escalate_to_human"
    }
    
    return routing.get(intent, "handle_general")

原因：Conditional Edgesでは全てのpath関数が同じ型の値を返すべきですが、混合させるとグラフ構築時にエラー。
解決：typing.Literalを使用して可能な戻り値を明示的に列挙し、型の安全性を確保してください。

性能最適化：バッチ处理と并发処理

私的实际项目中遇到的大规模并发处理需求では、以下の最適化を実施しました：

import asyncio
from concurrent.futures import ThreadPoolExecutor

class BatchCustomerService:
    """バッチ處理対応の客户服务クラス"""
    
    def __init__(self, api_key: str):
        self.holysheep = HolySheepClient(api_key)
        self.executor = ThreadPoolExecutor(max_workers=10)
    
    async def process_batch(self, messages: list[str]) -> list[dict]:
        """複数メッセージを并发処理"""
        
        async def process_single(msg: str) -> dict:
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(
                self.executor,
                self._sync_classify,
                msg
            )
        
        tasks = [process_single(msg) for msg in messages]
        results = await asyncio.gather(*tasks)
        
        return results
    
    def _sync_classify(self, message: str) -> dict:
        """同期的なIntent分類"""
        return self.holysheep.classify_intent(message)
    
    # 10件のメッセージを并发處理
    async def main():
        batch_processor = BatchCustomerService("YOUR_HOLYSHEEP_API_KEY")
        
        messages = [
            "注文内容を確認したい",
            "配送状況は？",
            "キャンセルしたい",
            # ... 10件のメッセージ
        ]
        
        results = await batch_processor.process_batch(messages)
        
        for msg, result in zip(messages, results):
            print(f"メッセージ: {msg}")
            print(f"Intent: {result['intent']}, 信頼度: {result['confidence']}")

まとめ

本記事では、LangGraphの条件路由（Conditional Edges）を活用したAIカスタマーサービスシステムの構築方法を解説しました。ポイントまとめ：

• 状態設計：Intent、Confidence、抽出Entityを状態に含める
• Router関数：typing.Literalで戻り値の型を明示的に定義
• HolySheep AI活用：¥1=$1の特恵レートでGPT-4.1やDeepSeek V3.2を利用可能
• エラー処理：APIキー、Base URL、path_mapの三重点を確認

条件路由をマスターすることで、単純なQAボットから複雑なビジネスロジックを持つ対話システムまで、柔軟な構築が可能になります。

👉 HolySheep AI に登録して無料クレジットを獲得