multi-agent-systems

マルチエージェントシステム

マルチエージェントアーキテクチャを使うタイミング

マルチエージェントシステムにはオーバーヘッドが伴います。エージェントを追加するたびに、障害が発生する可能性のある箇所、保守が必要なプロンプト、予期しない動作の発生源が増えます。

マルチエージェントシステムは、シングルエージェントのアプローチに比べてトークンを3～10倍使用します。その理由は次のとおりです。

• エージェント間でコンテキストが重複する
• エージェント間の調整メッセージが発生する
• 引き継ぎ時に結果を要約する

まずはシングルエージェントから始める

適切なツールを備えた、適切に設計されたシングルエージェントは、想定以上のことを実行できます。次の場合はシングルエージェントを使います。

• タスクが逐次的で、コンテキストに依存する
• ツール数が15～20未満である
• 並列化による明確な利点がない

マルチエージェントが特に有効な3つのケース

1. コンテキストの汚染 - サブタスクで1000トークンを超える情報が生成されるものの、大半はメインタスクに関係ない
2. 並列化 - タスクを独立して実行でき、より広い検索範囲を調査できる
3. 専門化 - タスクごとに異なるツール、プロンプト、または分野の専門知識が必要となる

判断フレームワーク

コンテキスト保護パターン

サブタスクで大量のコンテキストが生成されるものの、メインタスクでは要約のみが必要な場合に使います。

例：カスタマーサポート

class OrderLookupAgent:
    def lookup_order(self, order_id: str) -> dict:
        messages = [{"role": "user", "content": f"Get essential details for order {order_id}"}]
        response = client.messages.create(
            model="claude-sonnet-4-5", max_tokens=1024,
            messages=messages, tools=[get_order_details_tool]
        )
        return extract_summary(response)  # Returns 50-100 tokens, not 2000+

class SupportAgent:
    def handle_issue(self, user_message: str):
        if needs_order_info(user_message):
            order_id = extract_order_id(user_message)
            order_summary = OrderLookupAgent().lookup_order(order_id)
            context = f"Order {order_id}: {order_summary['status']}, purchased {order_summary['date']}"
        # Main agent gets clean context

特に有効な条件：

• サブタスクで1000トークンを超える情報が生成され、その大半が無関係である
• サブタスクが明確な抽出基準を備え、十分に定義されている
• 検索・取得した情報を使用前にフィルタリングする必要がある

並列化パターン

より広い検索範囲や、独立した調査の観点を調べる場合に使います。

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def research_topic(query: str) -> dict:
    facets = await lead_agent.decompose_query(query)
    tasks = [research_subagent(facet) for facet in facets]
    results = await asyncio.gather(*tasks)
    return await lead_agent.synthesize(results)

async def research_subagent(facet: str) -> dict:
    messages = [{"role": "user", "content": f"Research: {facet}"}]
    response = await client.messages.create(
        model="claude-sonnet-4-5", max_tokens=4096,
        messages=messages, tools=[web_search, read_document]
    )
    return extract_findings(response)

利点： 速度ではなく、網羅性です。トークンコストは高くなりますが、より広い範囲をカバーできます。

専門化パターン

ツールセットの専門化

エージェントが20個以上のツールを持つ場合、分野の混同が見られる場合、またはパフォーマンスが低下した場合は、分野ごとに分割します。

専門化が必要な兆候：

1. 数量：20個以上のツール
2. 分野の混同：ツールが互いに関連のない分野にまたがっている
3. パフォーマンスの低下：新しいツールの追加により、既存タスクの性能が落ちる

システムプロンプトの専門化

タスクごとに、相反する行動モードが必要な場合があります。

• カスタマーサポート：共感的で、忍耐強い
• コードレビュー：正確で、批判的
• コンプライアンス：厳格にルールに従う
• ブレインストーミング：創造性を柔軟に発揮する

分野の専門知識による専門化

汎用エージェントには負担が大きすぎる、深い分野固有のコンテキストが必要な場合があります。

