階層型マルチAIエージェント開発完全ガイド：LangGraphとMCPで構築する自律型システム【実践コード付】

階層型マルチAIエージェント開発ガイド：LangGraphとMCP実装 AI開発（自作AI）
階層型マルチAIエージェント開発完全ガイド：LangGraphとMCPで構築する自律型システム【実践コード付】

はじめに：なぜ「階層型」マルチエージェントが必要なのか

こんにちは、AIコンサルタントのユイです。

「生成AIに複雑なタスクを依頼すると、途中で指示を忘れたり、迷走したりしてしまう」
皆様も、AI開発の現場でこのような壁にぶつかったことはないでしょうか？

単一のLLM（大規模言語モデル）に全ての役割を担わせるには、コンテキストの維持や専門性の発揮において限界があります。そこで現在、世界の開発トレンドは「マルチエージェントシステム」へと急速にシフトしています。

特に注目されているのが、階層型（Hierarchical）アーキテクチャです。これは、人間の組織図のように「上司（スーパーバイザー）」と「部下（ワーカー）」の役割を明確に分ける手法です。

本記事では、最新のオーケストレーションフレームワークであるLangGraphと、AIツールの標準規格MCP (Model Context Protocol)を組み合わせ、実用的な階層型マルチエージェントシステムを構築する方法を解説します。

自律型AIエージェントとは？2025年業務自動化の決定版と導入ガイド

階層型マルチエージェントシステムの仕組み

組織図のような役割分担

階層型システムでは、エージェントが対等に話し合うのではなく、明確な指揮命令系統を持ちます。

役割 機能・タスク 使用技術の例
スーパーバイザー
(Supervisor)		 ユーザーの指示を理解し、適切なワーカーにタスクを振り分け、結果を統合・承認する「司令塔」。 GPT-4o, Claude 3.5 Sonnet
(LangGraphでルーティング制御)
ワーカー
(Worker)		 「Web検索」「コード生成」「データ分析」など、特定のタスクに特化した専門家。 Specific Tools, MCP Servers

MCP (Model Context Protocol) の重要性

ここで重要なのが、2024年後半にAnthropic社が提唱し、急速に普及しているMCPです。これは、AIエージェントが外部データやツールと接続するための「USB規格」のようなものです。

従来、各エージェントに独自のツール接続コードを書いていましたが、MCP準拠のツール（MCPサーバー）を用意することで、どのエージェントからも標準化された方法でデータベースやAPIにアクセス可能になります。これにより、システムの拡張性が飛躍的に向上します。

ビジネスへの導入メリットとROI

技術的な面白さだけでなく、ビジネス視点での導入メリットも明確です。

ケーススタディ：カスタマーサポートの自動化

あるSaaS企業では、一次対応を階層型エージェントに置き換えることで、以下の成果を上げました。

  • スーパーバイザー： 問い合わせ内容を分類（技術的な問題 vs 請求の問題）。
  • 技術ワーカー： マニュアル検索やログ解析ツール（MCP接続）を実行。
  • 請求ワーカー： CRMデータ（MCP接続）を参照し、支払い状況を確認。

成果（ROI）：

  • 有人対応が必要なチケット数が65%削減。
  • 解決までの平均時間が4時間から5分に短縮。
  • APIコストは増加したが、人件費削減によりトータルコストは40%ダウン。
Microsoft、Anthropic・NVIDIAと巨額提携。AI覇権は「エコシステム」競争へ

【実践ガイド】LangGraph × Streamlit開発ハンズオン

ここからは、実際に動くシステムを構築していきます。今回は、「リサーチャー（検索担当）」と「ライター（執筆担当）」を束ねる編集長エージェントを作成します。

事前準備・環境設定

以下のライブラリをインストールしてください。

pip install langchain langchain-openai langgraph streamlit
# MCPツールを使用する場合は対応パッケージも必要ですが、今回は簡易化のため標準ツールを使用します

ステップ1：エージェントの状態（State）を定義

まず、エージェント間で受け渡す「会話の履歴」と「次に誰が動くか」の情報を定義します。

from typing import Annotated, List, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
import operator

# グラフ全体で共有する状態
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    next: str  # 次に実行するエージェント名

ステップ2：ワーカーエージェントの作成

特定の役割を持つエージェントを定義します。ここでは簡易的にプロンプトで役割を与えます。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI(model="gpt-4o")

def create_agent(system_prompt, tools):
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="messages"),
    ])
    # bind_toolsを使用することで、ツール実行能力を持たせる
    if tools:
        return prompt | llm.bind_tools(tools)
    return prompt | llm

# リサーチャー（実際にはTavilyなどの検索ツールを持たせると良い）
research_agent = create_agent(
    "あなたは優秀なリサーチャーです。与えられたトピックについて正確な情報を提供してください。",
    [] # ここにMCPツールや検索ツールリストを入れる
)

