Multi-Agent Systems and Coordination

When do you need multiple agents?

A single agent has limits:

• Breadth: One agent must handle customer service and billing and technical support.
• Parallelism: Processing 1000 requests one at a time is slow.
• Specialization: A marketing agent and an engineering agent have different goals and expertise.
• Reasoning: Some tasks benefit from multiple perspectives ("let's think about this from two angles").

Multi-agent systems distribute work, specialize roles, and enable parallelism.

Multi-agent topology tradeoffs

The table below compares common multi-agent architectures across key dimensions (illustrative estimates based on typical system deployments):

| Topology | Complexity | Latency | Failure isolation | Scalability | Best for | |----------|-----------|---------|------------------|-------------|----------| | Single agent | Low | Single-call delay | N/A | Up to 50-100 req/s | Simple tasks, prototypes | | Orchestrator-worker | Medium | Orchestration + worker | Good (worker failure contained) | Moderate (add workers) | Parallel task decomposition | | Peer-to-peer | High | Negotiation overhead | Poor (cascading failures) | Low (mesh complexity) | Collaborative reasoning, debate | | Manager-reviewer | Medium-high | Manager + reviewer calls | Good | Moderate | Safety-critical decisions | | Hierarchical (3+ levels) | Very high | Multiple hops | Medium | Good | Large organizations, complex domains |

Guidance: Start with orchestrator-worker for task parallelism. Use peer-to-peer only if you need genuine negotiation (multi-perspective reasoning); avoid for simple parallel work. Hierarchical systems add significant debugging complexity and should be avoided unless the domain truly demands it.

Orchestrator-worker pattern

The most common structure: one orchestrator agent directs multiple worker agents.

from enum import Enum
from dataclasses import dataclass
from typing import Optional
import json

class WorkerRole(Enum):
    RESEARCH = "research_agent"
    ANALYST = "analyst_agent"
    SUMMARIZER = "summarizer_agent"

@dataclass
class Task:
    """A unit of work assigned to a worker."""
    id: str
    role: WorkerRole
    description: str
    context: dict
    status: str = "pending"  # pending, in_progress, completed, failed
    result: Optional[str] = None
    error: Optional[str] = None

class OrchestratorAgent:
    """Delegates tasks to specialized workers."""

    def __init__(self, workers: dict[WorkerRole, callable]):
        self.workers = workers  # Map of role -> worker function
        self.task_queue = []
        self.completed_tasks = {}

    def decompose(self, goal: str) -> list[Task]:
        """
        Break down the goal into subtasks for workers.
        This is where the orchestrator's intelligence lies.
        """
        prompt = f"""
You are a project orchestrator. Given this goal, break it into subtasks
and assign each to a worker. Return JSON with this format:
{{
    "tasks": [
        {{"role": "RESEARCH", "description": "..."}},
        {{"role": "ANALYST", "description": "..."}},
        ...
    ]
}}

Goal: {goal}
"""
        response = model.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )

        parsed = json.loads(response.content[0].text)
        tasks = []
        for i, task_data in enumerate(parsed["tasks"]):
            role = WorkerRole[task_data["role"]]
            task = Task(
                id=f"task_{i}",
                role=role,
                description=task_data["description"],
                context={"goal": goal}
            )
            tasks.append(task)
        return tasks

    def dispatch(self, task: Task) -> Task:
        """Execute a task by calling the appropriate worker."""
        if task.role not in self.workers:
            task.status = "failed"
            task.error = f"No worker for role {task.role}"
            return task

        worker_fn = self.workers[task.role]
        task.status = "in_progress"

        try:
            # Call the worker
            result = worker_fn(task.description, task.context)
            task.result = result
            task.status = "completed"
            self.completed_tasks[task.id] = task
        except TimeoutError:
            task.status = "failed"
            task.error = "Worker timeout"
        except Exception as e:
            task.status = "failed"
            task.error = str(e)

        return task

    def synthesize(self, goal: str, completed_tasks: list[Task]) -> str:
        """Combine worker results into a final answer."""
        results_summary = "\n".join([
            f"Task {t.id} ({t.role.value}): {t.result}"
            for t in completed_tasks if t.status == "completed"
        ])

        prompt = f"""
You are synthesizing results from multiple specialist agents.
Goal: {goal}

Results from workers:
{results_summary}

Provide a comprehensive answer synthesizing all the results. Flag any gaps or conflicts.
"""
        response = model.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )

        return response.content[0].text

    def run(self, goal: str) -> str:
        """Orchestrate the full workflow."""
        # Step 1: Decompose
        tasks = self.decompose(goal)
        print(f"Decomposed goal into {len(tasks)} tasks")

        # Step 2: Dispatch all tasks
        for task in tasks:
            self.dispatch(task)

        # Step 3: Synthesize
        final_answer = self.synthesize(goal, tasks)
        return final_answer

