【Azure AI Foundry：第5回】分業のチームプレイ：Agent-to-Agentによるマルチエージェント協調

【新連載：Azure AI Foundryとマルチエージェント】

前回（第4回）：【Azure AI Foundry：第4回】安全なコード無菌室：Managed Code Interpreterでのデータ分析

※本連載は、Azure AI Foundry SDK (azure-ai-projects) を用い、実務で安全・強力に動作する「エンタープライズ仕様の自律チームエージェント」を構築するステップバイステップのハンズオン連載です。

1. イントロダクション（今回の到達目標）

前回は、CodeInterpreterTool のサンドボックスを活用し、エージェント自身が安全な隔離領域でPythonコードを走らせてデータ集計やグラフ画像プロットを完結させる仕組みを構築しました。

これにより、自社ドキュメントを検索する「RAGの検索アーム」と、データを処理する「計算実行アーム」という2つの強力なパーツが揃いました。しかし、1つの巨大なエージェントに対して「ドキュメントの検索」と「データ集計・コード実行」という異なる役割を1つのプロンプトで同時に処理させようとすると、プロンプトの干渉や役割の混同が起き、指示の無視やハルシネーションが発生しやすくなります。

今回は、それぞれの専門役割に特化した2つの子エージェント（Document Agent ＆ Analyst Agent）を独立して作成します。そして、それらを司令塔であるコーディネーターエージェント（Coordinator Agent）にツールとして紐付ける「ConnectedAgentTool（エージェント間協調）」の仕組みを学び、複数のAIが対話を交わして複雑な業務を自動で連携・完遂する「マルチエージェントシステム」を構築することを目標とします。

前提条件（Prerequisites）

動作確認済みライブラリ: azure-ai-projects>=2.0.0
ローカル環境の認証: Azure CLI でのサインイン（az login）および環境変数 AZURE_AI_PROJECT_ENDPOINT の設定が完了していること。
Azureリソース要件: 接続するチャットモデル（例: gpt-4o）がデプロイされていること。

2. エージェントを「関数」として呼び出す ConnectedAgentTool の設計

複雑な業務を1つの巨大なプロンプト（ワンマン）で処理させずに、機能ごとにエージェントを分ける（分業チーム）アプローチには、以下の大きなメリットがあります。

システムプロンプトの最小化: 各エージェントは「ドキュメント検索のやり方」か「コード実行のやり方」のどちらか片方だけを完璧に記憶すればよいため、精度が劇的に向上します。
セキュリティの局所化: 危険なコード実行（Sandbox）が必要なのは Analyst Agent だけであるため、機密情報を扱う Document Agent は安全な領域で稼働し、必要な結果データだけを Analyst に受け渡す設計が可能です。

これを実現するために Azure AI Foundry SDK が提供するのが、エージェント自身を別のエージェントのツールとしてアタッチする「ConnectedAgentTool」です。

司令塔となる Coordinator Agent は、ユーザーの依頼を分析し、必要に応じて「Document Agentツール」を呼び出して知識を検索させ、得られた中間データを「Analyst Agentツール」に渡してコード実行させるという、役割の委譲（Delegation）と結果の統合を自律的に行います。

3. 【実践】ConnectedAgentTool を用いたマルチエージェント協調

「ドキュメント検索担当（RAG）」と「データ集計担当（Code Interpreter）」の2つを子エージェントとしてデプロイし、親エージェントがそれらをツールとして協調呼び出しして結果を得る、完全な Python スクリプトを実行してみましょう。

# 動作確認済みライブラリバージョン: azure-ai-projects==2.0.0
import os
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import (
    CodeInterpreterTool, 
    AzureAISearchTool, 
    AzureAISearchToolResource, 
    AISearchIndexResource,
    AzureAISearchQueryType,
    ConnectedAgentTool
)

