Multi-Agent Architect

マルチエージェントアーキテクト

複数の専門エージェントが協力して複雑な問題を解決するシステムを設計します。

基本原則

複雑なタスクを専門エージェントに分担させます。各エージェントはそれぞれの領域を担当し、明確なコミュニケーションパターンを通じて連携します。

マルチエージェントシステムを使用する場面

マルチエージェントを使用する場合：

• ✅ タスクに複数の専門分野が必要な場合（調査＋執筆＋コーディング）
• ✅ 並列処理によって解決が速くなる場合（独立したサブタスク）
• ✅ 相互レビューによる自己修正が必要な場合
• ✅ 判断ポイントを含む複雑なワークフローの場合
• ✅ 単一エージェントでの拡張が扱いにくくなった場合

マルチエージェントを使用しない場合：

• ❌ 単一のエージェントで効率よくタスクを処理できる場合
• ❌ タスクがシンプルで直線的な場合
• ❌ コミュニケーションのオーバーヘッドが並列化のメリットを上回る場合
• ❌ チームにマルチエージェントのデバッグ経験がない場合

---

マルチエージェントのパターン

パターン 1：逐次パイプライン

用途：各エージェントが前のエージェントの成果を引き継ぐ複数ステップのワークフロー

User Query → Researcher → Analyst → Writer → Editor → Output

例：調査レポートの作成

1. 調査担当：情報源を収集する
2. 分析担当：調査結果を統合する
3. 執筆担当：レポートの草稿を作成する
4. 編集担当：内容を整え、フォーマットする

メリット：依存関係が明確で、デバッグしやすい デメリット：逐次処理（並列化できない）、ボトルネックが発生する

---

パターン 2：階層型（マネージャー・ワーカー）

用途：複雑なタスクを並列処理するサブタスクに分割する場合

              Manager Agent
              /     |     \
    Worker 1   Worker 2   Worker 3
    (Search)   (Analyze)  (Summarize)
              \     |     /
              Aggregator Agent

例：競合他社を対象とした市場調査

• マネージャー：競合他社ごとの分析に分解する
• ワーカー：競合他社 A、B、C を並列で調査する
• 集約担当：調査結果をまとめる

メリット：並列化、専門性 デメリット：マネージャーの複雑さ、調整のオーバーヘッド

---

パターン 3：ピアコラボレーション（ラウンドテーブル）

用途：複数の視点によって品質が向上する場合

Coder ↔ Reviewer ↔ Tester
  ↓        ↓        ↓
      Consensus

例：レビューを伴うコード生成

1. コーダー：初期コードを作成する
2. レビュアー：問題がないか確認する
3. テスター：機能を検証する
4. 合意に達するまで繰り返す

メリット：レビューによる品質向上、自己修正 デメリット：合意に至らない場合がある、コストが高い（LLM の呼び出しが複数回必要）

---

パターン 4：エージェントスウォーム

用途：多数のエージェントが独立して解決策の空間を探索する場合

Agent 1 → Candidate Solution 1
Agent 2 → Candidate Solution 2
Agent 3 → Candidate Solution 3
   ↓
Selector (pick best)

例：創造的なブレインストーミング

• 5 つのエージェントが異なるアプローチを生成する
• 選定担当が評価して最適なものを選ぶ

メリット：探索、創造性 デメリット：コスト（N 個のエージェント）、似た解決策が生成される場合がある

---

コミュニケーションパターン

1. 共有メモリ

shared_state = {
    "research_findings": [],
    "current_task": "analyze_competitors",
    "decisions": []
}

# All agents read/write to shared state
researcher.execute(shared_state)
analyst.execute(shared_state)

メリット：シンプルで、すべてのエージェントが完全なコンテキストを確認できる デメリット：競合状態が発生しやすく、誰が何を変更したのかデバッグしにくい

---

2. メッセージパッシング

# Agent A sends message to Agent B
message = {
    "from": "researcher",
    "to": "analyst",
    "content": research_findings,
    "metadata": {"confidence": 0.9}
}

message_queue.send(message)

メリット：コミュニケーションの流れが明確で、追跡可能 デメリット：実装がより複雑

---

3. イベント駆動型

# Agents subscribe to events
event_bus.subscribe("research_complete", analyst.on_research_complete)
event_bus.subscribe("analysis_complete", writer.on_analysis_complete)

# Agent publishes event when done
event_bus.publish("research_complete", research_data)

メリット：疎結合で、スケーラブル デメリット：実行の流れを追いにくい

---

エージェントの調整戦略

1. 固定ワークフロー

動的な判断を行わない、事前定義された順序

workflow = [
    ("researcher", gather_info),
    ("analyst", analyze_data),
    ("writer", create_report)
]

for agent_name, task in workflow:
    result = agents[agent_name].execute(task, context)
    context.update(result)

用途：予測可能なタスク、明確な依存関係

---

2. 動的ルーティング

マネージャーがコンテキストに基づいて次のエージェントを決定する

class ManagerAgent:
    def route_task(self, task, context):
        if requires_technical_expertise(task):
            return tech_specialist
        elif requires_creative_input(task):
            return creative_agent
        else:
            return generalist

用途：タスクの変化が大きく、柔軟性が必要な場合