# ライター
writer_agent = create_agent(
    "あなたはプロのライターです。リサーチャーの情報を元に、ブログ記事を執筆してください。",
    []
)

# ノードとして実行する関数
def research_node(state):
    result = research_agent.invoke(state)
    return {"messages": [result]}

def writer_node(state):
    result = writer_agent.invoke(state)
    return {"messages": [result]}

ステップ3：スーパーバイザー（司令塔）の定義

これが階層型の肝です。状況を見て「誰に任せるか」または「完了するか」を判断させます。

from langchain_core.output_parsers.openai_functions import JsonOutputFunctionsParser

members = ["Researcher", "Writer"]
system_prompt = (
    "あなたは編集長です。以下のメンバーを管理しています：{members}。"
    "ユーザーの要望に応じて、次に誰に行動させるか決めてください。"
    "作業が完了したら 'FINISH' と答えてください。"
)

options = ["FINISH"] + members

function_def = {
    "name": "route",
    "description": "Select the next role.",
    "parameters": {
        "title": "routeSchema",
        "type": "object",
        "properties": {
            "next": {
                "title": "Next Role",
                "anyOf": [
                    {"enum": options}
                ],
            }
        },
        "required": ["next"],
    },
}

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder(variable_name="messages"),
    ("system", "Given the conversation above, who should act next? Select one of: {options}"),
]).partial(options=str(options), members=", ".join(members))

supervisor_chain = (
    prompt 
    | llm.bind_functions(functions=[function_def], function_call="route") 
    | JsonOutputFunctionsParser()
)

ステップ4：LangGraphの構築

定義したノードを繋ぎ合わせ、ワークフローを構築します。

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# ノードの追加
workflow.add_node("Supervisor", supervisor_chain)
workflow.add_node("Researcher", research_node)
workflow.add_node("Writer", writer_node)

# エントリーポイント
workflow.set_entry_point("Supervisor")

# 条件付きエッジ（スーパーバイザーの決定に従って分岐）
for member in members:
    workflow.add_edge(member, "Supervisor") # ワーカーは終わったら必ずスーパーバイザーに戻る

workflow.add_conditional_edges(
    "Supervisor",
    lambda x: x["next"],
    {
        "Researcher": "Researcher",
        "Writer": "Writer",
        "FINISH": END
    }
)

graph = workflow.compile()

ステップ5：Streamlit UIの実装

最後に、チャットUIを作成します。app.pyとして保存し、streamlit run app.pyで実行します。

import streamlit as st
from langchain_core.messages import HumanMessage

st.title("🤖 階層型AI編集部")

if "messages" not in st.session_state:
    st.session_state["messages"] = []

# チャット履歴の表示
for msg in st.session_state["messages"]:
    st.chat_message(msg.type).write(msg.content)

user_input = st.chat_input("記事のテーマを入力してください")

if user_input:
    st.chat_message("user").write(user_input)
    st.session_state["messages"].append(HumanMessage(content=user_input))
    
    # グラフの実行
    config = {"recursion_limit": 20} # 無限ループ防止
    initial_state = {"messages": st.session_state["messages"]}
    
    with st.spinner("AIチームが協働中..."):
        for output in graph.stream(initial_state, config=config):
            for key, value in output.items():
                if key == "Supervisor":
                    continue # スーパーバイザーの思考は隠す（必要なら表示）
                
                # ワーカーの出力を表示
                state_messages = value.get("messages", [])
                if state_messages:
                    last_msg = state_messages[-1]
                    st.chat_message("assistant").write(f"**{key}:** {last_msg.content}")
                    st.session_state["messages"].append(last_msg)

導入時のリスクと対策

素晴らしい技術ですが、以下のリスクには注意が必要です。

  • 無限ループ： スーパーバイザーとワーカーがお互いに仕事を投げ合う可能性があります。LangGraphのrecursion_limit設定は必須です。
  • コストとレイテンシ： 複数のエージェントが思考するため、単一の呼び出しよりトークン消費量が増え、応答速度も遅くなります。ROIに見合うか検討が必要です。
  • セキュリティ： エージェントが勝手に外部ツールを操作しないよう、MCP側で権限管理（Read-onlyなど）を徹底しましょう。
AIエージェントのセキュリティリスク｜自律型AIの悪用事例と対策

まとめ

LangGraphによる階層型マルチエージェントシステムは、複雑なビジネスプロセスを自律的に処理するための強力なソリューションです。最初は「スーパーバイザー＋1人のワーカー」から始め、徐々に専門家（ワーカー）を増やしていくアジャイルな開発をおすすめします。

ぜひ、上記のコードをベースに、あなただけの「AIチーム」を結成してみてください。

コメント

タイトルとURLをコピーしました