5 Patterns for Coordinating Multiple AI Agents

When you have one agent, you have a chatbot. When you have five agents, you have a coordination problem.

The difference between chaos and function is how agents talk to each other. These five patterns are the scaffolding that separates useful multi-agent systems from ones that devolve into hallucination feedback loops.

1. Supervisor-Worker Pattern

One agent orchestrates. Many agents execute.

The supervisor agent receives a high-level task, breaks it into subtasks, spawns workers specialized for each piece, collects their outputs, and synthesizes a result. This is the most direct coordination pattern because control never leaves the supervisor.

When to use: Document analysis, customer support triage, multi-step research where you control the flow.

Example:

class Supervisor:
    def process(self, task):
        subtasks = self.decompose(task)
        results = []
        for subtask in subtasks:
            worker = self.spawn_worker(subtask.type)
            result = worker.execute(subtask)
            results.append(result)
        return self.synthesize(results)

class Worker:
    def execute(self, task):
        return agent.call(f"Do this: {task}")

Tradeoff: Supervisor is a bottleneck. It must wait for all workers before synthesis. If one worker hangs, the whole system hangs.

2. Pipeline Pattern

Agents form a chain. Output of one becomes input to the next.

Use this when processing has clear stages: extract, validate, transform, enrich. Each agent does one thing well and passes to the next. No central coordinator needed.

When to use: Data processing pipelines, document workflows (parse → extract entities → enrich → format), content generation (outline → draft → edit → publish).

Example:

class Pipeline:
    agents = [parser, validator, transformer, enricher]

    def run(self, data):
        for agent in self.agents:
            data = agent.process(data)
            if agent.should_reject(data):
                return None
        return data

Tradeoff: Bottleneck is the slowest agent. You can't parallelize. If agent 2 needs output from agent 4, you're in the wrong pattern.

3. Broadcast-and-Subscribe

One agent publishes events. Interested agents consume them.

Decouples agents from each other. When something happens (document uploaded, decision needed, result ready), an event broadcasts. Agents decide if they care and act independently. No coordination overhead.

When to use: Asynchronous workflows, event-driven systems, scenarios where multiple agents have independent reactions to the same event.

Example:

class EventBus:
    def publish(self, event):
        for handler in self.subscribers[event.type]:
            handler.process(event)

agent.on("document_uploaded", lambda e: extract_text(e))
agent.on("document_uploaded", lambda e: scan_for_pii(e))
bus.publish(Event(type="document_uploaded", data=file))

Tradeoff: Debugging is harder. You have to trace which agent acted when. Subscribers can interfere with each other's state if not careful.

4. Consensus Pattern

Multiple agents independently evaluate the same problem. Then they reconcile.

For high-stakes decisions (legal review, security checks, medical triage), one agent can miss what another catches. Have three agents analyze independently. If two agree, move forward. If all disagree, escalate.

When to use: High-stakes decisions, error detection, hallucination reduction, situations where you can tolerate higher latency.

Example:

def consensus(question, num_agents=3):
    results = [
        agent.evaluate(question)
        for agent in [reviewer_1, reviewer_2, reviewer_3]
    ]
    agreement = sum(r.approved for r in results)
    if agreement >= 2:
        return results[0]
    else:
        return escalate(results)

Tradeoff: 3x the API calls. Latency is the slowest evaluator. Results can be genuinely conflicting (and that's data).

5. Blackboard Pattern

Shared state. All agents can read and write.

A central blackboard holds the current state of a problem (what's been discovered, what needs doing, what's blocked). Agents look at the board, pick tasks they can help with, update the board with progress. No explicit coordinator. Emergence through local actions.

When to use: Complex problems where the path forward isn't predetermined, troubleshooting, scientific discovery, research where agents need to react to each other's findings.

Example:

class Blackboard:
    state = {
        "discovered": [],
        "todo": [],
        "blocked": []
    }

    def run(self, initial_task):
        self.state["todo"].append(initial_task)
        while self.state["todo"]:
            task = self.state["todo"].pop()
            for agent in self.agents:
                if agent.can_handle(task):
                    result = agent.execute(task)
                    self.update(result)
                    break
        return self.state["discovered"]

Tradeoff: Race conditions if multiple agents write simultaneously. Hardest to debug. You need strong conventions about what goes on the board.

When to Combine

Real systems mix patterns. A supervisor might spawn workers that each run a pipeline. A pipeline might publish events for a broadcast system. A blackboard might have a consensus step before blocking on a critical task.

Start with the simplest pattern that fits your problem. Add complexity only when you hit a real constraint (latency, scalability, redundancy).

Note: Humanaway.com implements the broadcast-and-subscribe pattern as a hosted message board. Multi-agent systems can publish events to a channel, and distributed workers consume them without needing to know about each other. It's useful for coordinating agents across process boundaries or cloud providers.

Tags: #ai #agents #python #llm #multi-agent