マルチエージェント協調：スキルチェーンでAIチームを構築

単独からチーム協力へ

単一のAIエージェントの能力には限界があります。計画は得意でも実行は苦手、コーディングは得意でもテストは苦手、といった具合です。これは、一人の人間が同時にトップレベルのアーキテクト、プログラマー、テストエンジニアにはなれないのと同じです。マルチエージェント協調（Multi-Agent Collaboration）は、まさにこの問題を解決するために生まれました。複雑なタスクを複数のサブタスクに分解し、それぞれの専門性を持つエージェントが担当し、協力して全体の目標を達成します。

マルチエージェント協調の核となる考え方は、人間のチームワークのパターンを参考にしています。計画と分解を担当する者（プロジェクトマネージャー）、具体的な実行を担当する者（エンジニア）、品質を担当する者（レビュアー）がいます。各エージェントは自分の領域に集中し、標準化された通信プロトコルで情報を交換し、最終的に高品質な結果を提供します。

マルチエージェント協調の3つのモード

1. シーケンシャルパイプライン（Sequential Pipeline）：エージェントが固定順序で実行し、各エージェントの出力が次のエージェントの入力になります。文書処理パイプラインなど、プロセスが明確なタスクに適しています。

2. 階層型（Hierarchical）：「マネージャー」エージェントがタスクの割り当てと結果の集約を担当し、複数の「ワーカー」エージェントが並行して作業します。並列分解可能なタスクに適しています。

3. ディベート（Debate）：複数のエージェントが独立して案を生成し、互いにレビューや議論を行い、最終的に合意に達します。多角的な分析が必要な意思決定タスクに適しています。

この記事では、階層型モードに焦点を当てます。これは「マルチエージェント協調チェーン」スキルチェーンが採用しているモードでもあります。

スキルチェーンの分解：マルチエージェント協調チェーン

「マルチエージェント協調チェーン」を例に、階層型モードのノード編成を示します。

ノード1：タスク計画（sp-188）——「プランナー」エージェントはユーザーの要件を受け取り、複雑なタスクを実行可能なサブタスクに分解し、適切な実行者に割り当てます。このノードの品質が、協調チェーン全体の効率を直接決定します。

ノード2：コード実行（sp-187）——「実行者」エージェントは、プランナーから割り当てられたタスクに基づいて、具体的なコーディング作業を完了します。複数の実行者が並行して作業し、それぞれ異なるモジュールを担当できます。

ノード3：品質レビュー（sp-195）——「レビュアー」エージェントは、実行者の成果物の品質をチェックします。レビュアーは実行者から独立し、客観的な基準でコード品質、論理的正しさ、セキュリティを評価する必要があります。

ノード4：結果集約（sp-122）——プランナーが再び介入し、全実行者の成果物とレビュアーのフィードバックを収集し、最終成果物に統合します。レビューが不合格の場合、プランナーはタスクを実行ノードに再割り当てします。

実践：マルチエージェント協調フレームワーク

以下のコードは、階層型マルチエージェント協調の実装を示しています。

import json
from openai import OpenAI
from concurrent.futures import ThreadPoolExecutor

client = OpenAI(
    api_key="your-deepseek-api-key",
    base_url="https://api.deepseek.com"
)

class AgentRole:
    """エージェントの役割を定義"""
    def __init__(self, name, system_prompt):
        self.name = name
        self.system_prompt = system_prompt

    def execute(self, task):
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": task}
            ]
        )
        return response.choices[0].message.content

# 役割の定義
PLANNER = AgentRole("プランナー", """あなたはシニア技術アーキテクトです。
あなたの責務：
1. ユーザーの要件を分析し、2〜4つの並列実行可能なサブタスクに分解する
2. 各サブタスクの実行者タイプ（フロントエンド/バックエンド/テスト/ドキュメント）を指定する
3. サブタスク間の依存関係を定義する

出力形式（JSON）：
{
  "subtasks": [
    {"id": 1, "type": "タイプ", "description": "タスク説明", "depends_on": []}
  ]
}""")

EXECUTOR = AgentRole("実行者", """あなたはフルスタック開発エンジニアです。
タスクの説明に基づいて、高品質なコード実装を生成してください。
含めるもの：
1. 完全なコード実装
2. 主要な設計決定の説明
3. 潜在的なリスクポイント""")