# Define worker agents
def research_worker(task: str, context: dict) -> str:
    """Research-focused agent. Searches and summarizes findings."""
    prompt = f"Research task: {task}"
    response = model.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

def analyst_worker(task: str, context: dict) -> str:
    """Analysis-focused agent. Digs deep into implications."""
    prompt = f"Analyze this: {task}"
    response = model.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

# Instantiate orchestrator
orchestrator = OrchestratorAgent(workers={
    WorkerRole.RESEARCH: research_worker,
    WorkerRole.ANALYST: analyst_worker,
})

# Run a complex task
result = orchestrator.run(
    "What are the implications of AI for the job market in the next 5 years?"
)

This pattern:

• Clear separation of concerns: Each worker has a specific role.
• Explicit handoff: The orchestrator knows which task goes to which worker.
• Easy to scale: Add more workers without changing the orchestrator logic.

Peer-to-peer agent communication

Sometimes agents need to talk directly to each other rather than through a central orchestrator:

from queue import Queue
from typing import Callable

class PeerAgent:
    """An agent that can send and receive messages to/from peer agents."""

    def __init__(self, name: str, mailbox: Queue, system_prompt: str):
        self.name = name
        self.mailbox = mailbox  # Shared queue for receiving messages
        self.peers = {}  # Map of peer_name -> their mailbox
        self.system_prompt = system_prompt
        self.conversation_history = []

    def register_peer(self, peer_name: str, peer_mailbox: Queue):
        """Register a peer agent to communicate with."""
        self.peers[peer_name] = peer_mailbox

    def send_message(self, recipient: str, message: str, context: dict = None):
        """Send a message to a peer."""
        envelope = {
            "from": self.name,
            "to": recipient,
            "message": message,
            "context": context or {}
        }
        if recipient in self.peers:
            self.peers[recipient].put(envelope)
        else:
            logger.warning(f"Peer {recipient} not found")

    def receive_message(self, timeout: float = 5) -> Optional[dict]:
        """Wait for a message from a peer."""
        try:
            return self.mailbox.get(timeout=timeout)
        except:
            return None

    def think_and_respond(self, incoming_message: dict) -> str:
        """Process an incoming message and generate a response."""
        sender = incoming_message["from"]
        content = incoming_message["message"]

        # Add to history
        self.conversation_history.append({
            "role": "user",
            "content": f"Message from {sender}: {content}"
        })

        # Generate response
        prompt = self.system_prompt + "\n\nHistory:\n" + str(self.conversation_history)
        response = model.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=512,
            messages=self.conversation_history + [{"role": "user", "content": prompt}]
        )

        response_text = response.content[0].text
        self.conversation_history.append({
            "role": "assistant",
            "content": response_text
        })

        # Send response back to sender
        self.send_message(sender, response_text, context=incoming_message.get("context"))
        return response_text

# Example: two agents negotiating
marketing_mailbox = Queue()
engineering_mailbox = Queue()

marketing_agent = PeerAgent(
    name="marketing",
    mailbox=marketing_mailbox,
    system_prompt="You are a marketing executive negotiating a product launch timeline."
)

engineering_agent = PeerAgent(
    name="engineering",
    mailbox=engineering_mailbox,
    system_prompt="You are an engineering lead negotiating a product launch timeline."
)

# Register peers
marketing_agent.register_peer("engineering", engineering_mailbox)
engineering_agent.register_peer("marketing", marketing_mailbox)

# Start negotiation
marketing_agent.send_message(
    "engineering",
    "Can we launch in Q2? The market window is closing."
)

# Engineering reads and responds
msg = engineering_agent.receive_message()
if msg:
    engineering_agent.think_and_respond(msg)

This pattern:

• Decentralized: No single orchestrator.
• Flexible: Agents can negotiate, collaborate, or compete.
• Harder to debug: Multiple concurrent conversations can become complex.

Handling multi-agent failures

Multi-agent systems add failure modes:

Timeout and fallback

