Antigravityで作るマルチエージェント協調システム：設計から実装まで

# Multi-Agent System Configuration
 
## Orchestrator Agent
役割: 全体の調整・タスク分解・品質管理
権限: 他の全エージェントへのタスク割り当て・状態確認
ツール: task_manager, status_monitor
 
## Researcher Agent
役割: 情報収集・データ取得・一次情報の整理
権限: web_search, file_read（読み取りのみ）
制約: 外部への書き込みを行わない
ツール: web_search, url_fetch, file_reader
 
## Analyst Agent
役割: 収集した情報の分析・パターン抽出・洞察生成
権限: データ処理ツールへのアクセス
入力: Researcher Agent からのデータ
ツール: data_processor, chart_generator
 
## Writer Agent
役割: 最終レポートの作成・フォーマット・ユーザー向け整理
権限: file_write（指定ディレクトリのみ）
入力: Analyst Agent からの分析結果
ツール: document_writer, markdown_formatter

from antigravity_sdk import AgentOrchestrator, Agent, Task
 
# エージェントの定義
orchestrator = Agent(
    name="orchestrator",
    instruction_file="agents/orchestrator.md",
    model="gemini-2.5-pro-latest",  # 高品質な判断が必要
)
 
researcher = Agent(
    name="researcher",
    instruction_file="agents/researcher.md",
    model="gemini-2.5-flash",  # 速度重視
    tools=["web_search", "url_fetch"]
)
 
analyst = Agent(
    name="analyst",
    instruction_file="agents/analyst.md",
    model="gemini-2.5-pro-latest",
    tools=["data_processor"]
)
 
writer = Agent(
    name="writer",
    instruction_file="agents/writer.md",
    model="gemini-2.5-flash",
    tools=["document_writer"]
)
 
# オーケストレーターにエージェントを登録
orch = AgentOrchestrator(
    master=orchestrator,
    workers=[researcher, analyst, writer]
)
 
# タスクの実行
result = orch.run(
    task="2026年の生成AI市場動向を調査・分析し、投資家向けレポートを作成してください",
    output_format="markdown",
    max_iterations=10
)
 
print(result.content)

import asyncio
from antigravity_sdk import ParallelAgentRunner
 
async def run_parallel_analysis(topic: str) -> dict:
    """
    複数の観点から同時に分析を行う並列マルチエージェント
    """
    runner = ParallelAgentRunner()
    
    # 同一トピックを3つの異なる観点から並列分析
    tasks = [
        runner.submit(
            agent="analyst",
            task=f"{topic}の技術的な観点からの分析",
            context={"perspective": "technical"}
        ),
        runner.submit(
            agent="analyst",
            task=f"{topic}のビジネス・市場観点からの分析",
            context={"perspective": "business"}
        ),
        runner.submit(
            agent="analyst",
            task=f"{topic}のリスク・課題観点からの分析",
            context={"perspective": "risks"}
        )
    ]
    
    # 並列で全タスクを実行（3x高速化）
    results = await asyncio.gather(*tasks)
    
    # 結果を統合エージェントに渡す
    aggregator = runner.get_agent("aggregator")
    final_report = await aggregator.synthesize(results)
    
    return final_report
 
# 実行
report = asyncio.run(run_parallel_analysis("量子コンピュータの商用化"))

from antigravity_sdk import AgentOrchestrator, AgentError, RetryPolicy
 
class ResilientOrchestrator:
    """
    フォールトトレラントなマルチエージェントオーケストレーター
    """
    
    def __init__(self):
        self.orchestrator = AgentOrchestrator(...)
        self.retry_policy = RetryPolicy(
            max_retries=3,
            backoff_factor=2.0,
            retry_on=[AgentError.TIMEOUT, AgentError.RATE_LIMIT]
        )
        self.fallback_agents = {
            "researcher": "researcher_backup",  # バックアップエージェント
        }
    
    async def run_with_recovery(self, task: str) -> dict:
        """エラーリカバリー付きでタスクを実行"""
        for attempt in range(self.retry_policy.max_retries):
            try:
                return await self.orchestrator.run(task)
                
            except AgentError.CONTEXT_OVERFLOW as e:
                # コンテキスト超過 → タスクを分割して再実行
                subtasks = self._split_task(task, e.failed_agent)
                return await self._run_subtasks(subtasks)
                
            except AgentError.AGENT_FAILURE as e:
                # エージェント失敗 → フォールバックエージェントで再試行
                failed_name = e.agent_name
                if failed_name in self.fallback_agents:
                    self.orchestrator.replace_agent(
                        failed_name,
                        self.fallback_agents[failed_name]
                    )
                    
            except AgentError.TIMEOUT:
                wait_time = self.retry_policy.backoff_factor ** attempt
                print(f"タイムアウト。{wait_time}秒後に再試行...")
                await asyncio.sleep(wait_time)
        
        raise Exception("最大リトライ回数超過")
    
    def _split_task(self, task: str, failed_agent: str) -> list[str]:
        """コンテキスト超過時のタスク分割ロジック"""
        # タスクを意味的に分割する実装
        ...

from antigravity_sdk import SharedMemory, AgentContext
 
# 共有メモリの初期化
shared_memory = SharedMemory(
    backend="redis",  # または "sqlite" / "in-memory"
    namespace="project-x"
)
 
class ResearchAgent:
    def __init__(self, memory: SharedMemory):
        self.memory = memory
    
    async def research(self, topic: str) -> None:
        # 調査結果を共有メモリに保存
        results = await self._web_search(topic)
        await self.memory.set(
            key=f"research:{topic}",
            value=results,
            ttl=3600  # 1時間キャッシュ
        )
 
class AnalystAgent:
    def __init__(self, memory: SharedMemory):
        self.memory = memory
    
    async def analyze(self, topic: str) -> str:
        # 共有メモリから調査結果を読み取り
        research_data = await self.memory.get(f"research:{topic}")
        if not research_data:
            raise ValueError(f"Research data for '{topic}' not found")
        
        return await self._analyze(research_data)

import logging
from antigravity_sdk import AgentLogger
 
# 構造化ログの設定
agent_logger = AgentLogger(
    output="./logs/agent_runs.jsonl",
    include_prompts=True,       # プロンプトをログに含める
    include_responses=True,     # 応答をログに含める
    include_tool_calls=True,    # ツール呼び出しをログに含める
    log_level=logging.INFO
)
 
# ログから実行コストを分析
from antigravity_sdk import CostAnalyzer
 
analyzer = CostAnalyzer("./logs/agent_runs.jsonl")
report = analyzer.generate_report()
print(f"総トークン消費: {report.total_tokens:,}")
print(f"推定コスト: ${report.estimated_cost:.4f}")
print(f"最もコスト高なエージェント: {report.most_expensive_agent}")