マルチエージェントシステム（Multi-Agent System）
この章ではマルチエージェントシステム（Multi-Agent System）を紹介します。
マルチエージェントシステムは、複数のエージェントの協調によって複雑なタスクを完了します。
これは大規模AIシステムを構築するための重要な技術であり、単一のエージェントでは実現できない複雑な機能を実現できます。
マルチエージェントアーキテクチャの基礎
マルチエージェントシステムの中核的な課題には3つの側面があります。
分担の問題：タスクを異なるエージェントに合理的に割り当てる方法。
通信の問題：エージェント間で情報をどのように伝達するか。
調整の問題：複数のエージェントの行動が一貫していて効率的であることを保証する方法。
基本的なアーキテクチャパターン
階層アーキテクチャ（Hierarchical）
階層アーキテクチャはツリー構造を採用し、1つのメインエージェントがスケジューリングを担当します。
メインエージェント（オーケストレーター）はタスクの分解と結果の統合を担当します。
サブエージェント（Subagent）は具体的なタスクの実行を担当します。
Orchestrator (调度者)
|
+-- Agent A (专家A)
+-- Agent B (专家B)
+-- Agent C (专家C)
ピアツーピアアーキテクチャ（Peer-to-Peer）
ピアツーピアアーキテクチャでは、すべてのエージェントが対等な立場で直接通信し協力します。
エージェント間で頻繁に対等な対話が必要なシーンに適しています。
Agent A <--> Agent B
| |
v v
Agent C <--> Agent D
役割分担と協調
効果的なマルチエージェントシステムには明確な役割定義が必要です。
各エージェントは自分の専門性と職務範囲を持つべきです。
協調メカニズムにより、複数のエージェントが調和して動作できます。
一般的な役割の定義
プランナー（Planner）：複雑なタスクを分解し、実行計画を策定します。
実行者（Executor）：ツールを呼び出し、具体的な操作を実行します。
レビュアー（Reviewer）：結果を評価し、改善提案を行います。
コーディネーター（Coordinator）：エージェント間の通信とタスク割り当てを管理します。
コード実装
マルチエージェントシステムの基本実装
class Agent:
"""
基本エージェントクラス
各エージェントは自身の役割、ツール、実行ロジックを持ちます
"""
def __init__(self, role, llm, tools):
# エージェントの役割名
self.role = role
# 関連する言語モデル
self.llm = llm
# 利用可能なツールのリスト
self.tools = tools
def execute(self, task, context=None):
"""
タスクを実行
:param task: タスクの説明
:param context: コンテキスト情報
:return: 実行結果
"""
# プロンプトを構築
prompt = self.build_prompt(task, context)
# 応答を生成
response = self.llm.generate(prompt)
# ツール呼び出しが必要な場合
if response.needs_tool_call:
return self.execute_tool(response)
return response.content
def build_prompt(self, task, context):
"""エージェントのプロンプトを構築"""
return f"役割：{self.role}\nタスク：{task}\nコンテキスト：{context}"
def execute_tool(self, response):
"""ツール呼び出しを実行"""
tool_name = response.tool_name
tool_input = response.tool_input
# ツールリストから対応するツールを見つける
tool = self.get_tool(tool_name)
# ツールを実行
return tool.execute(tool_input)
class MultiAgentSystem:
"""
マルチエージェント協調システム
複数のエージェント間の協調を管理する
"""
def __init__(self, agents, coordinator):
# エージェントの辞書、キーは役割名
self.agents = {a.role: a for a in agents}
# コーディネーターエージェント
self.coordinator = coordinator
def solve(self, task):
"""
複雑なタスクを解決する
複数のエージェントの協調を通じて完了する
"""
# ステップ1：コーディネーターがタスクを分析し、どのエージェントの参加が必要かを決定
plan = self.coordinator.decompose(task)
# ステップ2：タスクを該当するエージェントに配布
results = {}
for role, subtask in plan.subtasks.items():
# 対応する役割のエージェントを取得
agent = self.agents.get(role)
if agent:
# サブタスクを実行
results[role] = agent.execute(subtask)
# ステップ3：コーディネーターが結果を統合
final_result = self.coordinator.integrate(results)
return final_result
class CoordinatorAgent:
"""コーディネーターエージェント"""
def __init__(self, llm):
self.llm = llm
def decompose(self, task):
"""
タスクを分解
タスクを分析し、どのエージェントが必要か、どのように分担するかを決定
"""
prompt = f"""
次のタスクを分析し、サブタスクに分解して適切なエージェントに割り当ててください。
タスク：{task}
出力してください：
1. 必要なエージェント（役割）
2. 各エージェントが担当するサブタスク
3. サブタスク間の依存関係
出力形式：JSON
"""
response = self.llm.generate(prompt)
return Plan.parse(response)
def integrate(self, results):
"""
複数のエージェントの結果を統合
"""
prompt = f"""
以下のエージェントの実行結果を統合し、最終回答を生成してください。
結果リスト：
{results}
統合して最終結果を出力してください。
"""
return self.llm.generate(prompt)
"""
基本エージェントクラス
各エージェントは自身の役割、ツール、実行ロジックを持ちます
"""
def __init__(self, role, llm, tools):
# エージェントの役割名
self.role = role
# 関連する言語モデル
self.llm = llm
# 利用可能なツールのリスト
self.tools = tools
def execute(self, task, context=None):
"""
タスクを実行
:param task: タスクの説明
:param context: コンテキスト情報
:return: 実行結果
"""
# プロンプトを構築
prompt = self.build_prompt(task, context)
# 応答を生成
response = self.llm.generate(prompt)
# ツール呼び出しが必要な場合
if response.needs_tool_call:
return self.execute_tool(response)
return response.content
def build_prompt(self, task, context):
"""エージェントのプロンプトを構築"""
return f"役割：{self.role}\nタスク：{task}\nコンテキスト：{context}"
def execute_tool(self, response):
"""ツール呼び出しを実行"""
tool_name = response.tool_name
tool_input = response.tool_input
# ツールリストから対応するツールを見つける
tool = self.get_tool(tool_name)
# ツールを実行
return tool.execute(tool_input)
class MultiAgentSystem:
"""
マルチエージェント協調システム
複数のエージェント間の協調を管理する
"""
def __init__(self, agents, coordinator):
# エージェントの辞書、キーは役割名
self.agents = {a.role: a for a in agents}
# コーディネーターエージェント
self.coordinator = coordinator
def solve(self, task):
"""
複雑なタスクを解決する
複数のエージェントの協調を通じて完了する
"""
# ステップ1：コーディネーターがタスクを分析し、どのエージェントの参加が必要かを決定
plan = self.coordinator.decompose(task)
# ステップ2：タスクを該当するエージェントに配布
results = {}
for role, subtask in plan.subtasks.items():
# 対応する役割のエージェントを取得
agent = self.agents.get(role)
if agent:
# サブタスクを実行
results[role] = agent.execute(subtask)
# ステップ3：コーディネーターが結果を統合
final_result = self.coordinator.integrate(results)
return final_result
class CoordinatorAgent:
"""コーディネーターエージェント"""
def __init__(self, llm):
self.llm = llm
def decompose(self, task):
"""
タスクを分解
タスクを分析し、どのエージェントが必要か、どのように分担するかを決定
"""
prompt = f"""
次のタスクを分析し、サブタスクに分解して適切なエージェントに割り当ててください。
タスク：{task}
出力してください：
1. 必要なエージェント（役割）
2. 各エージェントが担当するサブタスク
3. サブタスク間の依存関係
出力形式：JSON
"""
response = self.llm.generate(prompt)
return Plan.parse(response)
def integrate(self, results):
"""
複数のエージェントの結果を統合
"""
prompt = f"""
以下のエージェントの実行結果を統合し、最終回答を生成してください。
結果リスト：
{results}
統合して最終結果を出力してください。
"""
return self.llm.generate(prompt)
AutoGen フレームワーク
AutoGen はマイクロソフトが開発したマルチエージェントプログラミングフレームワークです。
マルチエージェント対話システムを構築するためのシンプルなAPIを提供します。
AutoGen はマルチエージェントシステムの開発の複雑さを大幅に簡素化します。
中核概念
AssistantAgent：ツールを呼び出せるインテリジェントアシスタントエージェント。
UserProxyAgent：ユーザーの行動を代表し、コードとツール呼び出しを実行できます。
GroupChat：複数のエージェント間のグループチャット協調をサポート。
GroupChatManager：グループチャット内のエージェントのやり取りを管理。
コード例
AutoGen の基本的な使い方
# autogen フレームワークをインポート
import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# ==================== LLM の設定 ====================
# 設定辞書を作成
llm_config = {
"model": "gpt-4",
"api_key": "your-api-key", # 実際の使用時は環境変数から読み取る
"temperature": 0.7
}
# ==================== 単一エージェントの作成 ====================
# アシスタントエージェントを作成
assistant = AssistantAgent(
name="assistant",
system_message="""
あなたは役に立つPythonプログラミングアシスタントです。
ユーザーがコードを作成、デバッグ、最適化するのを支援できます。
""",
llm_config=llm_config
)
# ユーザープロキシエージェントを作成
# human_input_mode="NEVER" は人間の入力が不要であることを示す
# max_consecutive_auto_reply=10 は最大連続自動応答が10回であることを示す
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10
)
# ==================== 単一エージェントの対話 ====================
# 対話を開始
user_proxy.initiate_chat(
assistant,
message="クイックソートアルゴリズムを書いてください"
)
# ==================== マルチエージェントのグループチャット ====================
# 複数のエージェントを作成
coder = AssistantAgent(
name="coder",
system_message="あなたはPythonプログラミングの専門家で、コードの作成を担当します。",
llm_config=llm_config
)
reviewer = AssistantAgent(
name="reviewer",
system_message="あなたはコードレビューの専門家で、コードの品質を審査します。",
llm_config=llm_config
)
# グループチャットを作成
group_chat = GroupChat(
agents=[coder, reviewer],
messages=[],
max_round=10 # 最大対話10ラウンド
)
# グループチャットマネージャーを作成
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config
)
# グループチャットを開始
user_proxy.initiate_chat(
manager,
message="クイックソートアルゴリズムを作成し、コードをレビューしてください"
)
import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# ==================== LLM の設定 ====================
# 設定辞書を作成
llm_config = {
"model": "gpt-4",
"api_key": "your-api-key", # 実際の使用時は環境変数から読み取る
"temperature": 0.7
}
# ==================== 単一エージェントの作成 ====================
# アシスタントエージェントを作成
assistant = AssistantAgent(
name="assistant",
system_message="""
あなたは役に立つPythonプログラミングアシスタントです。
ユーザーがコードを作成、デバッグ、最適化するのを支援できます。
""",
llm_config=llm_config
)
# ユーザープロキシエージェントを作成
# human_input_mode="NEVER" は人間の入力が不要であることを示す
# max_consecutive_auto_reply=10 は最大連続自動応答が10回であることを示す
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10
)
# ==================== 単一エージェントの対話 ====================
# 対話を開始
user_proxy.initiate_chat(
assistant,
message="クイックソートアルゴリズムを書いてください"
)
# ==================== マルチエージェントのグループチャット ====================
# 複数のエージェントを作成
coder = AssistantAgent(
name="coder",
system_message="あなたはPythonプログラミングの専門家で、コードの作成を担当します。",
llm_config=llm_config
)
reviewer = AssistantAgent(
name="reviewer",
system_message="あなたはコードレビューの専門家で、コードの品質を審査します。",
llm_config=llm_config
)
# グループチャットを作成
group_chat = GroupChat(
agents=[coder, reviewer],
messages=[],
max_round=10 # 最大対話10ラウンド
)
# グループチャットマネージャーを作成
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config
)
# グループチャットを開始
user_proxy.initiate_chat(
manager,
message="クイックソートアルゴリズムを作成し、コードをレビューしてください"
)
より複雑な AutoGen の例
AutoGen のツール呼び出し
import autogen
from autogen import AssistantAgent, UserProxyAgent
# ツールリストを定義
# これらのツールはエージェントのツールリストに登録されます
tools = [
{
"name": "calculate",
"description": "数学計算を実行",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "計算する数学式。例：'2+3*5'"
}
},
"required": ["expression"]
}
},
{
"name": "search_web",
"description": "ウェブ情報を検索",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "検索キーワード"
}
},
"required": ["query"]
}
}
]
# ツール呼び出し機能を持つエージェントを作成
assistant = AssistantAgent(
name="assistant",
system_message="""
あなたは役に立つアシスタントです。
計算や情報検索が必要な場合は、関連ツールを呼び出すことができます。
""",
llm_config={
"model": "gpt-4",
"api_key": "your-api-key"
},
tools=tools # ツールの登録
)
# ユーザーエージェントの作成
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER"
)
# 会話を開始すると、Agentが自動的にツールを呼び出します
user_proxy.initiate_chat(
assistant,
message="(15 + 25) * 2 はいくらですか？"
)
# 応答例：
# Agentは次のように応答する場合があります：
# "計算してみます：(15 + 25) * 2 = 40 * 2 = 80"
# または、calculateツールを直接呼び出して結果を取得します
from autogen import AssistantAgent, UserProxyAgent
# ツールリストを定義
# これらのツールはエージェントのツールリストに登録されます
tools = [
{
"name": "calculate",
"description": "数学計算を実行",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "計算する数学式。例：'2+3*5'"
}
},
"required": ["expression"]
}
},
{
"name": "search_web",
"description": "ウェブ情報を検索",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "検索キーワード"
}
},
"required": ["query"]
}
}
]
# ツール呼び出し機能を持つエージェントを作成
assistant = AssistantAgent(
name="assistant",
system_message="""
あなたは役に立つアシスタントです。
計算や情報検索が必要な場合は、関連ツールを呼び出すことができます。
""",
llm_config={
"model": "gpt-4",
"api_key": "your-api-key"
},
tools=tools # ツールの登録
)
# ユーザーエージェントの作成
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER"
)
# 会話を開始すると、Agentが自動的にツールを呼び出します
user_proxy.initiate_chat(
assistant,
message="(15 + 25) * 2 はいくらですか？"
)
# 応答例：
# Agentは次のように応答する場合があります：
# "計算してみます：(15 + 25) * 2 = 40 * 2 = 80"
# または、calculateツールを直接呼び出して結果を取得します
A2A と MCP プロトコル
マルチAgentシステムには標準化された通信プロトコルが必要です。
A2A と MCP は2つの重要なプロトコル標準です。
A2A（Agent-to-Agent）プロトコル
A2Aプロトコルは、Agent間の通信の標準形式を定義します。
エージェント発見、タスク連携、状態同期をサポートします。
コア機能
エージェント発見（Agent Discovery）：Agentが他のAgentの能力とサービスを発見できます。
タスク連携（Task Collaboration）：複数のAgentが連携して複雑なタスクを完了できます。
状態同期（State Synchronization）：Agent間で状態情報を同期できます。
MCP（Model Context Protocol）プロトコル
MCPはオープンな標準プロトコルです。
AIモデルが外部ツールやデータソースと安全に接続できるようにします。
MCPはツール接続の標準プロトコルであり、USBインターフェースのような役割を果たします。
コアコンポーネント
Host：AIアプリケーションを実行するホスト環境。
Client：MCPサーバーとの接続を確立するクライアント。
Server：ツールとリソースを提供するサーバー側。
MCPツール定義の例
MCP ツール定義
{
"name": "filesystem",
"description": "ファイルシステム操作ツール",
"version": "1.0.0",
"tools": [
{
"name": "read_file",
"description": "ファイル内容の読み取り",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ファイルパス、必須"
},
"encoding": {
"type": "string",
"description": "ファイルエンコーディング、オプション、デフォルトはutf-8"
}
},
"required": ["path"]
}
},
{
"name": "write_file",
"description": "ファイル内容の書き込み",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ファイルパス、必須"
},
"content": {
"type": "string",
"description": "ファイル内容、必須"
}
},
"required": ["path", "content"]
}
},
{
"name": "list_directory",
"description": "ディレクトリ内容の一覧表示",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ディレクトリパス"
}
}
}
}
],
"resources": [
{
"uri": "file://config",
"description": "設定ファイル"
}
]
}
"name": "filesystem",
"description": "ファイルシステム操作ツール",
"version": "1.0.0",
"tools": [
{
"name": "read_file",
"description": "ファイル内容の読み取り",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ファイルパス、必須"
},
"encoding": {
"type": "string",
"description": "ファイルエンコーディング、オプション、デフォルトはutf-8"
}
},
"required": ["path"]
}
},
{
"name": "write_file",
"description": "ファイル内容の書き込み",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ファイルパス、必須"
},
"content": {
"type": "string",
"description": "ファイル内容、必須"
}
},
"required": ["path", "content"]
}
},
{
"name": "list_directory",
"description": "ディレクトリ内容の一覧表示",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ディレクトリパス"
}
}
}
}
],
"resources": [
{
"uri": "file://config",
"description": "設定ファイル"
}
]
}
注意：A2A と MCP が解決する問題は異なります。A2A は Agent 間の通信問題を解決し、MCP は Agent と外部ツールの接続問題を解決します。両者を組み合わせて使用することで、完全なマルチAgentシステムを構築できます。
主従エージェントパターン
マスタースレーブパターンは、古典的なマルチAgentアーキテクチャパターンです。
メインAgent（オーケストレーター）はタスクの分解と結果の統合を担当します。
サブAgent（Subagent）は具体的なタスクの実行を担当します。
適用シナリオ
タスクを独立したサブタスクに明確に分解できるシナリオ。
異なる専門スキルを持つAgentの連携が必要なシナリオ。
中央での調整と制御が必要な複雑なワークフロー。
コード実装
主従エージェントパターンの実装
class OrchestratorAgent:
"""
メインオーケストレーションAgent（メインAgent）
タスクの分解、割り当て、結果の統合を担当
"""
def __init__(self, llm, subagents):
# 言語モデル
self.llm = llm
# サブAgent辞書
self.subagents = {a.role: a for a in subagents}
def handle_request(self, request):
"""
ユーザーリクエストの処理
完全なオーケストレーションフロー
"""
# ==================== 分析フェーズ ====================
# リクエストを分析し、複数のAgentの連携が必要かどうかを判断
plan = self.llm.plan(request)
if len(plan.subtasks) == 1:
# 単一タスクの場合は、対応するAgentに直接配布
target_role = plan.target_role
return self.subagents[target_role].execute(plan.subtasks[0])
# ==================== マルチAgent連携フェーズ ====================
# タスクキューの作成
task_queue = self.create_task_queue(plan.subtasks)
# 完了した結果の保存
results = []
# すべてのタスクが完了するまで処理ループ
while task_queue:
# 並列実行可能なタスクの取得
parallel_tasks = self.get_parallel_tasks(task_queue)
# 並列実行
parallel_results = self.execute_in_parallel(parallel_tasks)
results.extend(parallel_results)
# タスクキューの更新
# 依存関係の処理、完了したタスクの削除、新しいタスクの追加
task_queue = self.update_queue(task_queue, parallel_results)
# ==================== 統合フェーズ ====================
# すべての結果の統合
return self.integrate_results(results)
def create_task_queue(self, subtasks):
"""タスクキューの作成"""
return TaskQueue(subtasks)
def get_parallel_tasks(self, task_queue):
"""
並列実行可能なタスクの取得
依存関係のあるタスクを除外
"""
ready_tasks = []
for task in task_queue.tasks:
# タスクの依存関係がすべて満たされているか確認
if all(dep in task_queue.completed for dep in task.dependencies):
ready_tasks.append(task)
return ready_tasks
def execute_in_parallel(self, tasks):
"""複数タスクの並列実行"""
import concurrent.futures
results = []
with concurrent.futures.ThreadPoolExecutor() as executor:
# すべてのタスクを送信
future_to_task = {
executor.submit(self.execute_task, task): task
for task in tasks
}
# 結果の収集
for future in concurrent.futures.as_completed(future_to_task):
task = future_to_task[future]
try:
result = future.result()
results.append(result)
except Exception as e:
results.append(TaskResult(task=task, error=str(e)))
return results
def execute_task(self, task):
"""単一タスクの実行"""
agent = self.subagents.get(task.role)
if not agent:
return TaskResult(task=task, error=f"Unknown role: {task.role}")
return agent.execute(task.description)
def update_queue(self, task_queue, completed_results):
"""タスクキューの更新"""
完了したタスクを完了としてマーク
for result in completed_results:
task_queue.completed.add(result.task.id)
# 完了したタスクを削除
task_queue.tasks = [
t for t in task_queue.tasks
if t.id not in task_queue.completed
]
return task_queue
def integrate_results(self, results):
"""
複数のAgentの結果の統合
最終回答の生成
"""
prompt = f"""
以下のAgentの実行結果を統合し、最終回答を生成してください。
結果リスト：
{results}
完全で一貫性のある回答に統合してください。
"""
return self.llm.generate(prompt)
class Subagent:
"""サブAgent基底クラス"""
def __init__(self, role, llm, tools=None):
self.role = role
self.llm = llm
self.tools = tools or []
def execute(self, task):
"""タスク実行の具体的なロジック"""
raise NotImplementedError
"""
メインオーケストレーションAgent（メインAgent）
タスクの分解、割り当て、結果の統合を担当
"""
def __init__(self, llm, subagents):
# 言語モデル
self.llm = llm
# サブAgent辞書
self.subagents = {a.role: a for a in subagents}
def handle_request(self, request):
"""
ユーザーリクエストの処理
完全なオーケストレーションフロー
"""
# ==================== 分析フェーズ ====================
# リクエストを分析し、複数のAgentの連携が必要かどうかを判断
plan = self.llm.plan(request)
if len(plan.subtasks) == 1:
# 単一タスクの場合は、対応するAgentに直接配布
target_role = plan.target_role
return self.subagents[target_role].execute(plan.subtasks[0])
# ==================== マルチAgent連携フェーズ ====================
# タスクキューの作成
task_queue = self.create_task_queue(plan.subtasks)
# 完了した結果の保存
results = []
# すべてのタスクが完了するまで処理ループ
while task_queue:
# 並列実行可能なタスクの取得
parallel_tasks = self.get_parallel_tasks(task_queue)
# 並列実行
parallel_results = self.execute_in_parallel(parallel_tasks)
results.extend(parallel_results)
# タスクキューの更新
# 依存関係の処理、完了したタスクの削除、新しいタスクの追加
task_queue = self.update_queue(task_queue, parallel_results)
# ==================== 統合フェーズ ====================
# すべての結果の統合
return self.integrate_results(results)
def create_task_queue(self, subtasks):
"""タスクキューの作成"""
return TaskQueue(subtasks)
def get_parallel_tasks(self, task_queue):
"""
並列実行可能なタスクの取得
依存関係のあるタスクを除外
"""
ready_tasks = []
for task in task_queue.tasks:
# タスクの依存関係がすべて満たされているか確認
if all(dep in task_queue.completed for dep in task.dependencies):
ready_tasks.append(task)
return ready_tasks
def execute_in_parallel(self, tasks):
"""複数タスクの並列実行"""
import concurrent.futures
results = []
with concurrent.futures.ThreadPoolExecutor() as executor:
# すべてのタスクを送信
future_to_task = {
executor.submit(self.execute_task, task): task
for task in tasks
}
# 結果の収集
for future in concurrent.futures.as_completed(future_to_task):
task = future_to_task[future]
try:
result = future.result()
results.append(result)
except Exception as e:
results.append(TaskResult(task=task, error=str(e)))
return results
def execute_task(self, task):
"""単一タスクの実行"""
agent = self.subagents.get(task.role)
if not agent:
return TaskResult(task=task, error=f"Unknown role: {task.role}")
return agent.execute(task.description)
def update_queue(self, task_queue, completed_results):
"""タスクキューの更新"""
完了したタスクを完了としてマーク
for result in completed_results:
task_queue.completed.add(result.task.id)
# 完了したタスクを削除
task_queue.tasks = [
t for t in task_queue.tasks
if t.id not in task_queue.completed
]
return task_queue
def integrate_results(self, results):
"""
複数のAgentの結果の統合
最終回答の生成
"""
prompt = f"""
以下のAgentの実行結果を統合し、最終回答を生成してください。
結果リスト：
{results}
完全で一貫性のある回答に統合してください。
"""
return self.llm.generate(prompt)
class Subagent:
"""サブAgent基底クラス"""
def __init__(self, role, llm, tools=None):
self.role = role
self.llm = llm
self.tools = tools or []
def execute(self, task):
"""タスク実行の具体的なロジック"""
raise NotImplementedError
主従パターンの長所と短所
| 長所 | 短所 |
|---|---|
| 構造が明確で、理解と実装が容易 | メインAgentが単一障害点になる |
| 中央調整により制御が容易 | メインAgentがパフォーマンスのボトルネックになる可能性がある |
| 階層構造が明確なタスクに適している | 柔軟性が低く、動的な調整が難しい |
| デバッグが容易で、各サブタスクを追跡できる | 対等な相互作用が頻繁なシナリオには適さない |
章のまとめ
この章では、マルチエージェントシステムのコア概念と実装方法を紹介しました。
マルチAgentアーキテクチャの基礎階層アーキテクチャとフラットアーキテクチャの2つの基本パターンを紹介しました。
役割分担と連携明確な役割定義によりシステム効率を向上させます。
AutoGenフレームワークマルチAgent開発を簡素化するプログラミングフレームワークを提供します。
A2A と MCP プロトコルAgent通信とツール接続の標準プロトコルです。
マスタースレーブAgentパターン明確な階層構造を持つ複雑なタスクに適しています。
適切なマルチAgentアーキテクチャの選択は、具体的なシナリオに基づく必要があります。
単純なシナリオでは、単一のAgentで要件を満たせます。
複雑なシナリオでは、タスクの特性に応じて適切なアーキテクチャパターンを選択する必要があります。
その他の拡張