---

3. 合意ベース

エージェントが投票する、または合意に達する

proposals = [agent.propose_solution(task) for agent in agents]
scores = [agent.evaluate(proposals) for agent in agents]
best = proposals[argmax(mean(scores))]

用途：重大な判断、品質が最優先される場合

---

CrewAI による実装

CrewAI パターン（役割ベースのチーム）：

from crewai import Agent, Task, Crew

# Define specialized agents
researcher = Agent(
    role="Research Specialist",
    goal="Gather comprehensive information on {topic}",
    backstory="Expert researcher with 10 years experience",
    tools=[search_tool, scrape_tool]
)

analyst = Agent(
    role="Data Analyst",
    goal="Synthesize research findings into insights",
    backstory="Data scientist specialized in trend analysis",
    tools=[analysis_tool]
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, compelling reports",
    backstory="Professional writer with technical expertise",
    tools=[writing_tool]
)

# Define tasks
research_task = Task(
    description="Research {topic} thoroughly",
    agent=researcher,
    expected_output="Comprehensive research findings with sources"
)

analysis_task = Task(
    description="Analyze research findings for key insights",
    agent=analyst,
    context=[research_task],  # Depends on research_task
    expected_output="List of key insights and trends"
)

writing_task = Task(
    description="Write executive summary based on analysis",
    agent=writer,
    context=[research_task, analysis_task],
    expected_output="500-word executive summary"
)

# Create crew and execute
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    verbose=True
)

result = crew.kickoff(inputs={"topic": "AI market trends"})

---

LangGraph による実装

LangGraph パターン（ステートマシン）：

from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    input: str
    research: str
    analysis: str
    output: str

def research_node(state):
    research = researcher_agent.run(state["input"])
    return {"research": research}

def analysis_node(state):
    analysis = analyst_agent.run(state["research"])
    return {"analysis": analysis}

def writing_node(state):
    output = writer_agent.run(state["analysis"])
    return {"output": output}

# Build graph
workflow = StateGraph(AgentState)

workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("writing", writing_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "writing")
workflow.add_edge("writing", END)

app = workflow.compile()

# Execute
result = app.invoke({"input": "Analyze AI market trends"})

---

ベストプラクティス

1. エージェントの役割を明確にする

各エージェントには、具体的な専門知識と責務を持たせます。

2. コミュニケーションを最小限に抑える

エージェントが増えるほど、調整のオーバーヘッドも増加します。シンプルな構成から始めましょう。

3. 冪等な操作

副作用を伴わずにエージェントを再起動できるようにします。

4. 障害への対応

エージェントの障害を想定して設計します（再試行、フォールバック、スキップ）。

5. 実行を可視化する

エージェントの判断をログに記録し、実行の流れを追跡します。

6. コスト管理

エージェントごとのトークン使用量を追跡し、コストの高い呼び出しを最適化します。

---

マルチエージェントでよくある間違い

❌ エージェントが多すぎる → 2～3個から始め、必要な場合にのみ追加します ❌ 責任が不明確 → 明確な役割を定義します ❌ 障害への対処がない → 1つのエージェントの障害でシステム全体が停止します ❌ 同期処理がボトルネックになる → 独立したエージェントを並列化します ❌ コストを無視する → エージェントが N 個なら、LLM 呼び出しも N 倍になります ❌ 過剰な設計 → 多くの場合、単一のエージェントで十分です

---

判断フレームワーク：単一エージェントかマルチエージェントか

Task Complexity?
│
├─ Simple, linear → Single Agent
│
├─ Complex, requires specialization?
│  │
│  ├─ Sequential steps → Pipeline Pattern
│  ├─ Parallel subtasks → Hierarchical Pattern
│  ├─ Need review → Peer Collaboration
│  └─ Explore solutions → Swarm Pattern
│
└─ Uncertain → Start with Single Agent, refactor to Multi if needed

---

モニタリングとデバッグ

# Track agent execution
class TrackedAgent(Agent):
    def execute(self, task, context):
        start = time.time()
        logger.info(f"{self.name} starting: {task}")

        result = super().execute(task, context)

        duration = time.time() - start
        logger.info(f"{self.name} completed in {duration}s")

        metrics.record({
            "agent": self.name,
            "task": task,
            "duration": duration,
            "tokens": result.token_count,
            "cost": result.cost
        })

        return result

主なメトリクス：

• エージェントの実行時間
• エージェントごとのトークン使用量
• 成功率／失敗率
• 引き継ぎの遅延
• ワークフロー全体の所要時間

---

関連リソース

関連スキル：

• rag-implementer - ナレッジに基づくエージェント向け
• knowledge-graph-builder - エージェントのナレッジベース向け
• api-designer - エージェント間の通信 API 向け

関連パターン：

• META/DECISION-FRAMEWORK.md - フレームワークの選定（CrewAI と LangGraph の比較）
• STANDARDS/architecture-patterns/multi-agent-pattern.md - エージェントアーキテクチャ（作成後）

関連プレイブック：

• PLAYBOOKS/deploy-multi-agent-system.md - デプロイガイド（作成後）
• PLAYBOOKS/debug-agent-workflows.md - デバッグ手順（作成後）