Why Multi-Agent Over Single-Agent?

Single-agent systems hit ceilings when tasks require:

Architecture Pattern 1: Pipeline (Sequential)

Each agent's output becomes the next agent's input. Clean, predictable, easy to debug.

import asyncio
import anthropic

client = anthropic.Anthropic()

class PipelineAgent:
    def __init__(self, name: str, system_prompt: str):
        self.name = name
        self.system_prompt = system_prompt

    async def process(self, input_text: str) -> str:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1000,
            system=self.system_prompt,
            messages=[{"role": "user", "content": input_text}]
        )
        return response.content[0].text

class IntelligencePipeline:
    def __init__(self):
        self.agents = [
            PipelineAgent("researcher", "You gather and summarize raw information."),
            PipelineAgent("analyst", "You analyze data and identify patterns and insights."),
            PipelineAgent("writer", "You write clear, actionable reports from analysis."),
            PipelineAgent("reviewer", "You review reports for accuracy and completeness.")
        ]

    async def run(self, query: str) -> str:
        result = query
        for agent in self.agents:
            print(f"Running {agent.name}...")
            result = await agent.process(result)
        return result

Architecture Pattern 2: Supervisor-Worker

A supervisor agent decomposes tasks and delegates to specialized workers. Best for complex workflows with dynamic task allocation.

class SupervisorAgent:
    def __init__(self):
        self.workers = {
            "research": ResearchAgent(),
            "code_review": CodeAgent(),
            "risk_analysis": RiskAgent(),
            "write": WriterAgent()
        }

    async def run(self, task: str) -> str:
        # Supervisor plans which agents to use
        plan_response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=500,
            system="""You are a task planner. Given a task, output a JSON plan.
            Available workers: research, code_review, risk_analysis, write
            Output: {"steps": [{"agent": "name", "task": "specific subtask"}]}""",
            messages=[{"role": "user", "content": f"Plan this task: {task}"}]
        )

        plan = json.loads(plan_response.content[0].text)
        results = []

        for step in plan["steps"]:
            worker = self.workers[step["agent"]]
            result = await worker.run(step["task"])
            results.append({"agent": step["agent"], "result": result})

        # Supervisor synthesizes
        synthesis = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1000,
            system="Synthesize these agent outputs into a coherent final answer.",
            messages=[{"role": "user", "content": str(results)}]
        )
        return synthesis.content[0].text

Real Example: Listing Radar AI

Listing Radar monitors crypto exchange announcements for trading opportunities. Our 4-agent pipeline:

class ListingRadarPipeline:
    """
    Monitor → Intelligence (Claude Sonnet) → Risk (Gemini Flash) → CEO (Claude Sonnet)
    """
    async def analyze_announcement(self, announcement: str) -> dict:
        # Agent 1: Intelligence Analysis (Claude — needs reasoning)
        intelligence = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=800,
            system="Analyze crypto listing announcements. Extract: token, exchange, listing date, price action pattern, market cap estimate.",
            messages=[{"role": "user", "content": announcement}]
        )

        # Agent 2: Risk Assessment (Gemini Flash — cheaper for classification)
        risk = await gemini_client.generate(
            model="gemini-2.0-flash",
            prompt=f"Rate risk 1-10 for this listing opportunity: {intelligence.content[0].text}. Output JSON: {{risk_score, reasons}}"
        )

        # Agent 3: CEO Decision (Claude — final high-stakes reasoning)
        decision = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=500,
            system="You are a crypto trading CEO. Make final trade decisions. Be concise and direct.",
            messages=[{"role": "user", "content": f"Intelligence: {intelligence.content[0].text}\nRisk: {risk}\nDecision?"}]
        )

        return {
            "intelligence": intelligence.content[0].text,
            "risk": json.loads(risk),
            "decision": decision.content[0].text
        }

Cost optimization: We use Gemini Flash-Lite for simple classification tasks (~10x cheaper than Claude Sonnet). Only expensive models where complex reasoning actually matters. On 1000 analyses/day, this difference is significant.

Pattern 3: Parallel with Aggregation

Multiple agents run simultaneously on different aspects, results aggregated at the end.

async def parallel_research(topic: str) -> dict:
    """Run 4 specialized agents in parallel, combine results."""
    tasks = [
        market_agent.research(topic),
        technical_agent.research(topic),
        competitor_agent.research(topic),
        regulatory_agent.research(topic)
    ]

    # asyncio.gather runs all in parallel
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Filter out any failures
    valid_results = [r for r in results if not isinstance(r, Exception)]

    # Aggregate with a synthesis agent
    aggregated = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1500,
        system="Synthesize these parallel research results into a comprehensive report.",
        messages=[{"role": "user", "content": str(valid_results)}]
    )
    return aggregated.content[0].text

State Management Between Agents

Agents need shared state. Our pattern: PostgreSQL for persistence, Redis for ephemeral session state.

import redis.asyncio as redis
import json

class AgentState:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.redis = redis.from_url("redis://localhost")

    async def set(self, key: str, value: any, ttl: int = 3600):
        await self.redis.setex(
            f"agent:{self.session_id}:{key}",
            ttl,
            json.dumps(value, default=str)
        )

    async def get(self, key: str) -> any:
        data = await self.redis.get(f"agent:{self.session_id}:{key}")
        return json.loads(data) if data else None

    async def append_history(self, agent_name: str, output: str):
        """Track agent execution history for debugging."""
        history_key = f"agent:{self.session_id}:history"
        current = await self.get("history") or []
        current.append({"agent": agent_name, "output": output[:500]})
        await self.set("history", current)

Cost Management Rules

Real cost numbers: Our Listing Radar analyzes ~200 announcements/day. Using Claude Sonnet for analysis + Gemini Flash for risk scoring: ~$8/day. If we used Claude Sonnet for everything: ~$35/day. The hybrid approach saves $810/month at this volume.

Error Handling & Resilience

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class ResilientAgent:
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    async def run_with_retry(self, prompt: str) -> str:
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1000,
                messages=[{"role": "user", "content": prompt}]
            )
            return response.content[0].text
        except anthropic.RateLimitError:
            print("Rate limited, retrying...")
            raise  # tenacity will retry
        except anthropic.APIError as e:
            print(f"API error: {e}")
            return f"Error: {str(e)}"  # don't retry on non-rate-limit errors