Building Multi-Agent Collaboration Systems in Antigravity: From Design to Production

# Multi-Agent System Configuration
 
## Orchestrator Agent
Role: Coordinate overall execution, decompose tasks, manage quality
Authority: Assign tasks to all worker agents, query status
Tools: task_manager, status_monitor
 
## Researcher Agent
Role: Information gathering, data retrieval, raw source organization
Authority: Read-only access to external data sources
Constraint: No write operations to external systems
Tools: web_search, url_fetch, file_reader
 
## Analyst Agent
Role: Analyze collected information, extract patterns, generate insights
Input: Data from Researcher Agent
Tools: data_processor, chart_generator
 
## Writer Agent
Role: Create final report, handle formatting, tailor for audience
Authority: Write access to designated output directories only
Input: Insights from Analyst Agent
Tools: document_writer, markdown_formatter

from antigravity_sdk import AgentOrchestrator, Agent, Task
 
# Define agents
orchestrator = Agent(
    name="orchestrator",
    instruction_file="agents/orchestrator.md",
    model="gemini-2.5-pro-latest",  # High-quality reasoning required
)
 
researcher = Agent(
    name="researcher",
    instruction_file="agents/researcher.md",
    model="gemini-2.5-flash",       # Speed prioritized
    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"]
)
 
# Register all agents with the orchestrator
orch = AgentOrchestrator(
    master=orchestrator,
    workers=[researcher, analyst, writer]
)
 
# Execute the task
result = orch.run(
    task="Research 2026 generative AI market trends and write an investor briefing report",
    output_format="markdown",
    max_iterations=10
)
 
print(result.content)

import asyncio
from antigravity_sdk import ParallelAgentRunner
 
async def run_parallel_analysis(topic: str) -> dict:
    """
    Analyze a topic from three perspectives simultaneously.
    Reduces total execution time by ~3x compared to sequential.
    """
    runner = ParallelAgentRunner()
    
    tasks = [
        runner.submit(
            agent="analyst",
            task=f"Technical analysis of {topic}",
            context={"perspective": "technical"}
        ),
        runner.submit(
            agent="analyst",
            task=f"Business and market analysis of {topic}",
            context={"perspective": "business"}
        ),
        runner.submit(
            agent="analyst",
            task=f"Risk and challenge analysis of {topic}",
            context={"perspective": "risks"}
        )
    ]
    
    # Run all three in parallel
    results = await asyncio.gather(*tasks)
    
    # Synthesize with aggregator agent
    aggregator = runner.get_agent("aggregator")
    final_report = await aggregator.synthesize(results)
    
    return final_report
 
report = asyncio.run(run_parallel_analysis("Quantum computing commercialization"))

from antigravity_sdk import AgentOrchestrator, AgentError, RetryPolicy
 
class ResilientOrchestrator:
    """
    Fault-tolerant multi-agent orchestrator with error recovery.
    """
    
    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:
        """Execute task with automatic error recovery."""
        for attempt in range(self.retry_policy.max_retries):
            try:
                return await self.orchestrator.run(task)
                
            except AgentError.CONTEXT_OVERFLOW as e:
                # Context exceeded — split the task and retry
                subtasks = self._split_task(task, e.failed_agent)
                return await self._run_subtasks(subtasks)
                
            except AgentError.AGENT_FAILURE as e:
                # Agent failure — swap in the fallback agent
                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"Timeout. Retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)
        
        raise Exception("Max retries exceeded")
    
    def _split_task(self, task: str, failed_agent: str) -> list[str]:
        """Split an overflowing task into semantically coherent subtasks."""
        ...

from antigravity_sdk import SharedMemory
 
shared_memory = SharedMemory(
    backend="redis",   # or "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)
        # Write results to shared memory for other agents to consume
        await self.memory.set(
            key=f"research:{topic}",
            value=results,
            ttl=3600  # 1-hour cache
        )
 
class AnalystAgent:
    def __init__(self, memory: SharedMemory):
        self.memory = memory
    
    async def analyze(self, topic: str) -> str:
        # Read from shared memory — no direct coupling to ResearchAgent
        research_data = await self.memory.get(f"research:{topic}")
        if not research_data:
            raise ValueError(f"Research data for '{topic}' not found in shared memory")
        
        return await self._analyze(research_data)

from antigravity_sdk import AgentLogger
 
agent_logger = AgentLogger(
    output="./logs/agent_runs.jsonl",
    include_prompts=True,
    include_responses=True,
    include_tool_calls=True,
)
 
# Analyze cost from logs
from antigravity_sdk import CostAnalyzer
 
analyzer = CostAnalyzer("./logs/agent_runs.jsonl")
report = analyzer.generate_report()
print(f"Total tokens consumed: {report.total_tokens:,}")
print(f"Estimated cost: ${report.estimated_cost:.4f}")
print(f"Most expensive agent: {report.most_expensive_agent}")