REVIEWER = AgentRole("レビュアー", """あなたは厳格なコードレビュー専門家です。
レビュー基準：
1. コードの正確性（論理エラー）
2. セキュリティ（インジェクション、漏洩などのリスク）
3. パフォーマンス（明らかなボトルネック）
4. 保守性（コードが明確で理解しやすいか）

出力形式（JSON）：
{
  "approved": true/false,
  "score": 0-100,
  "issues": ["問題の説明"],
  "suggestions": ["改善提案"]
}""")

class MultiAgentSystem:
    def __init__(self, max_iterations=5):
        self.max_iterations = max_iterations

    def run(self, user_request):
        print(f"ユーザー要件：{user_request}\n")

        # フェーズ1：計画
        print("=== フェーズ1：タスク計画 ===")
        plan = PLANNER.execute(f"以下の要件に対する実行計画を作成してください：\n{user_request}")
        subtasks = json.loads(plan)["subtasks"]
        print(f"{len(subtasks)}個のサブタスクに分解\n")

        # フェーズ2：並列実行
        print("=== フェーズ2：並列実行 ===")
        results = {}
        with ThreadPoolExecutor(max_workers=4) as executor:
            futures = {}
            for task in subtasks:
                future = executor.submit(
                    EXECUTOR.execute,
                    f"サブタスク {task['id
']}：{task['description']}"
                )
                futures[future] = task['id']

            for future in futures:
                task_id = futures[future]
                results[task_id] = future.result()
                print(f"サブタスク {task_id} の実行が完了しました")

        # フェーズ 3：レビュー
        print("\n=== フェーズ 3：品質レビュー ===")
        all_approved = True
        for task_id, result in results.items():
            review = REVIEWER.execute(f"次のコードをレビューしてください：\n{result}")
            review_data = json.loads(review)
            status = "✓ 合格" if review_data["approved"] else "✗ 修正が必要"
            print(f"サブタスク {task_id}：{status}（スコア：{review_data['score']}）")
            if not review_data["approved"]:
                all_approved = False

        # フェーズ 4：集約
        print("\n=== フェーズ 4：結果の集約 ===")
        summary_prompt = f"""マルチエージェント協調の結果を以下のように要約してください：

元の要件：{user_request}
サブタスク実行結果：
{json.dumps(results, ensure_ascii=False, indent=2)}

最終納品レポートを生成してください。以下を含めてください：
1. プロジェクト概要
2. 各モジュールの説明
3. 統合計画
4. 注意事項"""

        final_report = PLANNER.execute(summary_prompt)
        return final_report

# 使用例
system = MultiAgentSystem(max_iterations=5)
report = system.run("CRUDとステータス管理をサポートする簡単なTodo Webアプリを開発する")
print(f"\n最終レポート：\n{report}")

re>
マルチエージェント協調の課題と対策
通信オーバーヘッド：エージェント間の情報伝達は大量のトークンを消費します。解決策：構造化された通信プロトコル（例：JSON Schema）を使用して冗長な情報を削減し、中間結果を要約・圧縮します。
一貫性の問題：異なるエージェントが矛盾する出力を生成する可能性があります。解決策：各エージェントの入出力形式を定義する明確なインターフェース契約を設定し、競合を解決する「調停者」ロールを導入します。
エラー伝播：上流のエージェントのエラーが下流に伝播する可能性があります。解決策：各ノードに検証ステップを追加し、レビューに合格しない場合は前のノードにロールバックして再実行します。
コスト管理：マルチエージェント協調のトークン消費はシングルエージェントの数倍です。解決策：タスクの複雑さに応じてエージェント数を動的に選択し、単純なタスクにはシングルエージェントを使用し、複雑なタスクにのみマルチエージェントを有効にします。
マルチエージェント協調の未来
AIの能力向上に伴い、マルチエージェント協調は実験から本番へと移行しています。将来のエージェントチームはより柔軟になり、エージェントは動的に参加・離脱し、タスク要件に応じてチーム規模を自動調整します。エージェント同士が互いに「雇用」し合い、分散型AI協調ネットワークを形成することも可能です。マルチエージェント協調を習得することは、将来のAIアプリケーション開発の中核的な競争力を習得することです。
このスキルチェーンを自分で編成してみませんか？
スキルチェーンで開く →