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, head to your [settings page](/settings).
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.