Multi-Agent Orchestratorを使用したAI旅行プランナーの構築

インテリジェントなAIアプリケーションの構築には、多くの場合、複数の専門AIエージェントの連携が必要です。この記事では、Multi-Agent Orchestratorフレームワークを使用して高度な旅行プランニングシステムを作成する方法を紹介し、複数のAIエージェントが協力して詳細な旅行プランを作成する方法を解説します。

AWSのエージェントフレームワーク「Multi-Agent AI framework」を使ってみた！
確かに普通に良さそう！これはもっと触ってみるか！ https://t.co/upT0eUPYdX pic.twitter.com/I0qBm6Ytav

— Maki@Sunwood AI Labs. (@hAru_mAki_ch) February 19, 2025

Multi-Agent Orchestratorの概要

Multi-Agent Orchestratorは、複雑な会話における複数のAIエージェントを管理・調整するために設計されたオープンソースフレームワークです。主な特徴は以下の通りです：

• 複数のAIエージェント間の対話を効率的に制御
• PythonとTypeScriptの両方で実装可能
• 会話の文脈を維持しながらリクエストを適切にルーティング
• Amazon Bedrock、Amazon Lex、Anthropic Claudeなど、様々なAIサービスをサポート
• カスタムエージェントの容易な統合
• AWS Lambdaからローカル環境まで、柔軟な展開オプション

旅行プランナーアプリケーションのアーキテクチャ

旅行プランナーアプリケーションは、スーパーバイザーによって調整される2つの専門エージェントを活用しています：

1. ResearcherAgent（調査担当）:

   • DuckDuckGo検索を使用してリアルタイムで情報収集
   • 観光スポット、アクティビティ、宿泊施設の分析
   • 検索結果から最も関連性の高い情報を抽出

2. PlannerAgent（プラン作成担当）:

   • ResearcherAgentの調査結果に基づいて旅程を作成
   • 期間に応じた効率的な観光プランを作成
   • マークダウン形式で見やすい旅程表を出力

実装の詳細

プロジェクト構成

travel-planner/
├── travel-planner-demo.py  # メインアプリケーション
├── search_web.py           # Web検索機能
└── requirements.txt        # 依存パッケージ

依存パッケージ

duckduckgo-search
dotenv
multi-agent-orchestrator[aws]
streamlit

Web検索ツールの実装

from duckduckgo_search import DDGS

def search_web(query: str, num_results: int = 2) -> str:
    """DuckDuckGoを使用してWeb検索を実行"""
    try:
        print(f"Searching DDG for: {query}")
        search = DDGS().text(query, max_results=num_results)
        return ('\n'.join(result.get('body','') for result in search))
    except Exception as e:
        print(f"Error searching for the query {query}: {e}")
        return f"Error searching for the query {query}: {e}"

エージェントの設定

ResearcherAgentの設定

# Web検索ツールの定義
search_web_tool = AgentTool(
    name='search_web',
    description='Web上で情報を検索',
    properties={
        'query': {
            'type': 'string',
            'description': '検索クエリ'
        }
    },
    func=search_web,
    required=['query']
)

# ResearcherAgentの初期化
researcher_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="ResearcherAgent",
    description="あなたは世界クラスの旅行リサーチャーです...",
    tool_config={
        'tool': AgentTools(tools=[search_web_tool]),
        'toolMaxRecursions': 20,
    },
    save_chat=False
))

PlannerAgentの設定

planner_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="PlannerAgent",
    description="あなたはシニアトラベルプランナーです..."
))

SupervisorAgentの設定

supervisor = SupervisorAgent(SupervisorAgentOptions(
    name="SupervisorAgent",
    lead_agent=planner_agent,
    team=[researcher_agent],
    trace=True
))

Orchestratorの設定

orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig(
    LOG_AGENT_CHAT=True,
    LOG_CLASSIFIER_CHAT=True,
    LOG_CLASSIFIER_RAW_OUTPUT=True,
    LOG_CLASSIFIER_OUTPUT=True,
    LOG_EXECUTION_TIMES=True,
    MAX_RETRIES=3,
    USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True,
    MAX_MESSAGE_PAIRS_PER_AGENT=10,
))

リクエストハンドラの実装

async def handle_request(_orchestrator: MultiAgentOrchestrator, 
                        _user_input: str, 
                        _user_id: str, 
                        _session_id: str):
    classifier_result = ClassifierResult(
        selected_agent=supervisor, 
        confidence=1.0
    )

    response: AgentResponse = await _orchestrator.agent_process_request(
        _user_input, 
        _user_id, 
        _session_id, 
        classifier_result
    )

    if isinstance(response, AgentResponse) and not response.streaming:
        if isinstance(response.output, str):
            return response.output
        elif isinstance(response.output, ConversationMessage):
            return response.output.content[0].get('text')

Streamlitインターフェースの実装

import streamlit as st

st.title("AI旅行プランナー ✈️")
st.caption("AI旅行プランナーで次の冒険を計画しましょう...")

destination = st.text_input("どこへ行きたいですか？")
num_days = st.number_input(
    "何日間の旅行を予定していますか？", 
    min_value=1, 
    max_value=30, 
    value=7
)

if st.button("旅程を生成"):
    with st.spinner("旅程を生成中..."):
        input_text = f"{destination}へ{num_days}日間"
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        response = loop.run_until_complete(
            handle_request(orchestrator, input_text, USER_ID, SESSION_ID)
        )
        st.write(response)

エージェント間の連携フロー

1. ユーザーが目的地と期間を入力
2. SupervisorAgentがリクエストを受信
3. ResearcherAgentが目的地の情報を収集：
   • 検索クエリの生成
   • Web検索の実行
   • 結果の分析と整理
4. PlannerAgentが収集した情報に基づいて旅程を作成
5. SupervisorAgentが最終的な旅行プランをユーザーに提供

まとめ

Multi-Agent Orchestratorフレームワークを使用することで、複数のAIエージェントを効率的に連携させ、高度なアプリケーションを作成することができます。この旅行プランナーの例では、情報収集と旅程作成という異なる専門性を持つエージェントを組み合わせて、包括的な旅行プランニングサービスを構築しました。

エージェントの役割を明確に分離し、SupervisorAgentで統括することで、保守性が高く、拡張性のある設計となっています。また、Streamlitを使用することで、シンプルながら使いやすいユーザーインターフェースを実現しています。

この実装は、カスタマーサポート、コンテンツ作成、データ分析など、調整されたAIエージェントを必要とする他のマルチエージェントシステムを構築する際のテンプレートとしても活用できます。

リポジトリ

GitHub - awslabs/multi-agent-orchestrator: Flexible and powerful framework for managing multiple AI agents and handling complex conversations

Flexible and powerful framework for managing multiple AI agents and handling complex conversations - awslabs/multi-agent-orchestrator

github.com