◉ANTIGRAVITY LABEN
記事一覧/Agents & Manager
◈ Agents & Manager/2026-03-14上級

マルチエージェント・オーケストレーション実装ガイド — Antigravity で本番環境に耐えるAIパイプラインを構築する

Antigravity を使った本番グレードのマルチエージェントシステム設計。オーケストレーター/ワーカー分離、状態管理、エラーリカバリー、並列実行パターンを実コードで解説します。

Antigravity338multi-agent21orchestration15AI pipelineproduction59

取り組みの背景：なぜマルチエージェントか

単一エージェントでは解決困難な複雑なタスク — 大規模コードリファクタリング、並列データ収集、長大なドキュメント生成 — は、マルチエージェントアーキテクチャによって初めて実用的になります。

Antigravity のマルチエージェントサポートを活用すれば、タスクを分割・並列化し、エラー耐性のある自律的なパイプラインを構築できます。

本記事で構築するシステムの構成：

[オーケストレーター]
    ├── [リサーチエージェント × 3] (並列)
    ├── [コードレビューエージェント] (逐次)
    └── [レポート生成エージェント] (最終集約)
ℹ️
Antigravity の API キーが必要です。まだ取得していない場合は [設定ページ](/settings) から取得してください。

1. エージェント設計の原則

オーケストレーター vs ワーカーの分離

オーケストレーター
  役割: タスク分解、状態管理、結果集約
  特徴: 長いコンテキスト、低頻度呼び出し

ワーカーエージェント
  役割: 特定タスクの実行
  特徴: 短いコンテキスト、高頻度呼び出し、並列化可能

オーケストレーターに強力なモデル（例：claude-opus-4）、ワーカーには高速・低コストモデル（例：claude-haiku-4-5）を使い分けることでコストとレイテンシを最適化します。

2. 基本実装：タスク分解エンジン

# orchestrator.py
import asyncio
import json
from typing import Any
from dataclasses import dataclass, field
from enum import Enum
import anthropic
 
class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    RETRYING = "retrying"
 