• 法務分析：判例法、規制の枠組み
• 医学研究：臨床試験の方法論

マルチプラットフォーム統合の例

class CRMAgent:
    system_prompt = """You are a CRM specialist. You manage contacts,
    opportunities, and account records. Always verify record ownership
    before updates and maintain data integrity across related records."""
    tools = [crm_get_contacts, crm_create_opportunity]  # 8-10 CRM tools

class MarketingAgent:
    system_prompt = """You are a marketing automation specialist. You
    manage campaigns, lead scoring, and email sequences."""
    tools = [marketing_get_campaigns, marketing_create_lead]  # 8-10 tools

class OrchestratorAgent:
    def execute(self, user_request: str):
        response = client.messages.create(
            model="claude-sonnet-4-5", max_tokens=1024,
            system="""Route to appropriate specialist:
    - CRM: Contacts, opportunities, accounts, sales pipeline
    - Marketing: Campaigns, lead nurturing, email sequences""",
            messages=[{"role": "user", "content": user_request}],
            tools=[delegate_to_crm, delegate_to_marketing]
        )
        return response

コンテキスト中心の分割

問題中心（逆効果）： 作業の種類（ライター、テスター、レビュー担当者）で分割すると、調整のオーバーヘッドが発生し、引き継ぎ時にコンテキストが失われます。

コンテキスト中心（効果的）： 機能を担当するエージェントが、そのテストも担当します。必要なコンテキストをすでに持っているためです。

効果的な境界

• 独立した調査経路
• 明確なAPI契約を持つ個別のコンポーネント
• ブラックボックス検証

問題のある境界

• 同一作業における逐次的なフェーズ
• 密結合したコンポーネント
• 共有状態を必要とする作業

検証サブエージェントのパターン

メインエージェントの作業をテスト・検証する専用エージェントです。検証では必要なコンテキストの引き継ぎが最小限で済むため、うまく機能します。

class CodingAgent:
    def implement_feature(self, requirements: str) -> dict:
        response = client.messages.create(
            model="claude-sonnet-4-5", max_tokens=4096,
            messages=[{"role": "user", "content": f"Implement: {requirements}"}],
            tools=[read_file, write_file, list_directory]
        )
        return {"code": response.content, "files_changed": extract_files(response)}

class VerificationAgent:
    def verify_implementation(self, requirements: str, files_changed: list) -> dict:
        messages = [{"role": "user", "content": f"""
Requirements: {requirements}
Files changed: {files_changed}

Run the complete test suite and verify:
1. All existing tests pass
2. New functionality works as specified
3. No obvious errors or security issues

You MUST run: pytest --verbose
Only mark as PASSED if ALL tests pass with no failures.
"""}]
        response = client.messages.create(
            model="claude-sonnet-4-5", max_tokens=4096,
            messages=messages, tools=[run_tests, execute_code, read_file]
        )
        return {"passed": extract_pass_fail(response), "issues": extract_issues(response)}

「早すぎる成功」問題を防ぐ

検証担当が十分なテストをせずに合格と判定してしまう問題です。次の方法で防止できます。

• 具体的な基準： 「動作することを確認する」ではなく「テストスイート全体を実行する」
• 包括的なチェック： 複数のシナリオやエッジケースをテストする
• ネガティブテスト： 失敗すべき入力が確実に失敗することを確認する
• 明示的な指示： 「完全なテストスイートを必ず実行すること」

今後のチェックリスト

マルチエージェントの複雑さを加える前に、次の点を確認してください。

1. 実際の制約（コンテキストの上限、並列化、専門化の必要性）がある
2. 分割が問題の種類ではなく、コンテキストに基づいている
3. サブエージェントが検証できる、明確な検証ポイントがある

機能する最もシンプルなアプローチから始めてください。根拠がある場合にのみ、複雑さを加えてください。

参考資料

• マルチエージェントシステムの構築：活用するタイミングと方法
• 効果的なエージェントの構築
• AI エージェントのための効果的なコンテキストエンジニアリング
• エージェント向けの効果的なツールの作成