Multi-Agent Pattern: Orchestrating Agent Teams

The multi-agent pattern takes AI systems to the next level: instead of one agent doing everything, specialized agents collaborate like a well-coordinated team. Each agent focuses on what it does best, and together they accomplish more than any single agent could alone.

This pattern mirrors how human organizations work—specialists collaborate, managers coordinate, and teams divide complex work into manageable pieces.

Why Multi-Agent Systems?

The Limits of Single Agents

Single agents struggle with:

• Breadth vs Depth: Can't be expert at everything
• Context Limits: Long complex tasks exhaust context windows
• Mode Confusion: Switching between different types of work
• Scalability: Can't parallelize their own work

Multi-Agent Advantages

Teams of agents overcome these limits:

• Specialization: Each agent masters one domain
• Parallelism: Multiple agents work simultaneously
• Focused Context: Each agent has relevant context only
• Modularity: Add, remove, or update agents independently
• Resilience: One agent's failure doesn't stop all work

Core Multi-Agent Architectures

Architecture 1: Manager-Worker

A manager agent delegates to specialized workers:

         ┌──────────┐
         │ Manager  │
         │  Agent   │
         └────┬─────┘
              │
    ┌─────────┼─────────┐
    │         │         │
    ▼         ▼         ▼
┌───────┐ ┌───────┐ ┌───────┐
│Worker │ │Worker │ │Worker │
│   A   │ │   B   │ │   C   │
└───────┘ └───────┘ └───────┘

from anthropic import Anthropic
from dataclasses import dataclass
from typing import Dict, List, Any
import json

@dataclass
class Agent:
    name: str
    role: str
    system_prompt: str

class ManagerWorkerSystem:
    def __init__(self):
        self.client = Anthropic()
        self.workers: Dict[str, Agent] = {}
        self.manager: Agent = None

    def set_manager(self, agent: Agent):
        self.manager = agent

    def add_worker(self, agent: Agent):
        self.workers[agent.name] = agent

    def run(self, task: str) -> str:
        # Manager analyzes and delegates
        delegation_plan = self._manager_plan(task)

        # Execute delegated tasks
        results = {}
        for assignment in delegation_plan["assignments"]:
            worker_name = assignment["worker"]
            subtask = assignment["task"]

            worker = self.workers.get(worker_name)
            if worker:
                result = self._worker_execute(worker, subtask, results)
                results[worker_name] = result

        # Manager synthesizes
        return self._manager_synthesize(task, results)

    def _manager_plan(self, task: str) -> dict:
        worker_descriptions = "\n".join([
            f"- {name}: {w.role}"
            for name, w in self.workers.items()
        ])

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=self.manager.system_prompt,
            messages=[{
                "role": "user",
                "content": f"""Analyze this task and delegate to workers:

Task: {task}

Available workers:
{worker_descriptions}

Return a JSON plan:
{{
    "assignments": [
        {{"worker": "worker_name", "task": "specific subtask"}}
    ]
}}"""
            }]
        )

        return json.loads(response.content[0].text)

    def _worker_execute(
        self,
        worker: Agent,
        task: str,
        context: dict
    ) -> str:
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            system=worker.system_prompt,
            messages=[{
                "role": "user",
                "content": f"""Complete this task:
{task}

Context from other workers:
{json.dumps(context, indent=2) if context else "None yet"}"""
            }]
        )

        return response.content[0].text

    def _manager_synthesize(self, original_task: str, results: dict) -> str:
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            system=self.manager.system_prompt,
            messages=[{
                "role": "user",
                "content": f"""Synthesize results for the original task:

Original Task: {original_task}

Worker Results:
{json.dumps(results, indent=2)}

Provide a comprehensive final response."""
            }]
        )

        return response.content[0].text


# Example usage
system = ManagerWorkerSystem()

system.set_manager(Agent(
    name="project_manager",
    role="Coordinates work across specialists",
    system_prompt="""You are a project manager. Analyze tasks, delegate to
    appropriate specialists, and synthesize their outputs into cohesive results."""
))

system.add_worker(Agent(
    name="researcher",
    role="Finds and analyzes information",
    system_prompt="You are a research specialist. Find relevant information and provide thorough analysis."
))

system.add_worker(Agent(
    name="writer",
    role="Creates polished content",
    system_prompt="You are a skilled writer. Create clear, engaging content."
))

system.add_worker(Agent(
    name="critic",
    role="Reviews and improves work",
    system_prompt="You are a constructive critic. Review work and suggest improvements."
))

