Multi-Agent Coordination Patterns

Patterns for orchestrating multiple agents: supervision, swarms, hierarchies, and marketplaces.

Coordination Taxonomy

Coordination
├── Centralized
│   ├── Supervisor/Orchestrator
│   ├── Hierarchical (Manager → Workers)
│   └── Blackboard
├── Decentralized
│   ├── Peer-to-Peer (Swarm)
│   ├── Marketplace/Auction
│   ├── Consensus/Voting
│   └── Stigmergy (Environment-mediated)
└── Hybrid
    ├── Hierarchical + Peer
    └── Supervisor + Swarm

1. Supervisor/Orchestrator

Pattern

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

Implementation

class Orchestrator:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents

    async def run(self, goal: str) -> Result:
        # 1. Plan decomposition
        plan = await self.planner.decompose(goal)

        # 2. Assign tasks
        assignments = self.assign_tasks(plan)

        # 3. Execute with monitoring
        results = await self.execute_parallel(assignments)

        # 4. Aggregate results
        return await self.aggregate(results)

    def assign_tasks(self, plan: Plan) -> dict[str, Task]:
        # Match task requirements to agent capabilities
        assignments = {}
        for task in plan.tasks:
            best_agent = self.select_agent(task)
            assignments[best_agent] = task
        return assignments

    async def execute_parallel(self, assignments: dict[str, Task]) -> dict[str, Result]:
        tasks = [
            self.agents[agent_id].run(task) 
            for agent_id, task in assignments.items()
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return {aid: r for aid, r in zip(assignments.keys(), results)}

When to Use

2. Hierarchical (Manager → Workers)

Pattern

Manager
├── Team Lead A
│   ├── Worker A1
│   ├── Worker A2
│   └── Worker A3
├── Team Lead B
│   ├── Worker B1
│   └── Worker B2
└── Team Lead C
    └── Worker C1

Implementation

class HierarchicalAgent:
    def __init__(self, subordinates: list[Agent], manager_llm: LLM):
        self.subordinates = {a.name: a for a in subordinates}
        self.manager = manager_llm

    async def delegate(self, task: Task) -> Result:
        # Manager decides delegation
        delegation = await self.manager.decide_delegation(
            task=task,
            subordinates=list(self.subordinates.keys())
        )

        # Recursive delegation
        if delegation.type == "single":
            return await self.subordinates[delegation.assignee].run(task)

        elif delegation.type == "parallel":
            subtasks = delegation.subtasks
            results = await asyncio.gather(*[
                self.subordinates[st.assignee].run(st.task)
                for st in subtasks
            ])
            return self.aggregate(results)

        elif delegation.type == "sequential":
            result = None
            for st in delegation.subtasks:
                result = await self.subordinates[st.assignee].run(st.task, context=result)
            return result

When to Use

3. Blackboard Pattern

Pattern

┌─────────────────────────────────────┐
│         Shared Blackboard           │
│  ┌──────┐ ┌──────┐ ┌──────┐        │
│  │ Data │ │Tasks │ │Results│        │
│  └──────┘ └──────┘ └──────┘        │
└─────────────────────────────────────┘
       ▲           ▲           ▲
       │           │           │
  ┌────┴────┐ ┌────┴────┐ ┌────┴────┐
  │ Agent 1 │ │ Agent 2 │ │ Agent 3 │
  └─────────┘ └─────────┘ └─────────┘

Implementation

class Blackboard:
    def __init__(self):
        self.data: dict[str, Any] = {}
        self.tasks: dict[str, Task] = {}
        self.results: dict[str, Result] = {}
        self.subscribers: dict[str, list[asyncio.Queue]] = defaultdict(list)
        self.lock = asyncio.Lock()

    async def write(self, key: str, value: Any, author: str):
        async with self.lock:
            self.data[key] = {"value": value, "author": author, "timestamp": time.time()}
            await self._notify(key, value)

    async def read(self, key: str) -> Any:
        async with self.lock:
            return self.data.get(key, {}).get("value")

    async def post_task(self, task: Task):
        async with self.lock:
            self.tasks[task.id] = task
            await self._notify("tasks", task)

    async def claim_task(self, agent_id: str, task_id: str) -> Task | None:
        async with self.lock:
            task = self.tasks.get(task_id)
            if task and task.status == "pending":
                task.status = "claimed"
                task.assignee = agent_id
                return task
            return None

    async def subscribe(self, agent_id: str, keys: list[str]) -> asyncio.Queue:
        queue = asyncio.Queue()
        for key in keys:
            self.subscribers[key].append(queue)
        return queue

When to Use

4. Swarm / Peer-to-Peer

Pattern

Agent 1 ◄──► Agent 2
  ▲  ▼         ▲  ▼
  │  └─────────┘  │
  ▼               ▼
Agent 3 ◄───────► Agent 4

Implementation

class SwarmAgent:
    def __init__(self, agent_id: str, peers: list[Agent]):
        self.agent_id = agent_id
        self.peers = {p.agent_id: p for p in peers}
        self.local_state = {}

    async def run(self, goal: Goal) -> Result:
        # 1. Local planning
        local_plan = await self.plan_locally(goal)

        # 2. Share intent with peers
        await self.broadcast_intent(local_plan)

        # 3. Receive peer intents
        peer_intents = await self.gather_intents()

        # 4. Resolve conflicts / merge
        merged_plan = self.resolve_conflicts(local_plan, peer_intents)

        # 5. Execute
        return await self.execute(merged_plan)

    async def broadcast_intent(self, plan: Plan):
        for peer in self.peers.values():
            await peer.receive_intent(self.agent_id, plan)

    def resolve_conflicts(self, local: Plan, peer_intents: dict[str, Plan]) -> Plan:
        # Merge strategies: union, priority, voting, negotiation
        all_tasks = local.tasks + sum([p.tasks for p in peer_intents.values()], [])
        # Deduplicate, resolve conflicts
        return self.merge_plans(all_tasks)

Emergent Behaviors

When to Use

4. Marketplace / Auction

Pattern

Task Posted → Agents Bid → Auctioneer Awards → Winner Executes

Implementation

class Marketplace:
    def __init__(self):
        self.tasks: dict[str, AuctionTask] = {}
        self.agents: dict[str, Agent] = {}

    async def post_task(self, task: Task, auction_type: str = "english") -> str:
        auction_id = str(uuid4())
        self.tasks[auction_id] = AuctionTask(
            id=auction_id,
            task=task,
            type=auction_type,
            status="open",
            bids={}
        )
        await self.notify_agents(auction_id)
        return auction_id

    async def submit_bid(self, auction_id: str, agent_id: str, bid: Bid):
        auction = self.tasks[auction_id]
        if auction.status != "open":
            raise AuctionClosed()
        auction.bids[agent_id] = bid

    async def close_auction(self, auction_id: str) -> AuctionResult:
        auction = self.tasks[auction_id]
        auction.status = "closed"

        # Determine winner
        if auction.type == "english":
            winner = max(auction.bids.items(), key=lambda x: x[1].price)
        elif auction.type == "dutch":
            winner = min(auction.bids.items(), key=lambda x: x[1].price)
        elif auction.type == "sealed":
            winner = max(auction.bids.items(), key=lambda x: x[1].score)

        # Award
        winner_agent = self.agents[winner[0]]
        result = await winner_agent.execute(auction.task)

        return AuctionResult(winner=winner[0], bid=winner[1], result=result)

Auction Types

When to Use

5. Consensus / Voting

Pattern

Proposal → Discussion → Vote → Consensus → Execute

Implementation

class ConsensusCoordinator:
    def __init__(self, agents: list[Agent], threshold: float = 0.66):
        self.agents = {a.name: a for a in agents}
        self.threshold = threshold  # Supermajority

    async def propose(self, proposer: str, proposal: Proposal) -> ConsensusResult:
        # 1. Broadcast proposal
        await self.broadcast(proposal)

        # 2. Discussion period (optional)
        await self.discussion_period(proposal)

        # 3. Vote
        votes = await self.collect_votes(proposal)

        # 4. Tally
        yes_votes = sum(1 for v in votes.values() if v == "yes")
        total = len(votes)

        if yes_votes / total >= self.threshold:
            return ConsensusResult(accepted=True, votes=votes)
        else:
            return ConsensusResult(accepted=False, votes=votes)

    async def collect_votes(self, proposal: Proposal) -> dict[str, str]:
        votes = {}
        for agent in self.agents.values():
            vote = await agent.vote(proposal)
            votes[agent.name] = vote
        return votes

Voting Rules

6. Stigmergy (Environment-Mediated)

Pattern

Agent 1 modifies environment ──► Agent 2 senses change ──► Agent 2 acts
                                     │
                                     ▼
                              Environment

Implementation

class StigmergicEnvironment:
    def __init__(self):
        self.pheromones: dict[str, float] = defaultdict(float)  # Virtual pheromones
        self.artifacts: dict[str, Artifact] = {}  # Persistent artifacts
        self.evaporation_rate = 0.1  # Per time unit

    def deposit_pheromone(self, location: str, strength: float, agent_id: str):
        self.pheromones[location] += strength
        self.pheromones[location] = max(0, self.pheromones[location])  # No negative

    def sense_pheromone(self, location: str) -> float:
        return self.pheromones.get(location, 0.0)

    def evaporate(self):
        for loc in self.pheromones:
            self.pheromones[loc] *= (1 - self.evaporation_rate)
            if self.pheromones[loc] < 0.01:
                del self.pheromones[loc]

    def leave_artifact(self, artifact: Artifact, location: str):
        self.artifacts[location] = artifact

    def sense_artifact(self, location: str) -> Artifact | None:
        return self.artifacts.get(location)

Ant Colony Optimization (ACO) for Task Routing

class ACORouter:
    def __init__(self, graph: Graph, alpha=1.0, beta=2.0, evaporation=0.1):
        self.graph = graph
        self.alpha = alpha  # Pheromone importance
        self.beta = beta    # Heuristic importance
        self.evaporation = evaporation
        self.pheromones = defaultdict(float)

    def select_next_node(self, current: str, visited: set) -> str:
        neighbors = [n for n in self.graph.neighbors(current) if n not in visited]
        if not neighbors:
            return None

        # Probability = (pheromone^alpha) * (heuristic^beta)
        probs = []
        for n in neighbors:
            pher = self.pheromones[(current, n)] ** self.alpha
            heur = (1.0 / self.graph.distance(current, n)) ** self.beta
            probs.append((n, pher * heur))

        # Normalize and sample
        total = sum(p for _, p in probs)
        probs = [(n, p/total) for n, p in probs]
        return random.choices([n for n, _ in probs], weights=[p for _, p in probs])[0]

    def update_pheromones(self, path: list[str], quality: float):
        for i in range(len(path) - 1):
            edge = (path[i], path[i+1])
            self.pheromones[edge] = (1 - self.evaporation) * self.pheromones[edge] + quality

Hybrid Patterns

Supervisor + Swarm

class HybridCoordinator:
    def __init__(self):
        self.orchestrator = Orchestrator(agents={})
        self.swarm = SwarmCoordinator(agents={})

    async def run(self, goal: Goal):
        # Orchestrator handles high-level planning
        plan = await self.orchestrator.plan(goal)

        # Decompose into swarm-suitable tasks
        swarm_tasks = self.decompose_for_swarm(plan)

        # Swarm executes
        swarm_results = await self.swarm.execute(swarm_tasks)

        # Orchestrator aggregates
        return await self.orchestrator.aggregate(swarm_results)

Hierarchical + Marketplace

# Manager delegates to teams, teams use internal marketplace
class HybridOrganization:
    def __init__(self):
        self.departments: dict[str, MarketplaceTeam] = {}

    async def execute(self, goal: Goal):
        # Top-level decomposition
        dept_tasks = self.decompose_by_department(goal)

        # Each department runs internal marketplace
        results = await asyncio.gather(*[
            dept.marketplace.execute(task) 
            for dept, task in dept_tasks.items()
        ])

        return self.merge_results(results)

Choosing Coordination Pattern

Related Notes

• Communication Protocols — Messages for coordination
• Negotiation Consensus — Consensus, voting
• Delegation Handoff — Delegation patterns
• Emergent Behavior — Swarm emergence
• Safety Coordination — Safe coordination
• Control Flow — Control flow patterns
• LangGraph — LangGraph subgraphs
• AutoGen — AutoGen GroupChat
• CrewAI — CrewAI processes
• OpenAI Agents SDK — Handoffs
• Team Topologies — Team structures
• Agent Societies — Large-scale coordination
• Research Directions — Coordination research

Sources: "Coordination in Multi-Agent Systems" surveys, "Swarm Intelligence" (Bonabeau et al.), "Market-Based Control" (Wellman et al.), "Stigmergy" (Grassé, 1959), ACO (Dorigo), Blackboard pattern (Corkill), Contract Net Protocol (Smith), FIPA ACL, LangGraph subgraphs, AutoGen GroupChat, CrewAI processes.