Multi-Agent Cooperation: Building Agent Teams

A single AI agent has limits. But a team of specialized agents, working together, can accomplish far more than any individual agent. This is the promise of multi-agent systems: collective intelligence that exceeds the sum of its parts.

This guide explores how to design, build, and orchestrate multi-agent systems that cooperate effectively.

Why Multi-Agent Systems?

Single Agent Limitations

Even the best single agent struggles with:

• Scope: Too many responsibilities dilute effectiveness
• Context: Long tasks exhaust context windows
• Consistency: Multiple personas in one agent create confusion
• Scalability: One agent can't parallelize work

Multi-Agent Advantages

Teams of agents overcome these limitations:

• Specialization: Each agent masters one thing
• Parallelization: Multiple agents work simultaneously
• Resilience: Failure of one doesn't stop all
• Modularity: Add, remove, or update agents independently
• Realism: Mirrors how human teams work

Core Cooperation Patterns

Pattern 1: Sequential Pipeline

Agents process in sequence, each adding value:

Input → Agent A → Agent B → Agent C → Output
         │          │          │
       Research   Draft      Review

Example: Content Production Pipeline

class ContentPipeline:
    def __init__(self):
        self.researcher = ResearchAgent()
        self.writer = WritingAgent()
        self.editor = EditingAgent()
        self.seo = SEOAgent()

    def produce_content(self, topic: str) -> str:
        # Stage 1: Research
        research = self.researcher.run(
            f"Research the topic: {topic}. Gather key facts, statistics, and expert opinions."
        )

        # Stage 2: Write
        draft = self.writer.run(
            f"Write an article about {topic}.\n\nResearch notes:\n{research}"
        )

        # Stage 3: Edit
        edited = self.editor.run(
            f"Edit this draft for clarity, flow, and accuracy:\n\n{draft}"
        )

        # Stage 4: SEO
        final = self.seo.run(
            f"Optimize this article for search engines:\n\n{edited}"
        )

        return final

When to use: Tasks with clear sequential stages.

Pattern 2: Hub and Spoke

A coordinator routes work to specialists:

              ┌─────────┐
              │Coordinator│
              └─────┬─────┘
                    │
     ┌──────────────┼──────────────┐
     │              │              │
     ▼              ▼              ▼
┌─────────┐  ┌─────────┐  ┌─────────┐
│Specialist│  │Specialist│  │Specialist│
│    A    │  │    B    │  │    C    │
└─────────┘  └─────────┘  └─────────┘

Example: Customer Support Router

class SupportSystem:
    def __init__(self):
        self.router = RouterAgent()
        self.specialists = {
            "billing": BillingAgent(),
            "technical": TechnicalAgent(),
            "general": GeneralAgent(),
            "escalation": HumanEscalationAgent()
        }

    def handle_ticket(self, ticket: str) -> str:
        # Router determines the right specialist
        routing = self.router.run(f"""
        Analyze this support ticket and determine the best handler:
        - billing: Payment, subscription, refund issues
        - technical: Bugs, errors, technical problems
        - general: How-to questions, feature info
        - escalation: Angry customers, complex issues

        Ticket: {ticket}

        Respond with just the handler name.
        """)

        handler = routing.strip().lower()
        specialist = self.specialists.get(handler, self.specialists["general"])

        return specialist.run(f"Handle this support ticket:\n{ticket}")

When to use: Diverse request types needing different expertise.

Pattern 3: Parallel Processing

Multiple agents work simultaneously:

              ┌─────────┐
              │  Task   │
              └─────┬───┘
                    │
    ┌───────────────┼───────────────┐
    │               │               │
    ▼               ▼               ▼
┌───────┐      ┌───────┐      ┌───────┐
│Agent A│      │Agent B│      │Agent C│
└───┬───┘      └───┬───┘      └───┬───┘
    │               │               │
    └───────────────┼───────────────┘
                    │
                    ▼
              ┌───────────┐
              │ Aggregator│
              └───────────┘

Example: Comprehensive Research

import asyncio