result = system.run("Create a blog post about AI agents in healthcare")

Architecture 2: Pipeline

Agents process sequentially, each transforming the output:

Input → Agent A → Agent B → Agent C → Output
         (Research) (Draft)   (Edit)

class PipelineSystem:
    def __init__(self):
        self.client = Anthropic()
        self.stages: List[Agent] = []

    def add_stage(self, agent: Agent):
        self.stages.append(agent)

    def run(self, initial_input: str) -> str:
        current = initial_input

        for stage in self.stages:
            current = self._process_stage(stage, current)

        return current

    def _process_stage(self, agent: Agent, input_data: str) -> str:
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=agent.system_prompt,
            messages=[{
                "role": "user",
                "content": f"Process this input according to your role:\n\n{input_data}"
            }]
        )

        return response.content[0].text


# Example: Content creation pipeline
pipeline = PipelineSystem()

pipeline.add_stage(Agent(
    name="researcher",
    role="Research",
    system_prompt="Research the topic thoroughly. Output key facts and sources."
))

pipeline.add_stage(Agent(
    name="outliner",
    role="Outline",
    system_prompt="Create a detailed outline based on the research. Include section headers and key points."
))

pipeline.add_stage(Agent(
    name="writer",
    role="Draft",
    system_prompt="Write a complete draft following the outline. Be engaging and clear."
))

pipeline.add_stage(Agent(
    name="editor",
    role="Edit",
    system_prompt="Edit for clarity, grammar, and flow. Improve weak sections."
))

result = pipeline.run("Write about sustainable energy solutions")

Architecture 3: Debate

Agents argue different perspectives:

      ┌──────────┐
      │   Task   │
      └────┬─────┘
           │
    ┌──────┴──────┐
    │             │
    ▼             ▼
┌───────┐    ┌───────┐
│Pro    │◄──►│ Con   │
│Agent  │    │ Agent │
└───────┘    └───────┘
    │             │
    └──────┬──────┘
           │
           ▼
      ┌──────────┐
      │  Judge   │
      └──────────┘

class DebateSystem:
    def __init__(self):
        self.client = Anthropic()
        self.pro_agent: Agent = None
        self.con_agent: Agent = None
        self.judge_agent: Agent = None

    def debate(self, topic: str, rounds: int = 2) -> dict:
        debate_history = []

        # Opening statements
        pro_opening = self._get_response(
            self.pro_agent,
            f"Argue IN FAVOR of: {topic}\n\nProvide your opening statement."
        )
        debate_history.append({"agent": "pro", "type": "opening", "content": pro_opening})

        con_opening = self._get_response(
            self.con_agent,
            f"Argue AGAINST: {topic}\n\nProvide your opening statement."
        )
        debate_history.append({"agent": "con", "type": "opening", "content": con_opening})

        # Rebuttals
        for round_num in range(rounds):
            # Pro rebuttal
            pro_rebuttal = self._get_response(
                self.pro_agent,
                f"""Topic: {topic}

Debate history:
{self._format_history(debate_history)}

Provide your rebuttal to the opposition's latest points."""
            )
            debate_history.append({"agent": "pro", "type": "rebuttal", "content": pro_rebuttal})

            # Con rebuttal
            con_rebuttal = self._get_response(
                self.con_agent,
                f"""Topic: {topic}

Debate history:
{self._format_history(debate_history)}

Provide your rebuttal to the opposition's latest points."""
            )
            debate_history.append({"agent": "con", "type": "rebuttal", "content": con_rebuttal})

        # Judge's verdict
        verdict = self._get_response(
            self.judge_agent,
            f"""Evaluate this debate:

Topic: {topic}

Full debate:
{self._format_history(debate_history)}

Provide:
1. Summary of both positions
2. Strengths and weaknesses of each argument
3. Your verdict and reasoning"""
        )

        return {
            "topic": topic,
            "debate": debate_history,
            "verdict": verdict
        }

    def _get_response(self, agent: Agent, prompt: str) -> str:
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=agent.system_prompt,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

    def _format_history(self, history: list) -> str:
        return "\n\n".join([
            f"[{h['agent'].upper()} - {h['type']}]\n{h['content']}"
            for h in history
        ])

Architecture 4: Consensus

Multiple agents must agree:

