[
  {
    "name": "NetShop Supervisor Agent",
    "description": "カスタマーサポートの全体を統括するエージェント。問い合わせを分析し、適切な専門エージェントにタスクを振り分けます。",
    "skills": [
      {
        "id": "classify_and_route",
        "name": "問い合わせ分類・ルーティング",
        "description": "問い合わせ内容を分析し、注文/配送/決済/一般の4カテゴリに分類して適切なエージェントに振り分ける"
      },
      {
        "id": "aggregate_responses",
        "name": "回答統合",
        "description": "各専門エージェントからの回答を統合して最終的な顧客向け回答を生成する"
      }
    ],
    "dependencies": ["order-agent", "shipping-agent", "payment-agent", "faq-agent"]
  },
  {
    "name": "NetShop Order Agent",
    "description": "注文の検索、詳細確認、キャンセルを担当するエージェント。注文DBにアクセスして情報を取得・更新します。",
    "skills": [
      {
        "id": "search_orders",
        "name": "注文検索",
        "description": "顧客ID/日付/ステータスで注文を検索",
        "inputModes": ["text"],
        "outputModes": ["text", "data"]
      },
      {
        "id": "get_order_details",
        "name": "注文詳細取得",
        "description": "注文番号で詳細情報を取得",
        "inputModes": ["text"],
        "outputModes": ["data"]
      },
      {
        "id": "cancel_order",
        "name": "注文キャンセル",
        "description": "pending/confirmedの注文をキャンセル",
        "inputModes": ["text"],
        "outputModes": ["text"]
      }
    ],
    "dependencies": []
  },
  {
    "name": "NetShop Shipping Agent",
    "description": "配送状況の追跡、再配達手配、配送日変更を担当するエージェント。配送業者APIと連携します。",
    "skills": [
      {
        "id": "track_shipment",
        "name": "配送追跡",
        "description": "追跡番号/注文番号で配送状況を追跡",
        "inputModes": ["text"],
        "outputModes": ["text", "data"]
      },
      {
        "id": "arrange_redelivery",
        "name": "再配達手配",
        "description": "不在持ち戻りの荷物の再配達を手配",
        "inputModes": ["text"],
        "outputModes": ["text"]
      },
      {
        "id": "change_delivery_date",
        "name": "配送日変更",
        "description": "配達予定日を変更",
        "inputModes": ["text"],
        "outputModes": ["text"]
      }
    ],
    "dependencies": ["order-agent"]
  },
  {
    "name": "NetShop Payment Agent",
    "description": "返金処理、決済状況確認、クーポン適用を担当するエージェント。決済ゲートウェイと連携します。",
    "skills": [
      {
        "id": "process_refund",
        "name": "返金処理",
        "description": "注文に対する返金を処理（承認が必要）",
        "inputModes": ["text"],
        "outputModes": ["text", "data"]
      },
      {
        "id": "check_payment_status",
        "name": "決済状況確認",
        "description": "注文の決済/返金ステータスを確認",
        "inputModes": ["text"],
        "outputModes": ["data"]
      },
      {
        "id": "apply_coupon",
        "name": "クーポン適用",
        "description": "お詫びクーポンを顧客に付与",
        "inputModes": ["text"],
        "outputModes": ["text"]
      }
    ],
    "dependencies": ["order-agent"]
  },
  {
    "name": "NetShop FAQ Agent",
    "description": "FAQ検索と一般的な質問への回答を担当するエージェント。ナレッジベースにアクセスします。",
    "skills": [
      {
        "id": "search_faq",
        "name": "FAQ検索",
        "description": "キーワードやカテゴリでFAQを検索",
        "inputModes": ["text"],
        "outputModes": ["text", "data"]
      },
      {
        "id": "answer_general_question",
        "name": "一般質問回答",
        "description": "返品ポリシー、配送日数など一般的な質問に回答",
        "inputModes": ["text"],
        "outputModes": ["text"]
      }
    ],
    "dependencies": []
  }
]

エージェント間の依存関係:

Supervisor → Order Agent → (独立)
           → Shipping Agent → Order Agent（注文情報が必要）
           → Payment Agent → Order Agent（注文情報が必要）
           → FAQ Agent → (独立)

from langgraph.graph import StateGraph, END

class MultiAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    intent: str | None
    required_agents: list[str]  # 必要なエージェントリスト
    agent_results: dict         # 各エージェントの結果
    current_agent_index: int
    final_response: str | None