class ResearchTeam:
    def __init__(self):
        self.web_researcher = WebResearchAgent()
        self.academic_researcher = AcademicAgent()
        self.news_researcher = NewsAgent()
        self.synthesizer = SynthesisAgent()

    async def research(self, topic: str) -> str:
        # Run research in parallel
        results = await asyncio.gather(
            self.web_researcher.arun(f"Research {topic} from web sources"),
            self.academic_researcher.arun(f"Find academic papers about {topic}"),
            self.news_researcher.arun(f"Find recent news about {topic}")
        )

        web_results, academic_results, news_results = results

        # Synthesize findings
        synthesis = await self.synthesizer.arun(f"""
        Synthesize these research findings on {topic}:

        Web Research:
        {web_results}

        Academic Research:
        {academic_results}

        News Coverage:
        {news_results}

        Provide a comprehensive, balanced summary.
        """)

        return synthesis

When to use: Independent subtasks that can run concurrently.

Pattern 4: Debate and Consensus

Agents argue perspectives, then converge:

┌──────────┐     ┌──────────┐
│Advocate A│ ←→  │Advocate B│
└─────┬────┘     └────┬─────┘
      │               │
      └───────┬───────┘
              │
              ▼
        ┌──────────┐
        │  Judge   │
        └──────────┘

Example: Decision Support

class DebateSystem:
    def __init__(self):
        self.advocate_pro = AdvocateAgent(position="support")
        self.advocate_con = AdvocateAgent(position="oppose")
        self.judge = JudgeAgent()

    def evaluate_decision(self, decision: str) -> str:
        # Get initial arguments
        pro_args = self.advocate_pro.run(
            f"Argue in favor of: {decision}"
        )
        con_args = self.advocate_con.run(
            f"Argue against: {decision}"
        )

        # Rebuttals
        pro_rebuttal = self.advocate_pro.run(
            f"Respond to these counter-arguments:\n{con_args}"
        )
        con_rebuttal = self.advocate_con.run(
            f"Respond to these supporting arguments:\n{pro_args}"
        )

        # Judge synthesizes
        verdict = self.judge.run(f"""
        Evaluate this decision: {decision}

        Arguments in favor:
        {pro_args}

        Rebuttal:
        {pro_rebuttal}

        Arguments against:
        {con_args}

        Rebuttal:
        {con_rebuttal}

        Provide a balanced assessment and recommendation.
        """)

        return verdict

When to use: Decisions requiring balanced analysis.

Pattern 5: Hierarchical Teams

Nested teams with managers:

                ┌──────────┐
                │ Director │
                └────┬─────┘
                     │
        ┌────────────┼────────────┐
        │            │            │
        ▼            ▼            ▼
   ┌────────┐   ┌────────┐   ┌────────┐
   │Manager │   │Manager │   │Manager │
   │   A    │   │   B    │   │   C    │
   └───┬────┘   └───┬────┘   └───┬────┘
       │            │            │
   ┌───┴───┐    ┌───┴───┐    ┌───┴───┐
   │  │  │ │    │  │  │ │    │  │  │ │
   ▼  ▼  ▼ ▼    ▼  ▼  ▼ ▼    ▼  ▼  ▼ ▼
  Workers      Workers       Workers

Example: Software Development Team

class DevelopmentTeam:
    def __init__(self):
        self.tech_lead = TechLeadAgent()

        self.frontend_team = TeamManager(
            manager=FrontendLeadAgent(),
            workers=[
                ReactDeveloperAgent(),
                UIDeveloperAgent(),
                TestingAgent()
            ]
        )

        self.backend_team = TeamManager(
            manager=BackendLeadAgent(),
            workers=[
                APIDeveloperAgent(),
                DatabaseAgent(),
                SecurityAgent()
            ]
        )

    def develop_feature(self, spec: str) -> dict:
        # Tech lead creates high-level plan
        plan = self.tech_lead.run(f"Create development plan for: {spec}")

        # Frontend and backend work in parallel
        frontend_result = self.frontend_team.execute(
            f"Implement frontend per this plan:\n{plan}"
        )
        backend_result = self.backend_team.execute(
            f"Implement backend per this plan:\n{plan}"
        )

        # Tech lead reviews and integrates
        integration = self.tech_lead.run(f"""
        Review and integrate these implementations:

        Frontend:
        {frontend_result}

        Backend:
        {backend_result}

        Ensure they work together correctly.
        """)

        return {
            "plan": plan,
            "frontend": frontend_result,
            "backend": backend_result,
            "integration": integration
        }