class ConsensusSystem:
    def __init__(self, agreement_threshold: float = 0.7):
        self.client = Anthropic()
        self.agents: List[Agent] = []
        self.threshold = agreement_threshold

    def add_agent(self, agent: Agent):
        self.agents.append(agent)

    def reach_consensus(self, question: str, max_rounds: int = 3) -> dict:
        """Have agents discuss until reaching consensus"""

        for round_num in range(max_rounds):
            # Get each agent's position
            positions = []
            for agent in self.agents:
                position = self._get_position(agent, question, positions)
                positions.append({
                    "agent": agent.name,
                    "position": position
                })

            # Check for consensus
            agreement_level = self._measure_agreement(positions)

            if agreement_level >= self.threshold:
                return {
                    "consensus_reached": True,
                    "agreement_level": agreement_level,
                    "positions": positions,
                    "rounds": round_num + 1
                }

            # Agents revise based on others' positions
            question = self._synthesize_for_next_round(question, positions)

        # No consensus after max rounds
        return {
            "consensus_reached": False,
            "agreement_level": agreement_level,
            "positions": positions,
            "rounds": max_rounds
        }

    def _get_position(
        self,
        agent: Agent,
        question: str,
        previous_positions: list
    ) -> str:
        context = ""
        if previous_positions:
            context = "\n\nOther agents' positions:\n" + "\n".join([
                f"- {p['agent']}: {p['position'][:200]}..."
                for p in previous_positions
            ])

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=agent.system_prompt,
            messages=[{
                "role": "user",
                "content": f"{question}{context}\n\nProvide your position."
            }]
        )
        return response.content[0].text

    def _measure_agreement(self, positions: list) -> float:
        """Use an LLM to measure agreement level"""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=256,
            messages=[{
                "role": "user",
                "content": f"""Rate the level of agreement between these positions from 0.0 to 1.0:

{json.dumps(positions, indent=2)}

Respond with just a decimal number."""
            }]
        )
        return float(response.content[0].text.strip())

Communication Patterns

Direct Messaging

Agents communicate point-to-point:

class DirectMessageBus:
    def __init__(self):
        self.mailboxes: Dict[str, List[dict]] = {}

    def send(self, from_agent: str, to_agent: str, message: str):
        if to_agent not in self.mailboxes:
            self.mailboxes[to_agent] = []

        self.mailboxes[to_agent].append({
            "from": from_agent,
            "message": message,
            "timestamp": datetime.now()
        })

    def receive(self, agent: str) -> List[dict]:
        messages = self.mailboxes.get(agent, [])
        self.mailboxes[agent] = []
        return messages

Shared Blackboard

Agents read/write to shared state:

class Blackboard:
    def __init__(self):
        self.state: Dict[str, Any] = {}
        self.history: List[dict] = []

    def write(self, agent: str, key: str, value: Any):
        self.state[key] = value
        self.history.append({
            "agent": agent,
            "action": "write",
            "key": key,
            "timestamp": datetime.now()
        })

    def read(self, key: str) -> Any:
        return self.state.get(key)

    def read_all(self) -> Dict[str, Any]:
        return self.state.copy()

Pub/Sub

Agents subscribe to topics:

class PubSubBus:
    def __init__(self):
        self.subscriptions: Dict[str, List[str]] = {}  # topic -> agents
        self.messages: Dict[str, List[dict]] = {}  # agent -> messages

    def subscribe(self, agent: str, topic: str):
        if topic not in self.subscriptions:
            self.subscriptions[topic] = []
        self.subscriptions[topic].append(agent)

    def publish(self, from_agent: str, topic: str, message: str):
        for subscriber in self.subscriptions.get(topic, []):
            if subscriber not in self.messages:
                self.messages[subscriber] = []
            self.messages[subscriber].append({
                "topic": topic,
                "from": from_agent,
                "message": message
            })

    def get_messages(self, agent: str) -> List[dict]:
        messages = self.messages.get(agent, [])
        self.messages[agent] = []
        return messages

Coordination Strategies

Token-Based Turn Taking

class TokenCoordinator:
    def __init__(self, agents: List[str]):
        self.agents = agents
        self.current_holder = 0

    def get_current_agent(self) -> str:
        return self.agents[self.current_holder]

    def pass_token(self):
        self.current_holder = (self.current_holder + 1) % len(self.agents)

    def run_round_robin(self, task: str, rounds: int = 1) -> List[dict]:
        results = []

        for _ in range(rounds * len(self.agents)):
            agent = self.get_current_agent()
            result = self.execute_turn(agent, task, results)
            results.append({"agent": agent, "result": result})
            self.pass_token()

        return results

