本記事では、Azure AI Agent Service を活用して、マルチエージェント構成を実装する手法をご紹介します。
はじめに
本記事について
本記事では、主に Azure SDK for Python の公式サンプルコードに含まれる AgentTeam クラスの紹介を行います。このクラスを用いることで、複数のエージェントが Azure AI Agent Service 上の同一スレッド内で連携し、自律的に会話しながらタスクを効率的に遂行する仕組みを構築できます。
以下の図は、本記事で実装するマルチエージェント構成の概要です。
Azure AI Agent Service とは
Azure AI Agent Service は、2024 年 11 月の Ignite 2024 で発表された、Azure 上で AI エージェントを開発するための統合基盤です。
このサービスは、従来の Assistants API を基盤に、Azure の多様なマネージドサービスと組み合わせることで、エンタープライズ向けのユースケースにも対応します。
2025年2月現在、Azure AI Foundry SDK に加え、Azure AI Foundry の Web UI を通じても利用可能となっており、パブリックプレビューとして提供されています。
Azure AI Agent Service による開発手法の詳細については、以下の記事をご参照ください。
Azure AI Agent Service の特徴
Azure AI Agent Service はフルマネージドサービスとして、AI エージェント開発を効率化するさまざまな機能を提供します。その主な特徴は次の通りです
-
会話履歴の自動管理
会話履歴を「スレッド（Threads）」として一元管理し、開発者が履歴管理の実装に手間をかける必要がありません。 -
自動ツール呼び出し
エージェントによるツールの呼び出しや応答処理をサーバーレス環境で自動的に実行します。 -
豊富なツール連携
Code Interpreter、Bing Search、Azure AI Search など、Microsoft Azure サービスとのシームレスな統合が可能です。
詳細については公式ドキュメントをご参照ください。
参考：https://learn.microsoft.com/ja-jp/azure/ai-services/agents/overview#why-use-azure-ai-agent-service
Azure AI Agent Service の課題
2025 年 2 月時点で、Azure AI Agent Service はシングルエージェントの構築をサポートしていますが、本記事で作成するようなマルチエージェント構成を実現する際にはいくつか課題が存在します。
-
エージェント間のタスク遷移
ユーザーのリクエストからタスクを生成し、最適なエージェントへタスクを割り当てるロジックが必要。 -
エージェント間の会話管理
複数のエージェント同士が同一スレッドを共有し、会話履歴を共有できる仕組みを自前で構築する必要がある。
解決アプローチ
こうした課題を解決する手段のひとつとして、AgentTeam クラスを使用します。
AgentTeam クラスでは、TeamLeader エージェントがタスクの生成と割り振りを担当します。
さらに、各エージェントは同一のスレッドを共有し、自律的に会話を進めながらタスクを効率的に処理していきます。
サンプルコード
以下は、Notebook 上のサンプルコードです。
免責事項
当記事で紹介するソースコードは、Azure SDK for Python の公式サンプルで提供されている AgentTeam クラスをベースに作成されています。
このソースコードは検証を目的として作成されたものであり、セキュリティやエラーハンドリングなど、実運用に必要な考慮事項を十分に網羅していない可能性があります。
そのため、本コードを利用する際は、必ずご自身の環境や要件に合わせて適切な改良や検証を行った上でご使用ください。
事前準備
1. ライブラリのインストール
ソースコード
Input[1]_InstallLibraries
!pip install azure-ai-projects==1.0.0b5 \
python-dotenv==1.0.1 \
azure-identity==1.19.0 --quiet
2. ユーザー関数定義
ユーザー定義関数には、以下の関数を 4 つ定義します。
- fetch_current_datetime 関数: 現在時刻を取得する関数
- fetch_weather 関数（ダミー関数）: 指定された場所の天気情報を取得
- send_email 関数（ダミー関数）: 指定の件名と本文を含むメールを宛先に送信
- convert_temperature 関数: 温度を摂氏から華氏に変換
ソースコード
Input[2]_UserFunctionDefinitions
import json
import datetime
from typing import Any, Callable, Set, Dict, List, Optional
def fetch_current_datetime(format: Optional[str] = None) -> str:
"""
現在の時刻を JSON 文字列として取得します。オプションでフォーマットを指定できます。
:param format (Optional[str]): 現在の時刻を返す形式。デフォルトは None で、標準形式を使用します。
:return: JSON 形式の現在の時刻。
:rtype: str
"""
current_time = datetime.datetime.now()
if format:
time_format = format
else:
time_format = "%Y-%m-%d %H:%M:%S"
time_json = json.dumps({"current_time": current_time.strftime(time_format)})
return time_json
def fetch_weather(location: str) -> str:
"""
指定された場所の天気情報を取得します。
:param location (str): 天気情報を取得する場所。
:return: JSON 文字列としての天気情報。
:rtype: str
"""
# ダミーの天気データを使用します。
dummy_weather_data = {"New York": "Sunny, 25°C", "London": "Cloudy, 18°C", "Tokyo": "Rainy, 22°C"}
weather = dummy_weather_data.get(location, "Weather data not available for this location.")
weather_json = json.dumps({"weather": weather})
return weather_json
def send_email(recipient: str, subject: str, body: str) -> str:
"""
指定の件名と本文を含むメールを宛先に送信します。
:param recipient (str): 宛先のメールアドレス。
:param subject (str): メールの件名。
:param body (str): メールの本文
:return: 確認メッセージ。
:rtype: str
"""
# ダミーのメール送信処理を実行します。
print(f"Sending email to {recipient}...")
print(f"Subject: {subject}")
print(f"Body:\n{body}")
message_json = json.dumps({"message": f"Email successfully sent to {recipient}."})
return message_json
def convert_temperature(celsius: float) -> str:
"""
温度を摂氏から華氏に変換します。
:param celsius (float): 摂氏での温度。
:rtype: float
:return: 華氏での温度。
:rtype: str
"""
fahrenheit = (celsius * 9 / 5) + 32
return json.dumps({"fahrenheit": fahrenheit})
3. 指示プロンプトテンプレート
TeamLeader および TeamMember エージェントの指示（システムメッセージ）で使うテンプレート群です。
-
TeamLeader エージェント:
タスクの生成と割り振りを行い、また、各タスク完了後に最終確認を行うエージェント。 -
TeamMember エージェント
ユーザーが作成するエージェント
ソースコード
Input[3]_InstructionTemplates
TEAM_LEADER_INSTRUCTIONS = """
あなたの名前は '{agent_name}' です。あなたはエージェントチームのリーダーです。
あなたのチーム名は '{team_name}' です。
あなたの役割は、ユーザーからのリクエストを受け取り、チームのメンバーを活用してタスクを完了させることです。
リクエストを受け取った際に行うべき唯一のことは、どのチームメンバーが次にどのタスクを担当すべきかを評価することです。
あなたは提供されている `create_task` 関数を使い、適切なエージェントにタスクを割り当てます。
その際、誰にタスクを割り当てたのか、そしてその理由を説明してください。
チーム内のすべてのメンバーのスキルを活用することが重要です。
元のユーザーリクエストが完全に処理されたと判断した場合は、新しいタスクを作成する必要はありません。
ここに、あなたのチームに所属する他のエージェントを示します：
"""
TEAM_LEADER_INITIAL_REQUEST = """
チーム内で次にこのリクエストを処理するのに最も適したエージェントにタスクを作成してください。
利用可能な `create_task` 関数を使用してタスクを作成してください。
リクエスト内容は以下の通りです：
f"{original_request}"
"""
TEAM_LEADER_TASK_COMPLETENESS_CHECK_INSTRUCTIONS = """
これまでの会話の流れ、特にスレッド内の最新のメッセージを確認してください。
最終的な結果を向上させる可能性のあるタスクが見つかった場合は、`create_task` 関数を使用してタスクを作成してください。
タスク作成についてユーザーに確認を求める必要はありません。
リクエストが完全に処理された場合は、新たなタスクを作成する必要はありません。
"""
TEAM_MEMBER_CAN_DELEGATE_INSTRUCTIONS = """
あなたの名前は '{name}' です。あなたはエージェントチームのメンバーです。
あなたのチーム名は '{team_name}' です。
{original_instructions}
- 必要に応じてタスクを委任することができます。委任する場合は、`create_task` 関数を使用し、自分の名前を 'requestor' に指定してください。
- 割り当てたタスクとその結果を簡潔に報告してください。
- 他のチームメンバーが専門知識を持っている場合は、助けを求めてください。
- 自分のタスクが完了したと判断したら、最終的な回答または実行したアクションを報告してください。
- 以下に、あなたのチームに所属する他のエージェントを示します：{team_description}
"""
TEAM_MEMBER_NO_DELEGATE_INSTRUCTIONS = """
あなたの名前は '{name}' です。あなたはエージェントチームのメンバーです。
あなたのチーム名は '{team_name}' です。
{original_instructions}
- あなたはタスクを委任せず、自分に割り当てられたタスクの遂行に集中してください。
- もし他のエージェントに適したタスクがあると感じた場合は、その旨をレスポンス内で言及してください。ただし、`create_task` を自ら呼び出してはいけません。
- 自分のタスクが完了したと判断したら、最終的な回答または実行したアクションを報告してください。
- 以下に、あなたのチームに所属する他のエージェントを示します：{team_description}
"""
4. AgentTeam Class
AgentTeam クラスを定義するソースコードです。
ソースコードの詳細は、処理の流れを追いながら解説していきます。
ソースコード
Input[4]_AgentTeamClassDefinition
# ------------------------------------
# Based on code from Microsoft Corporation.
# Copyright (c) Microsoft Corporation.
#
# Modified by Yuki Matayoshi, 2025.
#
# Licensed under the MIT License.
# ------------------------------------
from typing import Any, Dict, Optional, Set, List
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import FunctionTool, ToolSet, MessageRole, Agent
class _AgentTeamMember:
"""
Represents an individual agent on a team.
:param model: The model (e.g. GPT-4) used by this agent.
:param name: The agent's name.
:param instructions: The agent's initial instructions or "personality".
:param toolset: An optional ToolSet with specialized tools for this agent.
:param can_delegate: Whether this agent has delegation capability (e.g., 'create_task').
Defaults to True.
"""
def __init__(
self, model: str, name: str, instructions: str, toolset: Optional[ToolSet] = None, can_delegate: bool = True
) -> None:
self.model = model
self.name = name
self.instructions = instructions
self.agent_instance: Optional[Agent] = None
self.toolset: Optional[ToolSet] = toolset
self.can_delegate = can_delegate
class _AgentTask:
"""
Encapsulates a task for an agent to perform.
:param recipient: The name of the agent who should receive the task.
:param task_description: The description of the work to be done or question to be answered.
:param requestor: The name of the agent or user requesting the task.
"""
def __init__(self, recipient: str, task_description: str, requestor: str) -> None:
self.recipient = recipient
self.task_description = task_description
self.requestor = requestor
class AgentTeam:
"""
A class that represents a team of agents.
"""
# Static container to store all instances of AgentTeam
_teams: Dict[str, "AgentTeam"] = {}
_project_client: AIProjectClient
_thread_id: str = ""
_team_leader: Optional[_AgentTeamMember] = None
_members: List[_AgentTeamMember] = []
_tasks: List[_AgentTask] = []
_team_name: str = ""
def __init__(self, team_name: str, project_client: AIProjectClient):
"""
Initialize a new AgentTeam and set it as the singleton instance.
"""
# Validate that the team_name is a non-empty string
if not isinstance(team_name, str) or not team_name:
raise ValueError("Team name must be a non-empty string.")
# Check for existing team with the same name
if team_name in AgentTeam._teams:
raise ValueError(f"A team with the name '{team_name}' already exists.")
self.team_name = team_name
if project_client is None:
raise ValueError("No AIProjectClient provided.")
self._project_client = project_client
# Store the instance in the static container
AgentTeam._teams[team_name] = self
self.TEAM_LEADER_INSTRUCTIONS = TEAM_LEADER_INSTRUCTIONS
self.TEAM_LEADER_INITIAL_REQUEST = TEAM_LEADER_INITIAL_REQUEST
self.TEAM_LEADER_TASK_COMPLETENESS_CHECK_INSTRUCTIONS = TEAM_LEADER_TASK_COMPLETENESS_CHECK_INSTRUCTIONS
self.TEAM_MEMBER_CAN_DELEGATE_INSTRUCTIONS = TEAM_MEMBER_CAN_DELEGATE_INSTRUCTIONS
self.TEAM_MEMBER_NO_DELEGATE_INSTRUCTIONS = TEAM_MEMBER_NO_DELEGATE_INSTRUCTIONS
self.TEAM_LEADER_MODEL = "gpt-4o-mini"
@staticmethod
def get_team(team_name: str) -> "AgentTeam":
"""Static method to fetch the AgentTeam instance by name."""
team = AgentTeam._teams.get(team_name)
if team is None:
raise ValueError(f"No team found with the name '{team_name}'.")
return team
@staticmethod
def _remove_team(team_name: str) -> None:
"""Static method to remove an AgentTeam instance by name."""
if team_name not in AgentTeam._teams:
raise ValueError(f"No team found with the name '{team_name}'.")
del AgentTeam._teams[team_name]
def add_agent(
self, model: str, name: str, instructions: str, toolset: Optional[ToolSet] = None, can_delegate: bool = True
) -> None:
"""
Add a new agent (team member) to this AgentTeam.
:param model: The model name (e.g. GPT-4) for the agent.
:param name: The name of the agent being added.
:param instructions: The initial instructions/personality for the agent.
:param toolset: An optional ToolSet to configure specific tools (functions, etc.)
for this agent. If None, we'll create a default set.
:param can_delegate: If True, the agent can delegate tasks (via create_task).
If False, the agent does not get 'create_task' in its ToolSet
and won't mention delegation in instructions.
"""
if toolset is None:
toolset = ToolSet()
if can_delegate:
# If agent can delegate, ensure it has 'create_task'
try:
function_tool = toolset.get_tool(FunctionTool)
function_tool.add_functions(agent_team_default_functions)
except ValueError:
default_function_tool = FunctionTool(agent_team_default_functions)
toolset.add(default_function_tool)
member = _AgentTeamMember(
model=model,
name=name,
instructions=instructions,
toolset=toolset,
can_delegate=can_delegate,
)
self._members.append(member)
def _add_task(self, task: _AgentTask) -> None:
"""
Add a new task to the team's task list.
:param task: The task to be added.
"""
self._tasks.append(task)
def create_team_leader(self, model: str, name: str, instructions: str, toolset: Optional[ToolSet] = None) -> None:
"""
Add a new agent (team member) to this AgentTeam.
If team leader has not been created prior to the call to assemble_team,
then default team leader will be created automatically.
:param model: The model name (e.g. GPT-4) for the agent.
:param name: The name of the team leader agent being.
:param instructions: The initial instructions/personality for the agent.
:param toolset: An optional ToolSet to configure specific tools (functions, etc.) for the agent.
"""
assert self._project_client is not None, "project_client must not be None"
assert self._team_leader is None, "team leader has already been created"
# List all agents (will be empty at this moment if you haven't added any, or you can append after they're added)
for member in self._members:
instructions += f"- {member.name}: {member.instructions}\n"
self._team_leader = _AgentTeamMember(
model=model,
name=name,
instructions=instructions,
toolset=toolset,
can_delegate=True,
)
self._team_leader.agent_instance = self._project_client.agents.create_agent(
model=self._team_leader.model,
name=self._team_leader.name,
instructions=self._team_leader.instructions,
toolset=self._team_leader.toolset,
)
def _create_team_leader(self):
"""
Create and initialize the default 'TeamLeader' agent with awareness of all other agents.
"""
toolset = ToolSet()
toolset.add(default_function_tool)
instructions = self.TEAM_LEADER_INSTRUCTIONS.format(agent_name="TeamLeader", team_name=self.team_name) + "\n"
self.create_team_leader(
model=self.TEAM_LEADER_MODEL,
name="TeamLeader",
instructions=instructions,
toolset=toolset,
)
def assemble_team(self):
"""
Create the team leader agent and initialize all member agents with
their configured or default toolsets.
"""
assert self._project_client is not None, "project_client must not be None"
if self._team_leader is None:
self._create_team_leader()
for member in self._members:
if member is self._team_leader:
continue
team_description = ""
for other_member in self._members:
if other_member != member:
team_description += f"- {other_member.name}: {other_member.instructions}\n"
if member.can_delegate:
extended_instructions = self.TEAM_MEMBER_CAN_DELEGATE_INSTRUCTIONS.format(
name=member.name,
team_name=self._team_name,
original_instructions=member.instructions,
team_description=team_description,
)
else:
extended_instructions = self.TEAM_MEMBER_NO_DELEGATE_INSTRUCTIONS.format(
name=member.name,
team_name=self._team_name,
original_instructions=member.instructions,
team_description=team_description,
)
member.agent_instance = self._project_client.agents.create_agent(
model=member.model, name=member.name, instructions=extended_instructions, toolset=member.toolset
)
def dismantle_team(self) -> None:
"""
Delete all agents (including the team leader) from the project client.
"""
assert self._project_client is not None, "project_client must not be None"
if self._team_leader and self._team_leader.agent_instance:
print(f"Deleting team leader agent '{self._team_leader.name}'")
self._project_client.agents.delete_agent(self._team_leader.agent_instance.id)
for member in self._members:
if member is not self._team_leader and member.agent_instance:
print(f"Deleting agent '{member.name}'")
self._project_client.agents.delete_agent(member.agent_instance.id)
AgentTeam._remove_team(self.team_name)
def process_request(self, request: str, thread_id: str = None) -> None:
"""
Handle a user's request by creating a team and delegating tasks to
the team leader. The team leader may generate additional tasks.
:param request: The user's request or question.
"""
assert self._project_client is not None, "project client must not be None"
assert self._team_leader is not None, "team leader must not be None"
# --- Start of modified Code ---
# Create a new thread if no thread_id is provided
print("---")
if thread_id is None:
thread = self._project_client.agents.create_thread()
self._thread_id = thread.id
print(f"Created thread with ID: {self._thread_id}")
else:
self._thread_id = thread_id
print(f"Set thread with ID: {self._thread_id}")
# --- End of modified Code ---
team_leader_request = self.TEAM_LEADER_INITIAL_REQUEST.format(original_request=request)
self._add_task(_AgentTask(self._team_leader.name, team_leader_request, "InitialRequestToTeamLeader"))
while self._tasks:
task = self._tasks.pop(0)
print("---")
print(
f"Starting task for agent '{task.recipient}'. "
f"Requestor: '{task.requestor}'. "
f"Task description: '{task.task_description}'."
)
# --- Start of added Code ---
# Set the massage metadata
message_metadata = {"actual_role": task.requestor}
# Create the massage
message = self._project_client.agents.create_message(
thread_id=self._thread_id,
role="user",
content=task.task_description,
metadata=message_metadata # Add the massage metadata
)
# --- End of added Code ---
print(f"Created message with ID: {message.id} for task in thread {self._thread_id}")
agent = self._get_member_by_name(task.recipient)
if agent and agent.agent_instance:
run = self._project_client.agents.create_and_process_run(
thread_id=self._thread_id, assistant_id=agent.agent_instance.id
)
print(f"Created and processed run for agent '{agent.name}', run ID: {run.id}")
messages = self._project_client.agents.list_messages(
thread_id=self._thread_id
)
# --- Start of modified Code ---
# Find the message for the current run
for message in messages.data:
if message.run_id == run.id:
# Set the massage metadata
message_metadata = {"actual_role": task.recipient}
# Update the massage for the current run
self._project_client.agents.update_message(
thread_id=self._thread_id,
message_id=message.id,
metadata=message_metadata # Add the massage metadata
)
# --- End of modified Code ---
# If no tasks remain AND the recipient is not the TeamLeader,
# let the TeamLeader see if more delegation is needed.
if not self._tasks and not task.recipient == "TeamLeader":
team_leader_request = self.TEAM_LEADER_TASK_COMPLETENESS_CHECK_INSTRUCTIONS
task = _AgentTask(
recipient=self._team_leader.name, task_description=team_leader_request, requestor="RequestToTeamLeaderForTaskCompletenessCheck" # Change the requestor from original "user"
)
self._add_task(task)
def _get_member_by_name(self, name) -> Optional[_AgentTeamMember]:
"""
Retrieve a team member (agent) by name.
If no member with the specified name is found, returns None.
:param name: The agent's name within this team.
"""
if name == "TeamLeader":
return self._team_leader
for member in self._members:
if member.name == name:
return member
return None
def _create_task(team_name: str, recipient: str, request: str, requestor: str) -> str:
"""
Requests another agent in the team to complete a task.
:param team_name (str): The name of the team.
:param recipient (str): The name of the agent that is being requested to complete the task.
:param request (str): A description of the to complete. This can also be a question.
:param requestor (str): The name of the agent who is requesting the task.
:return: True if the task was successfully received, False otherwise.
:rtype: str
"""
task = _AgentTask(recipient=recipient, task_description=request, requestor=requestor)
team: Optional[AgentTeam] = None
try:
team = AgentTeam.get_team(team_name)
except:
pass
if team is not None:
team._add_task(task)
# --- Start of added Code ---
# Print the task details
print(f"Added task for agent '{recipient}'. Requestor: '{requestor}'. Task description: {request}.")
# --- End of added Code ---
return "True"
return "False"
# Any additional functions that might be used by the agents:
agent_team_default_functions: Set = {
_create_task,
}
default_function_tool = FunctionTool(functions=agent_team_default_functions)
動作確認
5. エージェントチームの初期化
Input[5]_InitializeAgentTeam
import os
from typing import Set
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import ToolSet, FunctionTool
from azure.identity import DefaultAzureCredential
# from agent_team import AgentTeam
os.environ["PROJECT_CONNECTION_STRING"] = "" # Enter the connection string if required
project_client = AIProjectClient.from_connection_string(
credential=DefaultAzureCredential(),
conn_str=os.environ["PROJECT_CONNECTION_STRING"]
)
# Define the user-defined functions
user_function_set_1: Set[Callable[..., Any]] = {
fetch_current_datetime,
fetch_weather
}
user_function_set_2: Set[Callable[..., Any]] = {
send_email
}
user_function_set_3: Set[Callable[..., Any]] = {
convert_temperature
}
functions = FunctionTool(functions=user_function_set_1)
toolset1 = ToolSet()
toolset1.add(functions)
# Create an AgentTeam
agent_team = AgentTeam("sample_agent", project_client=project_client)
# Add agents to the team
agent_team.add_agent(
model="gpt-4o-mini",
name="TimeWeatherAgent",
instructions="You are a specialized agent for time and weather queries.",
toolset=toolset1,
can_delegate=False,
)
functions = FunctionTool(functions=user_function_set_2)
toolset2 = ToolSet()
toolset2.add(functions)
agent_team.add_agent(
model="gpt-4o-mini",
name="SendEmailAgent",
instructions="You are a specialized agent for sending emails.",
toolset=toolset2,
can_delegate=False,
)
functions = FunctionTool(functions=user_function_set_3)
toolset3 = ToolSet()
toolset3.add(functions)
agent_team.add_agent(
model="gpt-4o-mini",
name="TemperatureAgent",
instructions="You are a specialized agent for temperature conversion.",
toolset=toolset3,
can_delegate=False,
)
# Initialize the team
agent_team.assemble_team()
add_agent メソッド
AgentTeam クラスを初期化したら、add_agent メソッドを用いてエージェントを追加できます。
add_agent メソッドで設定できるオプション一覧は以下の通りです。
- model: GPT-4 等のモデル名を指定
- name: エージェントの名前
- instructions: エージェントへの指示（original_instructions として代入されます）
- toolset: エージェントが利用できるツールを指定
- can_delegats: True の場合、他のエージェントへタスクを委任できる
特に can_delegate を True に設定すると、エージェントが必要に応じて他のエージェントにタスクを割り振ることができます。なお、TeamLeader エージェントはデフォルトで can_delegate が True に設定されています。
assemble_team メソッド
すべてのエージェントを追加した後は、assemble_team メソッドを使用してエージェントを初期化します。このメソッドを実行すると、Azure AI Agent Service の create_agent API がコールされ、各エージェントが Azure 上に作成されます。
また、この段階で TeamLeader エージェントも自動的に生成されます。
TeamLeader エージェント
TeamLeader エージェントの役割は、チームメンバー（TeamMember エージェント）にタスクを割り振り、全体のタスク進行を管理することです。このエージェントには、以下のような初期設定された指示（システムメッセージ）が与えられます。
TeamLeader エージェントの指示
あなたの名前は '{agent_name}' です。あなたはエージェントチームのリーダーです。
あなたのチーム名は '{team_name}' です。
あなたの役割は、ユーザーからのリクエストを受け取り、チームのメンバーを活用してタスクを完了させることです。
リクエストを受け取った際に行うべき唯一のことは、どのチームメンバーが次にどのタスクを担当すべきかを評価することです。
あなたは提供されている `create_task` 関数を使い、適切なエージェントにタスクを割り当てます。
その際、誰にタスクを割り当てたのか、そしてその理由を説明してください。
チーム内のすべてのメンバーのスキルを活用することが重要です。
元のユーザーリクエストが完全に処理されたと判断した場合は、新しいタスクを作成する必要はありません。
ここに、あなたのチームに所属する他のエージェントを示します：
- {member.name}: {member.instructions}
- {member.name}: {member.instructions}
- {member.name}: {member.instructions} ...
また、TeamLeader エージェントは create_task 関数を通じてタスクを生成できるツールを持ちます。このタスクの実体は、AgentTask クラスによって管理されます。
タスクの実態
AgentTeam で管理されるタスクは、AgentTask オブジェクトとして実体化されます。このオブジェクトはタスクの詳細を保持し、後述する process_request メソッドで順次処理されます。
AgentTask の主な属性は以下の通りです：
- requestor: タスクを要求したエージェント名（または User や System）
- recipient: タスクを受け取るエージェントの名前
- task_description: 実行する作業や回答する質問の説明
これにより、タスクのフローがシンプルかつ明確に管理されます。
6. リクエスト
Input[6]_SendProcessRequest
user_request = (
"こんにちは。現在の時刻を '2023-%m-%d %H:%M:%S' 形式で、また New York の天気を教えてください。"
"最後に、摂氏を華氏に変換し、結果の概要を記載したメールをサンプル受信者に送信してください。"
)
# Process the user request
agent_team.process_request(
request=user_request,
# thread_id=agent_team._thread_id # Uncomment this line to continue the conversation in the same thread
)
process_request メソッドの処理フロー
process_request メソッドでは、以下のフローでタスクを処理します。
最初に、TeamLeader エージェントに初期タスクが割り当てられます。この初期タスクの task_description は、リクエスト内容を基に以下のように構成されています。これはスレッド内のメッセージとして格納され、recipient である TeamLeader エージェントへのプロンプトとして機能します。
TEAM_LEADER_INITIAL_REQUEST
チーム内で次にこのリクエストを処理するのに最も適したエージェントにタスクを作成してください。
利用可能な `create_task` 関数を使用してタスクを作成してください。
リクエスト内容は以下の通りです：
f"{original_request}"
すべてのタスクが完了した後、TeamLeader エージェントは追加のタスクが必要かどうかを確認するための完了確認タスクを実行します。その際の task_description は以下のように設定されています。
TEAM_LEADER_TASK_COMPLETENESS_CHECK_INSTRUCTIONS
これまでの会話の流れ、特にスレッド内の最新のメッセージを確認してください。
最終的な結果を向上させる可能性のあるタスクが見つかった場合は、`create_task` 関数を使用してタスクを作成してください。
タスク作成についてユーザーに確認を求める必要はありません。
リクエストが完全に処理された場合は、新たなタスクを作成する必要はありません。
出力確認
以下は、process_request メソッドの出力例です（各 ID はマスキングされています）。この出力から、TeamLeader エージェントがタスクを割り振り、各 TeamMember エージェントが順次タスクを処理し、最後に TeamLeader エージェントが全体の進捗を確認する流れを確認できます。
---
Created thread with ID: thread_XXXXXXXXXXXXXXXXXXXXX
---
Starting task for agent 'TeamLeader'. Requestor: 'System'. Task description: '
チーム内で次にこのリクエストを処理するのに最も適したエージェントにタスクを作成してください。
利用可能な `create_task` 関数を使用してタスクを作成してください。
リクエスト内容は以下の通りです：
f"こんにちは。現在の時刻を '2023-%m-%d %H:%M:%S' 形式で、また New York の天気を教えてください。最後に、摂氏を華氏に変換し、結果の概要を記載したメールをサンプル受信者に送信してください。"
'.
Created message with ID: msg_XXXXXXXXXXXXXXXXXXXXX for task in thread thread_XXXXXXXXXXXXXXXXXXXXX
Added task for agent 'TimeWeatherAgent'. Requestor: 'TeamLeader'. Task description: 現在の時刻を '2023-%m-%d %H:%M:%S' 形式で教えてください。.
Added task for agent 'TimeWeatherAgent'. Requestor: 'TeamLeader'. Task description: New York の天気を教えてください。.
Added task for agent 'TemperatureAgent'. Requestor: 'TeamLeader'. Task description: 摂氏を華氏に変換してください。.
Added task for agent 'SendEmailAgent'. Requestor: 'TeamLeader'. Task description: 摂氏を華氏に変換した結果の概要をサンプル受信者に送信してください。.
Created and processed run for agent 'TeamLeader', run ID: run_XXXXXXXXXXXXXXXXXXXXX
---
Starting task for agent 'TimeWeatherAgent'. Requestor: 'TeamLeader'. Task description: '現在の時刻を '2023-%m-%d %H:%M:%S' 形式で教えてください。'.
Created message with ID: msg_XXXXXXXXXXXXXXXXXXXXX for task in thread thread_XXXXXXXXXXXXXXXXXXXXX
Created and processed run for agent 'TimeWeatherAgent', run ID: run_XXXXXXXXXXXXXXXXXXXXX
---
Starting task for agent 'TimeWeatherAgent'. Requestor: 'TeamLeader'. Task description: 'New York の天気を教えてください。'.
Created message with ID: msg_XXXXXXXXXXXXXXXXXXXXX for task in thread thread_XXXXXXXXXXXXXXXXXXXXX
Created and processed run for agent 'TimeWeatherAgent', run ID: run_XXXXXXXXXXXXXXXXXXXXX
---
Starting task for agent 'TemperatureAgent'. Requestor: 'TeamLeader'. Task description: '摂氏を華氏に変換してください。'.
Created message with ID: msg_XXXXXXXXXXXXXXXXXXXXX for task in thread thread_XXXXXXXXXXXXXXXXXXXXX
Created and processed run for agent 'TemperatureAgent', run ID: run_XXXXXXXXXXXXXXXXXXXXX
---
Starting task for agent 'SendEmailAgent'. Requestor: 'TeamLeader'. Task description: '摂氏を華氏に変換した結果の概要をサンプル受信者に送信してください。'.
Created message with ID: msg_XXXXXXXXXXXXXXXXXXXXX for task in thread thread_XXXXXXXXXXXXXXXXXXXXX
Sending email to sample@example.com...
Subject: 温度変換結果
Body:
現在のニューヨークの天気は晴れで、気温は25℃です。この温度は華氏に変換すると77.0°Fになります。
Created and processed run for agent 'SendEmailAgent', run ID: run_XXXXXXXXXXXXXXXXXXXXX
---
Starting task for agent 'TeamLeader'. Requestor: 'System'. Task description: '
これまでの会話の流れ、特にスレッド内の最新のメッセージを確認してください。
最終的な結果を向上させる可能性のあるタスクが見つかった場合は、`create_task` 関数を使用してタスクを作成してください。
タスク作成についてユーザーに確認を求める必要はありません。
リクエストが完全に処理された場合は、新たなタスクを作成する必要はありません。
'.
Created message with ID: msg_XXXXXXXXXXXXXXXXXXXXX for task in thread thread_XXXXXXXXXXXXXXXXXXXXX
Created and processed run for agent 'TeamLeader', run ID: run_XXXXXXXXXXXXXXXXXXXXX
7. 会話履歴の確認
仕様上の課題
これまでの会話履歴は、Azure AI Agent Service 上の単一スレッドで管理されています。以下のコードで会話履歴を確認できます。
Input[7]_FetchConversationHistoryOriginalRole
# Fetch and log all messages
messages = project_client.agents.list_messages(
thread_id=agent_team._thread_id,
)
for data_point in reversed(messages.data):
last_message_content = data_point.content[-1]
role = data_point.role.value
print(f"{role}: {last_message_content.text.value}")
print("---")
上記を実行すると、以下の出力が得られます。
user:
チーム内で次にこのリクエストを処理するのに最も適したエージェントにタスクを作成してください。
利用可能な `create_task` 関数を使用してタスクを作成してください。
リクエスト内容は以下の通りです：
f"こんにちは。現在の時刻を '2023-%m-%d %H:%M:%S' 形式で、また New York の天気を教えてください。最後に、摂氏を華氏に変換し、結果の概要を記載したメールをサンプル受信者に送信してください。"
---
assistant: 次のリクエストに対して、各エージェントにタスクを割り当てました。
1. **TimeWeatherAgent** に「現在の時刻を '2023-%m-%d %H:%M:%S' 形式で教えてください。」というタスクを作成しました。これは時間に関する専門家であるため、適任です。
2. **TimeWeatherAgent** に「New York の天気を教えてください。」というタスクも作成しました。同様に、天気に関する情報を扱うのに適切なエージェントです。
3. **TemperatureAgent** に「摂氏を華氏に変換してください。」というタスクを作成しました。温度変換の専門家であるため、最適です。
4. **SendEmailAgent** に「摂氏を華氏に変換した結果の概要をサンプル受信者に送信してください。」というタスクを作成しました。メール送信の専門家として、このタスクにふさわしいです。
これでリクエストは処理されましたので、新しいタスクは作成しません。
---
user: 現在の時刻を '2023-%m-%d %H:%M:%S' 形式で教えてください。
---
assistant: 現在の時刻は **2023-02-09 19:08:12** です。
---
user: New York の天気を教えてください。
---
assistant: New York の天気は **晴れ、25℃** です。
---
user: 摂氏を華氏に変換してください。
---
assistant: 25℃を華氏に変換すると **77.0°F** になります。
---
user: 摂氏を華氏に変換した結果の概要をサンプル受信者に送信してください。
---
assistant: 摂氏を華氏に変換した結果の概要を、サンプル受信者にメールで送信しました。メールは正常に送信されました。
---
user:
これまでの会話の流れ、特にスレッド内の最新のメッセージを確認してください。
最終的な結果を向上させる可能性のあるタスクが見つかった場合は、`create_task` 関数を使用してタスクを作成してください。
タスク作成についてユーザーに確認を求める必要はありません。
リクエストが完全に処理された場合は、新たなタスクを作成する必要はありません。
---
assistant: すでにリクエストはすべて処理されており、現在の時刻、New York の天気、摂氏から華氏への変換結果を提供し、サンプル受信者にメールを送信しました。これ以上のタスクは必要ありませんので、新たなタスクを作成する必要はありません。
---
この出力では、 各メッセージロールがタスクを実行したエージェント名になっておらず、どのエージェントがタスクを実行したのかを判別することができません。
原因は、Thread のメッセージオブジェクトの role 属性が、仕様上 user or assistant しかとることができないためです （2025 年 2 月時点）。
解決策：metadata を使用したロール名の管理
これを改善するため、Thread 内メッセージの metadata 属性に実際のロール名（actual_role）を記録する処理を追加しています。以下のコードで実行してみましょう。
Input[8]_FetchConversationHistoryMetadataRole
# Fetch and log all messages
messages = project_client.agents.list_messages(
thread_id=agent_team._thread_id,
)
for data_point in reversed(messages.data):
last_message_content = data_point.content[-1]
# Get the actual_role in the message metadata.
role = data_point.metadata["actual_role"]
print(f"{role}: {last_message_content.text.value}")
print("---")
上記の出力結果は以下になります。
System:
チーム内で次にこのリクエストを処理するのに最も適したエージェントにタスクを作成してください。
利用可能な `create_task` 関数を使用してタスクを作成してください。
リクエスト内容は以下の通りです：
f"こんにちは。現在の時刻を '2023-%m-%d %H:%M:%S' 形式で、また New York の天気を教えてください。最後に、摂氏を華氏に変換し、結果の概要を記載したメールをサンプル受信者に送信してください。"
---
TeamLeader: 次のリクエストに対して、各エージェントにタスクを割り当てました。
1. **TimeWeatherAgent** に「現在の時刻を '2023-%m-%d %H:%M:%S' 形式で教えてください。」というタスクを作成しました。これは時間に関する専門家であるため、適任です。
2. **TimeWeatherAgent** に「New York の天気を教えてください。」というタスクも作成しました。同様に、天気に関する情報を扱うのに適切なエージェントです。
3. **TemperatureAgent** に「摂氏を華氏に変換してください。」というタスクを作成しました。温度変換の専門家であるため、最適です。
4. **SendEmailAgent** に「摂氏を華氏に変換した結果の概要をサンプル受信者に送信してください。」というタスクを作成しました。メール送信の専門家として、このタスクにふさわしいです。
これでリクエストは処理されましたので、新しいタスクは作成しません。
---
TeamLeader: 現在の時刻を '2023-%m-%d %H:%M:%S' 形式で教えてください。
---
TimeWeatherAgent: 現在の時刻は **2023-02-09 19:08:12** です。
---
TeamLeader: New York の天気を教えてください。
---
TimeWeatherAgent: New York の天気は **晴れ、25℃** です。
---
TeamLeader: 摂氏を華氏に変換してください。
---
TemperatureAgent: 25℃を華氏に変換すると **77.0°F** になります。
---
TeamLeader: 摂氏を華氏に変換した結果の概要をサンプル受信者に送信してください。
---
SendEmailAgent: 摂氏を華氏に変換した結果の概要を、サンプル受信者にメールで送信しました。メールは正常に送信されました。
---
System:
これまでの会話の流れ、特にスレッド内の最新のメッセージを確認してください。
最終的な結果を向上させる可能性のあるタスクが見つかった場合は、`create_task` 関数を使用してタスクを作成してください。
タスク作成についてユーザーに確認を求める必要はありません。
リクエストが完全に処理された場合は、新たなタスクを作成する必要はありません。
---
TeamLeader: すでにリクエストはすべて処理されており、現在の時刻、New York の天気、摂氏から華氏への変換結果を提供し、サンプル受信者にメールを送信しました。これ以上のタスクは必要ありませんので、新たなタスクを作成する必要はありません。
---
これにより、どのエージェントがどのタスクを実行したのかを把握できるようになります。
今後、より良い実装方法がみつかればアップデートしていきます。
8. マルチエージェントチームを削除
最後に、以下のコードで作成したエージェントを削除します。
Input[8]_DeleteAgentTeam
# Dismantle the team
agent_team.dismantle_team()
動画で解説
AgentTeam が Azure AI Agent Service をどのように操作しているのか、今回のサンプルコードの流れを以下の動画で視覚的に解説しています。
備考
GitHub
当記事で使用した Notebook ファイルは以下になります。
その他のマルチエージェント戦略
当記事では、フレームワークを用いず、Azure AI Agent Service 上でマルチエージェント構成を実装してみました。しかし、より高度なマルチエージェント構成を構築したい場合は Microsoft が提供するオープンソースのエージェントフレームワークである AutoGen と Semantic Kernel を活用することで、より多彩なマルチエージェント構成が構築できます。
- AutoGen : 主に研究やプロトタイプ開発に適しており、柔軟かつ高度なカスタマイズ性を備えています。AI エージェント間の複雑な対話や新しいアルゴリズムの試験に活用されることが多いです。
- Semantic Kernel : 本番環境向けに設計されており、堅牢性とスケーラビリティを重視したフレームワークです。エンタープライズ環境での実運用を視野に入れたエージェント開発に最適です。
AutoGen の実装については、以下のブログに詳細が記載されていますので気になる方はご参照ください。