class TeamManager:
    def __init__(self, manager, workers):
        self.manager = manager
        self.workers = workers

    def execute(self, task: str) -> str:
        # Manager breaks down task
        assignments = self.manager.run(
            f"Break down this task for your team:\n{task}"
        )

        # Workers execute in parallel
        results = asyncio.gather(*[
            worker.arun(assignments[i])
            for i, worker in enumerate(self.workers)
        ])

        # Manager integrates
        return self.manager.run(
            f"Integrate these results:\n{results}"
        )

When to use: Complex projects requiring structured organization.

Communication Protocols

Direct Messaging

Agents communicate point-to-point:

class MessageBus:
    def __init__(self):
        self.agents = {}
        self.mailboxes = defaultdict(list)

    def register(self, agent_id: str, agent):
        self.agents[agent_id] = agent

    def send(self, from_id: str, to_id: str, message: str):
        self.mailboxes[to_id].append({
            "from": from_id,
            "message": message,
            "timestamp": time.time()
        })

    def receive(self, agent_id: str) -> list:
        messages = self.mailboxes[agent_id]
        self.mailboxes[agent_id] = []
        return messages

Broadcast Communication

One agent broadcasts to all:

class Broadcaster:
    def broadcast(self, sender: str, message: str, agents: list):
        for agent in agents:
            if agent.id != sender:
                agent.receive(sender, message)

Shared State

Agents read and write to shared context:

class SharedContext:
    def __init__(self):
        self.state = {}
        self.lock = threading.Lock()

    def update(self, agent_id: str, key: str, value: any):
        with self.lock:
            self.state[key] = {
                "value": value,
                "updated_by": agent_id,
                "timestamp": time.time()
            }

    def read(self, key: str) -> any:
        return self.state.get(key, {}).get("value")

    def get_context_for_agent(self, relevant_keys: list) -> dict:
        return {k: self.read(k) for k in relevant_keys}

Coordination Strategies

Token-Based Coordination

Only the agent holding the "token" can act:

class TokenCoordinator:
    def __init__(self, agents: list):
        self.agents = agents
        self.current_holder = 0

    def next_turn(self):
        agent = self.agents[self.current_holder]
        self.current_holder = (self.current_holder + 1) % len(self.agents)
        return agent

    def run_conversation(self, initial_message: str, max_turns: int = 10):
        messages = [initial_message]

        for _ in range(max_turns):
            agent = self.next_turn()
            response = agent.run(
                f"Previous discussion:\n{messages[-3:]}\n\nYour turn to contribute."
            )
            messages.append(f"{agent.name}: {response}")

            if "CONSENSUS REACHED" in response:
                break

        return messages

Voting Systems

Agents vote on decisions:

class VotingSystem:
    def __init__(self, agents: list):
        self.agents = agents

    def vote(self, proposal: str) -> dict:
        votes = {}

        for agent in self.agents:
            response = agent.run(f"""
            Proposal: {proposal}

            Vote YES or NO, with brief reasoning.
            """)

            votes[agent.name] = {
                "vote": "YES" if "YES" in response.upper() else "NO",
                "reasoning": response
            }

        yes_count = sum(1 for v in votes.values() if v["vote"] == "YES")
        passed = yes_count > len(self.agents) / 2

        return {
            "proposal": proposal,
            "votes": votes,
            "result": "PASSED" if passed else "REJECTED",
            "yes_count": yes_count,
            "no_count": len(self.agents) - yes_count
        }

Priority Queues

High-priority agents act first:

import heapq

class PriorityCoordinator:
    def __init__(self):
        self.queue = []

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

    def process_next(self) -> tuple:
        if not self.queue:
            return None
        priority, agent, task = heapq.heappop(self.queue)
        result = agent.run(task)
        return (agent.name, result)

    def process_all(self) -> list:
        results = []
        while self.queue:
            results.append(self.process_next())
        return results

Handling Conflicts

Conflict Detection

Identify when agents disagree:

class ConflictDetector:
    def __init__(self, arbiter):
        self.arbiter = arbiter

    def check_conflict(self, response_a: str, response_b: str) -> dict:
        assessment = self.arbiter.run(f"""
        Compare these two responses and identify conflicts:

        Response A:
        {response_a}

        Response B:
        {response_b}

        Identify:
        1. Are there conflicting claims or recommendations?
        2. What is the nature of the conflict?
        3. Which response is more likely correct and why?
        """)

        return {
            "has_conflict": "conflict" in assessment.lower(),
            "assessment": assessment
        }

