Why Multi-Agent Architecture?
Single-agent systems hit a ceiling with tasks requiring parallel data collection, large-scale refactoring, or long-form document generation. Multi-agent architecture solves this by decomposing complex work across specialized agents running concurrently.
The system we'll build:
[Orchestrator Agent]
├── [Research Agent × 3] (parallel)
├── [Code Review Agent] (sequential)
└── [Report Generation Agent] (final aggregation)
You'll need an Antigravity API key. If you haven't obtained one yet, obtain one from the Antigravity settings panel first.
1. Core Design Principles
Orchestrator vs Worker Separation
Orchestrator
Role: Task decomposition, state management, result aggregation
Characteristics: Long context window, infrequent calls, high-quality model
Worker Agents
Role: Execute specific sub-tasks
Characteristics: Short context, high frequency, parallelizable
Use a powerful model (e.g., claude-opus-4) for the orchestrator and fast/cheap models (e.g., claude-haiku-4-5) for workers to balance cost and latency.
2. Task Graph Engine
# task_graph.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:
"""Directed Acyclic Graph (DAG) for managing task execution order"""
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]:
"""Return tasks whose dependencies are all complete"""
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:
"""Gather completed dependency results as context"""
return {
dep_id: self.tasks[dep_id].result
for dep_id in task.dependencies
if dep_id in self.tasks and self.tasks[dep_id].result
}3. Worker Agent Implementation
# worker_agent.py
import asyncio
import json
from anthropic import AsyncAnthropic
class WorkerAgent:
"""Executes a single, well-scoped task"""
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## Context from completed dependencies\n"
for dep_id, result in context.items():
context_str += f"\n### {dep_id}\n{json.dumps(result, indent=2)}\n"
return [
{
"role": "user",
"content": f"{context_str}\n\n## Your task\n{task}"
}
]
async def _run_agent_loop(self, messages: list) -> str:
max_iterations = 10
for _ 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":
return "\n".join(b.text for b in response.content if hasattr(b, "text"))
messages.append({"role": "assistant", "content": response.content})
tool_results = await self._execute_tools(tool_uses)
messages.append({"role": "user", "content": tool_results})
return "Reached max iterations"
async def _execute_tools(self, tool_uses: list) -> list:
return [
{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": f"Tool {tool_use.name} not implemented",
}
for tool_use in tool_uses
]4. Parallel Orchestrator
# parallel_orchestrator.py
import asyncio
from task_graph import TaskGraph, Task, TaskStatus
from worker_agent import WorkerAgent
class ParallelOrchestrator:
"""Runs tasks in parallel respecting DAG dependency order"""
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="You are a general-purpose task execution agent.",
)
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) # exponential backoff
result = await worker.execute(
task_description=task.description,
context=context,
)
if result["status"] == "success":
task.result = result["result"]
task.status = TaskStatus.COMPLETED
print(f"✅ Completed: {task.name}")
return
task.retries += 1
task.error = result.get("error", "Unknown error")
print(f"⚠️ Failed (attempt {task.retries}/{task.max_retries}): {task.name}")
task.status = TaskStatus.FAILED
print(f"❌ Permanently failed: {task.name}")
async def run(self) -> dict:
print(f"🚀 Orchestrator starting: {len(self.graph.tasks)} 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("⛔ Deadlock detected — aborting")
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. Real-World Example: Codebase Analysis Pipeline
# codebase_analysis.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="""You are a code quality analyst.
Analyze code for quality issues, tech debt, and performance bottlenecks.
Return results as structured JSON.""",
model="claude-haiku-4-5-20251001",
)
security_reviewer = WorkerAgent(
agent_id="security-reviewer",
system_prompt="""You are a security expert.
Review code from an OWASP Top 10 perspective.
Identify and categorize all security vulnerabilities.""",
model="claude-haiku-4-5-20251001",
)
reporter = WorkerAgent(
agent_id="report-generator",
system_prompt="""You are a technical writer.
Synthesize analysis results into a prioritized executive summary
with actionable recommendations.""",
model="claude-sonnet-4-6", # Higher quality model for final aggregation
)
orchestrator.register_worker("analyze", analyzer)
orchestrator.register_worker("security", security_reviewer)
orchestrator.register_worker("report", reporter)
tasks = [
Task(
id="analyze_frontend",
name="analyze-frontend",
description=f"Analyze {repo_path}/src/frontend for code quality issues.",
),
Task(
id="analyze_backend",
name="analyze-backend",
description=f"Analyze {repo_path}/src/backend for code quality issues.",
),
Task(
id="security_check",
name="security-review",
description=f"Perform a security audit of {repo_path}.",
),
Task(
id="final_report",
name="report-generation",
description="Aggregate all analysis results into a prioritized improvement report.",
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", "Report generation failed")
if __name__ == "__main__":
result = asyncio.run(analyze_codebase("/path/to/your/project"))
print(result)6. Production Checklist
"""
Production multi-agent checklist
✅ Timeouts
- Every worker has asyncio.wait_for timeout (default: 120s)
- Global pipeline timeout to prevent runaway costs
✅ Retry Strategy
- max_retries: 3
- Exponential backoff: 2^n seconds (2, 4, 8)
- Idempotent tasks (safe to run multiple times)
✅ Cost Management
- Workers: claude-haiku-4-5 (fast, cheap)
- Orchestrator: claude-opus-4 (accurate planning)
- Explicit max_tokens on every call
✅ Observability
- Log task start/end/error with timestamps
- Track token usage per task
- Alert on consecutive failures
✅ Graceful Degradation
- Non-critical task failures: continue with None result
- Critical task failures: abort pipeline and notify
"""Looking back
Multi-agent systems demand more than stringing agents together — you need DAG-based task management, exponential backoff retries, and thoughtful cost allocation.
The architecture in this guide scales from a 2-agent automation script to 50+ agent pipelines without structural changes. Start small: build a 2-agent pipeline today, validate it end-to-end, then layer in complexity.