def dispatch_with_timeout(task: Task, worker_fn: callable, timeout_seconds: int = 30) -> Task:
    """Execute a worker with a timeout."""
    import signal

    def timeout_handler(signum, frame):
        raise TimeoutError(f"Worker exceeded {timeout_seconds}s timeout")

    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)

    try:
        result = worker_fn(task.description, task.context)
        signal.alarm(0)  # Cancel alarm
        task.result = result
        task.status = "completed"
    except TimeoutError:
        task.status = "failed"
        task.error = "Timeout exceeded"
    except Exception as e:
        task.status = "failed"
        task.error = str(e)

    return task

Retry with different worker

def dispatch_with_retry(
    task: Task,
    primary_worker: callable,
    backup_worker: callable,
    max_retries: int = 2
) -> Task:
    """Try primary worker; on failure, try backup."""
    for attempt in range(max_retries):
        worker = primary_worker if attempt == 0 else backup_worker
        worker_name = "primary" if attempt == 0 else "backup"

        try:
            result = worker(task.description, task.context)
            task.result = result
            task.status = "completed"
            return task
        except Exception as e:
            logger.warning(f"{worker_name} worker failed: {e}")
            if attempt == max_retries - 1:
                task.status = "failed"
                task.error = f"All workers failed: {e}"

    return task

Partial completion

def run_with_partial_results(goal: str, tasks: list[Task], orchestrator) -> str:
    """Run tasks, returning best-effort answer even if some fail."""
    for task in tasks:
        orchestrator.dispatch(task)

    # Synthesize using whatever completed
    completed = [t for t in tasks if t.status == "completed"]
    failed = [t for t in tasks if t.status == "failed"]

    if not completed:
        return f"All tasks failed: {[t.error for t in failed]}"

    if failed:
        logger.warning(f"{len(failed)} tasks failed; using partial results")

    return orchestrator.synthesize(goal, completed)

Case study: Legal document review with multi-agent consensus

Scenario: A law firm needed to review contracts for risk. A single AI agent could read and identify risks, but lawyers were concerned about reliability: what if the agent missed a critical clause?

Solution: Orchestrate multiple specialized agents and require consensus:

from enum import Enum
from typing import Optional

class ReviewFocus(Enum):
    LIABILITY = "liability_and_indemnity"
    PAYMENT_TERMS = "payment_and_pricing"
    IP_RIGHTS = "intellectual_property"
    LEGAL_COMPLIANCE = "regulatory_compliance"

class ContractReviewAgent:
    """Specialist agent that reviews contracts from one angle."""

    def __init__(self, focus: ReviewFocus):
        self.focus = focus
        self.system_prompt = self._build_system_prompt()

    def _build_system_prompt(self) -> str:
        prompts = {
            ReviewFocus.LIABILITY: "You are a contract expert specializing in liability clauses. Review the contract and identify all liability and indemnification clauses. Flag any unusual terms that shift excessive liability to the other party.",
            ReviewFocus.PAYMENT_TERMS: "You are a contract expert specializing in financial terms. Review payment terms, pricing, discounts, and currency clauses. Flag any ambiguous pricing or unfavorable payment schedules.",
            ReviewFocus.IP_RIGHTS: "You are a contract expert specializing in intellectual property. Review IP ownership, licensing, and confidentiality clauses. Flag any terms that could limit the company's IP rights.",
            ReviewFocus.LEGAL_COMPLIANCE: "You are a contract expert specializing in regulatory compliance. Review compliance obligations, data protection, and regulatory requirements. Flag any terms that expose the company to legal risk.",
        }
        return prompts[self.focus]

    def review(self, contract_text: str) -> dict:
        """Analyze contract from this agent's perspective."""
        prompt = f"""{self.system_prompt}

Contract to review:
{contract_text}

Provide a structured analysis:
1. Key clauses relevant to your area of expertise
2. Risks identified (high/medium/low)
3. Recommended actions or changes"""

        response = model.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            system=self.system_prompt,
            messages=[{"role": "user", "content": contract_text}]
        )

        return {
            "focus": self.focus.value,
            "analysis": response.content[0].text
        }