Priority Queue

import heapq

class PriorityCoordinator:
    def __init__(self):
        self.queue = []
        self.counter = 0  # For stable sorting

    def add_task(self, priority: int, agent: str, task: str):
        heapq.heappush(self.queue, (priority, self.counter, agent, task))
        self.counter += 1

    def get_next(self) -> tuple:
        if self.queue:
            priority, _, agent, task = heapq.heappop(self.queue)
            return agent, task, priority
        return None, None, None

Error Handling in Multi-Agent Systems

class ResilientMultiAgentSystem:
    def __init__(self):
        self.agents: Dict[str, Agent] = {}
        self.backup_agents: Dict[str, str] = {}  # primary -> backup mapping

    def set_backup(self, primary: str, backup: str):
        self.backup_agents[primary] = backup

    async def execute_with_fallback(
        self,
        agent_name: str,
        task: str,
        max_retries: int = 2
    ) -> dict:
        agents_to_try = [agent_name]

        # Add backup if exists
        if agent_name in self.backup_agents:
            agents_to_try.append(self.backup_agents[agent_name])

        for attempt, current_agent in enumerate(agents_to_try * max_retries):
            try:
                result = await self._execute_agent(current_agent, task)
                return {
                    "success": True,
                    "agent": current_agent,
                    "result": result,
                    "attempts": attempt + 1
                }
            except Exception as e:
                if attempt == len(agents_to_try) * max_retries - 1:
                    return {
                        "success": False,
                        "error": str(e),
                        "attempts": attempt + 1
                    }

        return {"success": False, "error": "All attempts failed"}

Monitoring Multi-Agent Systems

class MultiAgentMonitor:
    def __init__(self):
        self.metrics = {
            "tasks_completed": 0,
            "tasks_failed": 0,
            "agent_stats": {},
            "messages_sent": 0,
            "total_latency": 0
        }

    def record_task(self, agent: str, success: bool, latency: float):
        if success:
            self.metrics["tasks_completed"] += 1
        else:
            self.metrics["tasks_failed"] += 1

        self.metrics["total_latency"] += latency

        if agent not in self.metrics["agent_stats"]:
            self.metrics["agent_stats"][agent] = {
                "tasks": 0, "successes": 0, "total_latency": 0
            }

        self.metrics["agent_stats"][agent]["tasks"] += 1
        if success:
            self.metrics["agent_stats"][agent]["successes"] += 1
        self.metrics["agent_stats"][agent]["total_latency"] += latency

    def get_summary(self) -> dict:
        total_tasks = self.metrics["tasks_completed"] + self.metrics["tasks_failed"]
        return {
            "total_tasks": total_tasks,
            "success_rate": self.metrics["tasks_completed"] / total_tasks if total_tasks > 0 else 0,
            "avg_latency": self.metrics["total_latency"] / total_tasks if total_tasks > 0 else 0,
            "agent_performance": {
                agent: {
                    "success_rate": stats["successes"] / stats["tasks"] if stats["tasks"] > 0 else 0,
                    "avg_latency": stats["total_latency"] / stats["tasks"] if stats["tasks"] > 0 else 0
                }
                for agent, stats in self.metrics["agent_stats"].items()
            }
        }

Best Practices

1. Clear Agent Boundaries

Each agent should have:

• Single, focused responsibility
• Well-defined inputs and outputs
• Clear expertise domain

2. Explicit Communication

• Use structured message formats
• Log all inter-agent communication
• Make handoffs explicit

3. Graceful Degradation

• Plan for agent failures
• Implement fallback mechanisms
• Maintain partial functionality

4. Resource Management

• Monitor token usage
• Implement rate limiting
• Set timeouts for agent operations

5. Testing

• Test agents individually
• Test agent interactions
• Test failure scenarios

Conclusion

The multi-agent pattern enables sophisticated AI systems that mirror human organizational structures. By coordinating specialized agents, you can tackle complex problems that would overwhelm any single agent.

Key principles:

1. Specialize agents for focused expertise
2. Choose appropriate coordination architecture
3. Design clear communication protocols
4. Handle failures gracefully
5. Monitor and optimize performance

Multi-agent systems are the future of complex AI applications—start simple and scale up as you learn.

---

Want to start with simpler agent architectures? Check out Level 1 Agents: Basic Responders for foundational patterns.