# 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
```python
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
| Scenario | Example |
|---|---|
| Clear task decomposition | Software development (plan → code → test) |
| Known agent capabilities | Specialized agents (researcher, coder, tester) |
| Need central oversight | Compliance, quality gates |
| Deterministic workflow | ETL pipelines, report generation |

---

## 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
```python
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
| Scenario | Example |
|---|---|
| Deep task hierarchies | Large software projects (epic → story → task) |
| Specialized teams | Marketing → Content → SEO, Social, Email |
| Clear authority lines | Compliance, approval chains |
| Recursive decomposition | Research → Sub-topics → Papers → Synthesis |

---

## 3. Blackboard Pattern

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

### Implementation
```python
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
| Scenario | Example |
|---|---|
| Loosely coupled agents | Heterogeneous agents, different frameworks |
| Opportunistic processing | Any agent can pick up any task |
| Shared state | Collaborative writing, design |
| Asynchronous coordination | Long-running, human-in-loop |

---

## 4. Swarm / Peer-to-Peer

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

### Implementation
```python
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
| Behavior | Mechanism |
|---|---|
| **Load balancing** | Agents claim tasks based on capacity |
| **Specialization** | Agents gravitate to strengths |
| **Redundancy** | Multiple agents attempt critical tasks |
| **Self-organization** | Roles emerge from interaction |

### When to Use
| Scenario | Example |
|---|---|
| No central authority | Distributed sensor networks |
| Dynamic environments | Robot swarms, game AI |
| Fault tolerance | If one agent fails, others continue |
| Emergent optimization | Resource allocation, routing |

---

## 4. Marketplace / Auction

### Pattern
```
Task Posted → Agents Bid → Auctioneer Awards → Winner Executes
```

### Implementation
```python
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
| Type | Description | Use Case |
|---|---|---|
| **English** | Ascending price, open | Resource allocation |
| **Dutch** | Descending price, open | Urgent tasks |
| **Sealed-bid** | Hidden bids, single round | Sensitive pricing |
| **Vickrey** | Sealed, winner pays 2nd price | Truthful bidding |

### When to Use
| Scenario | Example |
|---|---|
| Competing resources | Compute allocation, bandwidth |
| Cost optimization | Choose cheapest provider |
| Dynamic pricing | Spot instances, surge pricing |
| Multi-agent negotiation | Contract negotiation |

---

## 5. Consensus / Voting

### Pattern
```
Proposal → Discussion → Vote → Consensus → Execute
```

### Implementation
```python
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
| Rule | Threshold | Use Case |
|---|---|---|
| **Simple majority** | > 50% | Low-stakes |
| **Supermajority** | > 66% | Important decisions |
| **Unanimous** | 100% | Safety-critical |
| **Weighted** | By stake/expertise | Expert panels |
| **Quorum** | Minimum participation | Governance |

---

## 6. Stigmergy (Environment-Mediated)

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

### Implementation
```python
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
```python
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
```python
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
```python
# 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

| Factor | Supervisor | Hierarchical | Blackboard | Swarm | Marketplace | Consensus |
|---|---|---|---|---|---|---|
| **Central Control** | High | High | Low | None | Medium | Distributed |
| **Scalability** | Low | Medium | High | High | High | Low |
| **Fault Tolerance** | Low | Medium | High | High | High | Medium |
| **Complexity** | Low | Medium | Medium | High | High | Medium |
| **Best For** | Clear workflows | Deep hierarchies | Loose coupling | Dynamic env | Resource allocation | Decisions |

---

## Related Notes

- [[MultiAgent/Communication-Protocols.md]] — Messages for coordination
- [[MultiAgent/Negotiation-Consensus.md]] — Consensus, voting
- [[MultiAgent/Delegation-Handoff.md]] — Delegation patterns
- [[MultiAgent/Emergent-Behavior.md]] — Swarm emergence
- [[MultiAgent/Safety-Coordination.md]] — Safe coordination
- [[Architectures/Control-Flow.md]] — Control flow patterns
- [[Frameworks/LangGraph.md]] — LangGraph subgraphs
- [[Frameworks/AutoGen.md]] — AutoGen GroupChat
- [[Frameworks/CrewAI.md]] — CrewAI processes
- [[Frameworks/OpenAI-Agents-SDK.md]] — Handoffs
- [[Methodologies/Team-Topologies.md]] — Team structures
- [[Frontiers/Agent-Societies.md]] — Large-scale coordination
- [[Frontiers/Research-Directions.md]] — 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.*