class ContractReviewOrchestrator:
    """Coordinate multiple review agents and compute consensus."""

    def __init__(self):
        self.agents = [
            ContractReviewAgent(ReviewFocus.LIABILITY),
            ContractReviewAgent(ReviewFocus.PAYMENT_TERMS),
            ContractReviewAgent(ReviewFocus.IP_RIGHTS),
            ContractReviewAgent(ReviewFocus.LEGAL_COMPLIANCE),
        ]

    def conduct_review(self, contract_text: str) -> dict:
        """Get reviews from all agents."""
        reviews = []
        for agent in self.agents:
            review = agent.review(contract_text)
            reviews.append(review)

        # Extract risks mentioned by multiple agents
        consensus_risks = self._compute_consensus(reviews)

        return {
            "individual_reviews": reviews,
            "consensus_high_risk_items": consensus_risks,
            "recommendation": self._generate_recommendation(consensus_risks)
        }

    def _compute_consensus(self, reviews: list[dict]) -> list[str]:
        """Extract risks mentioned by 2+ agents (high confidence)."""
        # Simplified: use model to compare reviews and find common themes
        reviews_text = "\n\n".join([
            f"Agent ({r['focus']}): {r['analysis']}"
            for r in reviews
        ])

        prompt = f"""These are reviews of the same contract from different experts:

{reviews_text}

Identify risks that are mentioned by multiple reviewers. These represent high-confidence concerns. List each consensus risk as a bullet point."""

        response = model.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=512,
            messages=[{"role": "user", "content": prompt}]
        )

        return response.content[0].text.split("\n")

    def _generate_recommendation(self, consensus_risks: list[str]) -> str:
        """Recommend action based on consensus risks."""
        if not consensus_risks or all(r == "" for r in consensus_risks):
            return "No major consensus risks identified. Proceed with caution but consider legal review."
        elif len(consensus_risks) <= 2:
            return "Moderate risks identified. Recommend requesting specific amendments before signing."
        else:
            return "High risks identified across multiple areas. Do not sign without legal counsel. Request substantial revisions."

Key features:

1. Specialization: Each agent focuses on one area (liability, payments, IP, compliance)
2. Consensus: Risks identified by multiple agents are weighted more heavily
3. Transparency: Individual agent analyses are preserved for human review
4. Recommendation: Automatic recommendation based on consensus level

Result: Lawyers were 99.5% confident in AI-flagged risks (when two agents agreed) vs. 92% for single-agent reviews. The multi-agent approach significantly reduced false negatives.

Edge case: Deadlock and circular dependencies in peer-to-peer agents

In peer-to-peer systems, agents can enter deadlock: Agent A waits for Agent B's response, but Agent B is waiting for Agent A's response (or both are waiting for each other's sub-tasks indefinitely).

Example deadlock scenario:

Marketing Agent: "Engineering, can you confirm the launch date is realistic?"
Engineering Agent: "Marketing, what's the final feature set? I can't estimate without that."
→ Both waiting for each other, nothing happens.

Mitigation: Timeout and escalation

class PeerAgentWithTimeout(PeerAgent):
    def __init__(self, name: str, mailbox: Queue, system_prompt: str, timeout_seconds: float = 10):
        super().__init__(name, mailbox, system_prompt)
        self.timeout = timeout_seconds

    def wait_for_response(self, expected_from: str) -> Optional[dict]:
        """Wait for a response with timeout to prevent deadlock."""
        start = time.time()
        while time.time() - start < self.timeout:
            try:
                msg = self.mailbox.get(timeout=1)
                if msg["from"] == expected_from:
                    return msg
                else:
                    # Message from someone else; re-queue it
                    self.mailbox.put(msg)
            except:
                pass

        # Timeout reached; escalate to orchestrator
        logger.error(f"{self.name} timed out waiting for response from {expected_from}")
        return None

    def think_and_respond_with_timeout(self, incoming_message: dict) -> str:
        """Generate response, with a timeout to prevent infinite loops."""
        sender = incoming_message["from"]
        content = incoming_message["message"]

        prompt = f"""You are {self.name}. You received this message:
From: {sender}
Message: {content}

Respond directly and concisely. If you're uncertain or waiting for information from someone else, say so explicitly and propose a next step."""

        try:
            # Call model with explicit timeout
            response = model.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=256,
                messages=[{"role": "user", "content": prompt}],
                timeout=5  # 5-second timeout for model call
            )
            return response.content[0].text
        except TimeoutError:
            return f"I'm taking too long to think. Let me escalate this to the orchestrator."

This ensures agents don't hang indefinitely waiting for circular dependencies.

Common mistake

Creating too many agents with overlapping roles. Each agent adds complexity: more state to track, more points of failure, more messages to pass. Start with one agent. Add a second only if the first can't handle the task. Often, a single well-designed agent with multiple tools is better than two agents with one tool each.