【Azure AI Foundry：第5回】分業のチームプレイ：Agent-to-Agentによるマルチエージェント協調
【新連載：Azure AI Foundryとマルチエージェント】
※本連載は、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可視化と自動精度評価」を実装し、マルチエージェント本編を完結させます。
知能の運用と品質保証の絶対防衛線を引く、最後の瞬間へ進みましょう。