def supervisor_classify(state: MultiAgentState) -> MultiAgentState:
    """Supervisor: 問い合わせを分類し必要なエージェントを決定"""
    response = llm.invoke([
        SystemMessage(content="""問い合わせを分析し、必要なエージェントを決定してください。
        利用可能: order_agent, shipping_agent, payment_agent, faq_agent
        依存関係: shipping_agent/payment_agent は order_agent の後に実行
        JSON形式で {"intent": "...", "agents": ["..."]} を返してください。"""),
        *state["messages"]
    ])
    result = json.loads(response.content)

    # 依存関係を考慮した実行順序に並べ替え
    agents = result["agents"]
    ordered_agents = []
    if "order_agent" in agents:
        ordered_agents.append("order_agent")
    for a in agents:
        if a != "order_agent":
            ordered_agents.append(a)

    return {
        "intent": result["intent"],
        "required_agents": ordered_agents,
        "current_agent_index": 0,
        "agent_results": {}
    }

def execute_next_agent(state: MultiAgentState) -> MultiAgentState:
    """次のエージェントを実行"""
    index = state["current_agent_index"]
    agents = state["required_agents"]
    agent_name = agents[index]

    # エージェントを実行（各エージェントの実装を呼び出し）
    result = agent_functions[agent_name](state)

    agent_results = {**state["agent_results"], agent_name: result}
    return {
        "agent_results": agent_results,
        "current_agent_index": index + 1
    }

def should_continue_agents(state: MultiAgentState) -> str:
    """全エージェントの実行が完了したか判定"""
    index = state["current_agent_index"]
    total = len(state["required_agents"])
    if index < total:
        return "execute_next_agent"
    return "supervisor_aggregate"

def supervisor_aggregate(state: MultiAgentState) -> MultiAgentState:
    """Supervisor: 全結果を統合して最終回答を生成"""
    results = state["agent_results"]
    response = llm.invoke([
        SystemMessage(content=f"""以下のエージェント結果を統合して、丁寧な顧客対応の回答を生成してください。
        結果: {json.dumps(results, ensure_ascii=False)}"""),
        *state["messages"]
    ])
    return {"final_response": response.content, "messages": [response]}

# グラフ構築
workflow = StateGraph(MultiAgentState)
workflow.add_node("supervisor_classify", supervisor_classify)
workflow.add_node("execute_next_agent", execute_next_agent)
workflow.add_node("supervisor_aggregate", supervisor_aggregate)

workflow.set_entry_point("supervisor_classify")
workflow.add_edge("supervisor_classify", "execute_next_agent")
workflow.add_conditional_edges("execute_next_agent", should_continue_agents)
workflow.add_edge("supervisor_aggregate", END)

app = workflow.compile()

Mermaid図:

graph TD
    start["問い合わせ受付"] --> classify["Supervisor<br/>分類・振り分け"]
    classify --> execute["エージェント実行"]
    execute -->|未完了| execute
    execute -->|全完了| aggregate["Supervisor<br/>結果統合"]
    aggregate --> respond["最終回答"]
    respond --> end_node["完了"]

シナリオ別対処方針:

部分的な結果での回答生成:

def generate_partial_response(state: MultiAgentState) -> str:
    results = state["agent_results"]
    errors = state.get("errors", [])

    available_info = []
    unavailable_info = []

    for agent in state["required_agents"]:
        if agent in results and results[agent].get("status") == "completed":
            available_info.append(f"{agent}: {results[agent]['data']}")
        else:
            unavailable_info.append(agent)

    prompt = f"""
    利用可能な情報で回答を生成してください。
    取得できなかった情報については正直に伝え、
    後ほど確認でき次第ご連絡する旨を添えてください。

    取得済み情報: {available_info}
    取得不可情報: {unavailable_info}
    """
    return llm.invoke(prompt).content

エスカレーション基準:

エスカレーション判定フロー:

1. 単一エージェント失敗
   → リトライ(3回) → フォールバック → 部分回答

2. 2つ以上のエージェント失敗
   → 部分回答が可能？
     ├── Yes → 部分回答 + 「詳細は後ほど」
     └── No → 人間オペレーターにエスカレーション

3. Supervisorエージェント自体の失敗
   → 即座に人間オペレーターにエスカレーション

4. 高リスク操作（返金、キャンセル）のエラー
   → 自動リトライせず、人間に判断を委譲

エスカレーション時の引き継ぎ情報:

• 元の問い合わせ内容
• 実行済みサブタスクとその結果
• 発生したエラーの詳細
• これまでの会話履歴
• 推奨される対応方針