Conflict Resolution

Resolve disagreements between agents:

class ConflictResolver:
    def __init__(self, resolver_agent):
        self.resolver = resolver_agent

    def resolve(self, agent_a, agent_b, topic: str) -> str:
        # Get both positions
        position_a = agent_a.run(f"State your position on: {topic}")
        position_b = agent_b.run(f"State your position on: {topic}")

        # Each agent responds to the other
        response_a = agent_a.run(f"Respond to this position: {position_b}")
        response_b = agent_b.run(f"Respond to this position: {position_a}")

        # Resolver makes final call
        resolution = self.resolver.run(f"""
        Resolve this disagreement:

        Agent A's position: {position_a}
        Agent A's response to B: {response_a}

        Agent B's position: {position_b}
        Agent B's response to A: {response_b}

        Make a final decision and explain the reasoning.
        """)

        return resolution

Real-World Implementation

Building a Multi-Agent System

Here's a complete example:

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

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

class Agent:
    def __init__(self, config: AgentConfig):
        self.config = config
        self.client = Anthropic()

    def run(self, message: str) -> str:
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            system=self.config.system_prompt,
            messages=[{"role": "user", "content": message}]
        )
        return response.content[0].text

    async def arun(self, message: str) -> str:
        return await asyncio.to_thread(self.run, message)


class MultiAgentSystem:
    def __init__(self):
        self.agents = {}
        self.coordinator = None

    def add_agent(self, agent: Agent):
        self.agents[agent.config.name] = agent

    def set_coordinator(self, agent: Agent):
        self.coordinator = agent

    async def execute_task(self, task: str) -> Dict[str, Any]:
        # Coordinator plans the approach
        plan = self.coordinator.run(f"""
        Create a plan to accomplish this task using these agents:
        {[a.config.name + ': ' + a.config.role for a in self.agents.values()]}

        Task: {task}

        Output a JSON plan with:
        - steps: list of {{agent, subtask}} assignments
        - parallel: boolean for each step (can it run parallel with others?)
        """)

        # Parse and execute plan
        # (Implementation depends on plan format)

        return {"task": task, "plan": plan, "result": "..."}


# Create a document review team
review_system = MultiAgentSystem()

review_system.add_agent(Agent(AgentConfig(
    name="content_reviewer",
    role="Reviews content accuracy and completeness",
    system_prompt="You review documents for factual accuracy..."
)))

review_system.add_agent(Agent(AgentConfig(
    name="style_reviewer",
    role="Reviews writing style and clarity",
    system_prompt="You review documents for clear, engaging writing..."
)))

review_system.add_agent(Agent(AgentConfig(
    name="legal_reviewer",
    role="Reviews for legal and compliance issues",
    system_prompt="You review documents for legal risks..."
)))

review_system.set_coordinator(Agent(AgentConfig(
    name="coordinator",
    role="Coordinates the review process",
    system_prompt="You coordinate document reviews across specialists..."
)))

# Run a review
result = asyncio.run(review_system.execute_task(
    "Review this contract for our new vendor partnership..."
))

Best Practices

1. Clear Agent Boundaries

Each agent should have:

• A single, well-defined responsibility
• Clear inputs and outputs
• Minimal overlap with other agents

2. Explicit Communication

Don't rely on implicit understanding:

• Use structured message formats
• Include context with each communication
• Log all inter-agent messages

3. Failure Handling

Design for individual agent failure:

• Timeouts for non-responsive agents
• Fallback agents for critical functions
• Graceful degradation

4. Observability

Monitor the system:

• Log all agent actions
• Track message flows
• Measure latency at each stage

5. Testing

Test at multiple levels:

• Individual agent behavior
• Pairwise interactions
• Full system integration

Conclusion

Multi-agent systems unlock capabilities beyond what single agents can achieve. By combining specialization with coordination, you can build systems that tackle complex, multi-faceted problems effectively.

Key principles:

1. Choose the right cooperation pattern for your task
2. Design clear communication protocols
3. Handle conflicts explicitly
4. Build for observability and debugging
5. Test interactions, not just individuals

The future of AI is not one super-agent—it's teams of specialized agents working together.

---

Want to ensure your agent teams stay on track? Check out Guardrails for AI Agents for safety patterns in multi-agent systems.