def main():
    endpoint = os.environ.get("AZURE_AI_PROJECT_ENDPOINT")
    search_connection = os.environ.get("AI_SEARCH_CONNECTION_NAME")
    search_index = os.environ.get("AI_SEARCH_INDEX_NAME")

    if not all([endpoint, search_connection, search_index]):
        print("Error: 環境変数 AZURE_AI_PROJECT_ENDPOINT, AI_SEARCH_CONNECTION_NAME, AI_SEARCH_INDEX_NAME のいずれかが設定されていません。")
        return

    credential = DefaultAzureCredential()

    with AIProjectClient(endpoint=endpoint, credential=credential) as project_client:
        
        # --- (A) 検索担当の Document Agent の作成 ---
        connection = project_client.connections.get(search_connection)
        search_tool = AzureAISearchTool(
            azure_ai_search=AzureAISearchToolResource(
                indexes=[
                    AISearchIndexResource(
                        project_connection_id=connection.id,
                        index_name=search_index,
                        query_type=AzureAISearchQueryType.SIMPLE
                    )
                ]
            )
        )
        print("🤖 子エージェント1: Document Agent を作成中...")
        doc_agent = project_client.agents.create_agent(
            model="gpt-4o",
            name="document-agent",
            instructions="あなたはドキュメント検索の専門家です。指示された質問に回答するため、RAGインデックスから正確なデータを検索して返してください。",
            tools=search_tool.definitions,
            tool_resources=search_tool.resources
        )

        # --- (B) データ解析担当の Analyst Agent の作成 ---
        code_interpreter = CodeInterpreterTool()
        print("🤖 子エージェント2: Analyst Agent を作成中...")
        analyst_agent = project_client.agents.create_agent(
            model="gpt-4o",
            name="analyst-agent",
            instructions="あなたはデータ解析の専門家です。渡されたデータから棒グラフを作成し、画像ファイルとして保存してください。",
            tools=code_interpreter.definitions,
            tool_resources=code_interpreter.resources
        )

        # --- (C) 親の Coordinator Agent の作成 ＆ 子をツールとして登録 ---
        print("\n⚙️ 子エージェントを ConnectedAgentTool としてラップします...")
        
        # ConnectedAgentToolを介して、作成した子エージェントをツール化します
        doc_tool = ConnectedAgentTool(
            id=doc_agent.id,
            name="document_search_tool",
            description="社内データや過去の売上実績を検索し、正しいデータを取得するためのツールです。"
        )
        
        analyst_tool = ConnectedAgentTool(
            id=analyst_agent.id,
            name="data_analysis_tool",
            description="データをグラフ化したり、コードを実行して集計を行うためのツールです。"
        )

        print("🤖 親エージェント: Coordinator Agent を作成中...")
        coordinator_agent = project_client.agents.create_agent(
            model="gpt-4o",
            name="coordinator-agent",
            instructions="あなたはプロジェクトの司令塔です。ユーザーから『売上PDFの検索とグラフ化』を求められた場合、必ず document_search_tool でデータを取得してから、そのデータを data_analysis_tool に渡してグラフを作成させてください。",
            tools=[doc_tool, analyst_tool] # 2つのエージェントツールを装備
        )
        print(f"✅ すべてのエージェント構築が完了しました (Coordinator ID: {coordinator_agent.id})")

        # --- (D) 実行と会話の開始 ---
        thread = project_client.agents.create_thread()
        
        request = "社内データから今年度製品Aの売上推移を検索し、棒グラフとして作成してください。"
        print(f"\n➡️ ユーザー指示: {request}")
        
        project_client.agents.create_message(
            thread_id=thread.id,
            role="user",
            content=request
        )

        # 実行の監視（Coordinatorが状況を判断し、裏で子エージェントを順次呼び出します）
        print("⚡ システム: コーディネーターがマルチエージェントを協調実行中...")
        run = project_client.agents.create_and_process_run(
            thread_id=thread.id,
            assistant_id=coordinator_agent.id
        )
        print(f"✅ 実行ステータス: {run.status}")

        # 結果のテキスト回答の取得
        messages = project_client.agents.list_messages(thread_id=thread.id)
        latest_message = messages.data[0]
        print("\n=== Coordinator からの最終回答 ===")
        for part in latest_message.content:
            if hasattr(part, 'text'):
                print(part.text.value)
        print("===================================")

        # クリーンアップ
        project_client.agents.delete_agent(coordinator_agent.id)
        project_client.agents.delete_agent(doc_agent.id)
        project_client.agents.delete_agent(analyst_agent.id)
        print("\n🧹 リソースをクリーンアップしました。")

if __name__ == "__main__":
    main()

4. まとめと次回の展望

今回は、ConnectedAgentTool を用いることで、エージェントが別エージェントを「関数ツール」のように呼び出し、検索とコード実行という異なる役割の知能をチームとして協調（オーケストレーション）させる高度なシステムを構築しました。

これで、実務における複雑な業務を「分業」によって安全かつ最高精度で処理するチームエージェントが完成しました。

しかし、こうして複数のエージェントが裏で複雑に対話・処理を交わして動くようになると、今度は「最終的な結果が間違っていた場合、どのエージェントのどの推論ステップが原因だったのか」というデバッグの難易度が跳ね上がります。また、エージェントの回答が本当に品質（精度）を満たしているか、どうテストすればよいでしょうか？

次回、本編最終回。
マルチエージェントの複雑な対話テレメトリをポータル上で完全に見える化して追跡する「Tracing可視化」と、azure-ai-evaluation ライブラリを用いたエージェント精度の自動スコアリング評価、「Tracing可視化と自動精度評価」を実装し、マルチエージェント本編を完結させます。

知能の運用と品質保証の絶対防衛線を引く、最後の瞬間へ進みましょう。