@dataclass
class Task:
    id: str
    name: str
    description: str
    dependencies: list[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    error: str | None = None
    retries: int = 0
    max_retries: int = 3
 
class TaskGraph:
    """有向非巡回グラフ（DAG）によるタスク管理"""
 
    def __init__(self):
        self.tasks: dict[str, Task] = {}
        self.client = anthropic.Anthropic()
 
    def add_task(self, task: Task):
        self.tasks[task.id] = task
 
    def get_ready_tasks(self) -> list[Task]:
        """依存タスクがすべて完了しているタスクを返す"""
        ready = []
        for task in self.tasks.values():
            if task.status != TaskStatus.PENDING:
                continue
            deps_done = all(
                self.tasks[dep].status == TaskStatus.COMPLETED
                for dep in task.dependencies
                if dep in self.tasks
            )
            if deps_done:
                ready.append(task)
        return ready
 
    def is_complete(self) -> bool:
        return all(
            t.status in (TaskStatus.COMPLETED, TaskStatus.FAILED)
            for t in self.tasks.values()
        )
 
    def get_context_for_task(self, task: Task) -> dict:
        """依存タスクの結果をコンテキストとして収集"""
        context = {}
        for dep_id in task.dependencies:
            if dep_id in self.tasks and self.tasks[dep_id].result:
                context[dep_id] = self.tasks[dep_id].result
        return context

3. ワーカーエージェント実装

# worker_agent.py
import asyncio
import json
from anthropic import AsyncAnthropic
 
class WorkerAgent:
    """独立したタスクを実行するワーカーエージェント"""
 
    def __init__(
        self,
        agent_id: str,
        system_prompt: str,
        model: str = "claude-haiku-4-5-20251001",
        max_tokens: int = 4096,
    ):
        self.agent_id = agent_id
        self.system_prompt = system_prompt
        self.model = model
        self.max_tokens = max_tokens
        self.client = AsyncAnthropic()
        self.tools = []
 
    def add_tool(self, tool_definition: dict):
        self.tools.append(tool_definition)
 
    async def execute(
        self,
        task_description: str,
        context: dict = None,
        timeout: float = 120.0,
    ) -> dict:
        """タスクを実行して結果を返す"""
        messages = self._build_messages(task_description, context or {})
 
        try:
            response = await asyncio.wait_for(
                self._run_agent_loop(messages),
                timeout=timeout,
            )
            return {"status": "success", "result": response, "agent_id": self.agent_id}
        except asyncio.TimeoutError:
            return {"status": "timeout", "result": None, "agent_id": self.agent_id}
        except Exception as e:
            return {"status": "error", "result": None, "error": str(e), "agent_id": self.agent_id}
 
    def _build_messages(self, task: str, context: dict) -> list:
        context_str = ""
        if context:
            context_str = "\n\n## 依存タスクの結果\n"
            for dep_id, result in context.items():
                context_str += f"\n### {dep_id}\n{json.dumps(result, ensure_ascii=False, indent=2)}\n"
 
        return [
            {
                "role": "user",
                "content": f"{context_str}\n\n## 実行するタスク\n{task}"
            }
        ]
 
    async def _run_agent_loop(self, messages: list) -> str:
        """ツール呼び出しを含むエージェントループ"""
        max_iterations = 10
 
        for iteration in range(max_iterations):
            kwargs = {
                "model": self.model,
                "max_tokens": self.max_tokens,
                "system": self.system_prompt,
                "messages": messages,
            }
            if self.tools:
                kwargs["tools"] = self.tools
 
            response = await self.client.messages.create(**kwargs)
 
            # ツール使用の処理
            tool_uses = [b for b in response.content if b.type == "tool_use"]
 
            if not tool_uses or response.stop_reason == "end_turn":
                # テキスト結果を抽出
                text_blocks = [b.text for b in response.content if hasattr(b, "text")]
                return "\n".join(text_blocks)
 
            # ツール結果をメッセージに追加
            messages.append({"role": "assistant", "content": response.content})
            tool_results = await self._execute_tools(tool_uses)
            messages.append({"role": "user", "content": tool_results})
 
        return "最大イテレーション数に達しました"
 
    async def _execute_tools(self, tool_uses: list) -> list:
        """ツール呼び出しを実行（サブクラスでオーバーライド）"""
        results = []
        for tool_use in tool_uses:
            results.append({
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": f"ツール {tool_use.name} は未実装です",
            })
        return results

4. 並列実行オーケストレーター

# parallel_orchestrator.py
import asyncio
from task_graph import TaskGraph, Task, TaskStatus
from worker_agent import WorkerAgent
 
class ParallelOrchestrator:
    """DAG に基づいてタスクを並列実行するオーケストレーター"""
 
    def __init__(self, max_concurrent: int = 5):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.graph = TaskGraph()
        self.workers: dict[str, WorkerAgent] = {}
 
    def register_worker(self, task_pattern: str, worker: WorkerAgent):
        """タスクのパターンに対応するワーカーを登録"""
        self.workers[task_pattern] = worker
 
    def _get_worker_for_task(self, task: Task) -> WorkerAgent:
        """タスク名に最適なワーカーを返す"""
        for pattern, worker in self.workers.items():
            if pattern in task.name or pattern == "*":
                return worker
        # デフォルトワーカー
        return WorkerAgent(
            agent_id="default",
            system_prompt="あなたは汎用タスク実行エージェントです。",
        )
 
    async def execute_task(self, task: Task) -> None:
        """セマフォで同時実行数を制御しながらタスクを実行"""
        async with self.semaphore:
            task.status = TaskStatus.RUNNING
            worker = self._get_worker_for_task(task)
            context = self.graph.get_context_for_task(task)
 
            while task.retries <= task.max_retries:
                if task.retries > 0:
                    task.status = TaskStatus.RETRYING
                    # 指数バックオフ
                    await asyncio.sleep(2 ** task.retries)
 
                result = await worker.execute(
                    task_description=task.description,
                    context=context,
                )
 
                if result["status"] == "success":
                    task.result = result["result"]
                    task.status = TaskStatus.COMPLETED
                    print(f"✅ タスク完了: {task.name}")
                    return
 
                task.retries += 1
                task.error = result.get("error", "不明なエラー")
                print(f"⚠️ タスク失敗 (試行 {task.retries}/{task.max_retries}): {task.name} — {task.error}")
 
            task.status = TaskStatus.FAILED
            print(f"❌ タスク最終失敗: {task.name}")
 
    async def run(self) -> dict:
        """DAG 全体を実行して結果を返す"""
        print(f"🚀 オーケストレーター開始: {len(self.graph.tasks)} タスク")
 
        while not self.graph.is_complete():
            ready_tasks = self.graph.get_ready_tasks()
 
            if not ready_tasks:
                # デッドロック検出
                running = [t for t in self.graph.tasks.values() if t.status == TaskStatus.RUNNING]
                if not running:
                    print("⛔ デッドロック検出 — 実行を中止します")
                    break
                await asyncio.sleep(0.1)
                continue
 
            # 準備できたタスクを並列実行
            await asyncio.gather(
                *[self.execute_task(task) for task in ready_tasks],
                return_exceptions=True,
            )
 
        return {
            task_id: task.result
            for task_id, task in self.graph.tasks.items()
            if task.status == TaskStatus.COMPLETED
        }

5. 実用例：コードベース分析パイプライン

# example_pipeline.py
import asyncio
from parallel_orchestrator import ParallelOrchestrator
from task_graph import Task
from worker_agent import WorkerAgent
 
async def analyze_codebase(repo_path: str) -> str:
    orchestrator = ParallelOrchestrator(max_concurrent=3)
 
    # ワーカー定義
    analyzer = WorkerAgent(
        agent_id="code-analyzer",
        system_prompt="""あなたはコードアナリストです。
コードの品質、セキュリティ問題、パフォーマンスボトルネックを分析してください。
結果は JSON 形式で返してください。""",
        model="claude-haiku-4-5-20251001",
    )
 
    reviewer = WorkerAgent(
        agent_id="security-reviewer",
        system_prompt="""あなたはセキュリティエキスパートです。
OWASP Top 10 の観点からコードを審査してください。""",
        model="claude-haiku-4-5-20251001",
    )
 
    reporter = WorkerAgent(
        agent_id="report-generator",
        system_prompt="""あなたは技術レポートライターです。
分析結果を経営陣向けのエグゼクティブサマリーにまとめてください。""",
        model="claude-sonnet-4-6",  # 最終集約は高品質モデル
    )
 
    orchestrator.register_worker("analyze", analyzer)
    orchestrator.register_worker("security", reviewer)
    orchestrator.register_worker("report", reporter)
 
    # タスクグラフ構築
    tasks = [
        Task(
            id="analyze_frontend",
            name="analyze-frontend",
            description=f"{repo_path}/src/frontend を分析し、品質問題を特定してください。",
        ),
        Task(
            id="analyze_backend",
            name="analyze-backend",
            description=f"{repo_path}/src/backend を分析し、品質問題を特定してください。",
        ),
        Task(
            id="security_check",
            name="security-review",
            description=f"{repo_path} 全体のセキュリティ問題を特定してください。",
        ),
        Task(
            id="final_report",
            name="report-generation",
            description="前の分析結果をすべて集約し、優先度付きの改善提案レポートを作成してください。",
            dependencies=["analyze_frontend", "analyze_backend", "security_check"],
        ),
    ]
 
    for task in tasks:
        orchestrator.graph.add_task(task)
 
    results = await orchestrator.run()
    return results.get("final_report", "レポート生成に失敗しました")
 
if __name__ == "__main__":
    result = asyncio.run(analyze_codebase("/path/to/your/project"))
    print(result)

6. 本番運用のチェックリスト

# production_checklist.py
 
"""
本番マルチエージェントシステムのチェックリスト
 
✅ タイムアウト設定
   - 各ワーカーにタイムアウトを設定 (asyncio.wait_for)
   - デフォルト: 120秒
 
✅ リトライ戦略
   - max_retries: 3
   - 指数バックオフ: 2^n 秒 (2, 4, 8 秒)
   - 冪等性の確保（同じタスクを複数回実行しても安全）
 
✅ コスト管理
   - ワーカー: claude-haiku-4-5 (低コスト・高速)
   - オーケストレーター: claude-opus-4 (高精度)
   - 最大トークン数の明示的な設定
 
✅ 観測可能性
   - 各タスクの開始・終了・エラーをログ
   - 実行時間の計測
   - コスト追跡 (usage.input_tokens + usage.output_tokens)
 
✅ グレースフルデグレード
   - 非クリティカルなタスク失敗時は結果なしで続行
   - クリティカルタスク失敗時はパイプライン停止
"""
 
import time
import logging
from dataclasses import dataclass
 
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
 
@dataclass
class ExecutionMetrics:
    task_id: str
    start_time: float
    end_time: float = 0.0
    input_tokens: int = 0
    output_tokens: int = 0
 
    @property
    def duration(self) -> float:
        return self.end_time - self.start_time
 
    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens
 
    def to_dict(self) -> dict:
        return {
            "task_id": self.task_id,
            "duration_sec": round(self.duration, 2),
            "total_tokens": self.total_tokens,
        }

まとめ

マルチエージェントシステムを本番で動かすには、単にエージェントを並べるだけでなく、タスクグラフ設計・エラーリカバリー・コスト管理 の三位一体が必要です。

このガイドで構築したアーキテクチャは、小規模な自動化スクリプトから数十エージェントの大規模パイプラインまでスケールします。まずは 2〜3 エージェントの小さなパイプラインから始め、段階的に複雑さを加えていくことをお勧めします。

個人開発12年とアーティスト活動から見るエージェント設計

線引きするときの3つの判断軸

  • 失敗時の影響が金銭やユーザー体験にどれだけ波及するか
  • 復旧オペレーションが明文化されているか
  • 観測ログから人間が再現できる粒度に整っているか

まとめ — 検証ステップの推奨

  1. 観測メトリクスとアラートを設置
  2. 限定ロールアウトで本番検証

12年の個人開発から見た所感

検証時のチェックリスト

  • 本番相当の負荷をかけた状態で再現できるか
  • ログだけでなくメトリクスにエラー率を出しているか
  • 失敗時の通知が遅延なく届くか

次のステップ

シェア

お読みいただきありがとうございます

Antigravity Lab は広告なしで運営しており、サーバー費用などの運営コストはメンバーシップのご支援で賄っています。実装コード・ベンチマーク・本番設計パターンなど、実務でお役立ていただける記事を毎日更新しています。もし読んでよかったと感じていただけましたら、ぜひご覧ください。

  • ✦コピー&ペーストで使える実装コード付き
  • ✦毎日新しい上級ガイドを追加
  • ✦¥580/月 または ¥1,480 の永久アクセス
メンバーシップを見る →

もしこの記事がお役に立ちましたら、チップ（¥150）で応援いただけると大変励みになります。広告なしでの運営を続けるため、皆さまのご支援が大きな力になっています。

関連記事

◈ Agents & Manager2026-06-15
Antigravity のマルチエージェントを本番で壊さない封じ込め設計 — 連鎖障害を止める3つの境界の実装メモ
Antigravity のマルチエージェント構成は単体では綺麗に動くのに、本番では小さな失敗が全体に波及します。連鎖障害を止めるための制御の階層化・信頼境界・観測と冪等性という3つの境界を、TOML 設定と相関IDラッパーの実装まで含めて整理しました。
◈ Agents & Manager2026-04-10
Antigravityマルチエージェント設計の実践ガイド — エージェント間通信エラーから本番安定稼働まで
Antigravityでマルチエージェントを設計・実装する完全ガイド。エージェント間通信エラーの解決から本番稼働パターン、パフォーマンス最適化まで体系的に解説します。
◈ Agents & Manager2026-03-21
Antigravity マルチエージェント本番パターン — 委譲設計・並列実行・コスト管理を個人開発の実務目線で
Antigravityの本番マルチエージェント編成を個人開発の実務目線で整理します。5つの編成パターンに加え、委譲の判断基準、並列度の決め方、コスト監視の指標まで、破綻させずに黒字で回すための設計を解説します。
📚RECOMMENDED BOOKS
大規模言語モデル入門
山田育矢
LLM開発
生成AIプロンプトエンジニアリング入門
我妻幸長
プロンプト
Claude CodeによるAI駆動開発入門
平川知秀
AI駆動開発
※ アフィリエイトリンクを含みます
もっと見る →