Chapter 10 · AI Architect

Multi-agent orchestration

coordinator · agent registry · task routing · result mergeARCH-90167 of 84

A real AI system is rarely one giant prompt. It's a coordinator handing sub-tasks to specialist agents — a researcher, a summariser, a formatter — then merging what comes back. Each agent is good at one thing; the coordinator's job is to route and combine.

This is the registry / dispatch pattern you already met, scaled up to agents. A *registry* is just a dictionary from a key to a callable: {"research": research_fn, "summarise": summarise_fn}. Agents register themselves by capability (a task kind); the coordinator looks up the right one and calls it.

Three moving parts:

register — store an agent under its kind in the registry dict.
dispatch — given a task {"kind": ..., "input": ...}, look up the agent for that kind and call it with the input. Unknown kind → raise an error rather than guess.
run_plan — dispatch a list of tasks in order and collect the results (the merge).

In production each agent is a real model call with its own system prompt and tools; here they're plain functions so the pattern is visible without the noise.

Syntax

class Coordinator:
    def __init__(self):
        self.agents = {}                 # registry: kind -> callable

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

    def dispatch(self, task):
        kind = task["kind"]
        if kind not in self.agents:
            raise KeyError(f"no agent for: {kind}")
        return self.agents[kind](task["input"])

    def run_plan(self, tasks):
        return [self.dispatch(t) for t in tasks]

Worked examples

class Coordinator:
    def __init__(self):
        self.agents = {}

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

    def dispatch(self, task):
        kind = task["kind"]
        if kind not in self.agents:
            raise KeyError(f"no agent for: {kind}")
        return self.agents[kind](task["input"])

coord = Coordinator()
coord.register("upper", lambda s: s.upper())
print(coord.dispatch({"kind": "upper", "input": "rag"}))
# Output:
# RAG

run_plan dispatches in order and merges the results

class Coordinator:
    def __init__(self):
        self.agents = {}
    def register(self, kind, agent):
        self.agents[kind] = agent
    def dispatch(self, task):
        return self.agents[task["kind"]](task["input"])
    def run_plan(self, tasks):
        return [self.dispatch(t) for t in tasks]

coord = Coordinator()
coord.register("double", lambda n: n * 2)
coord.register("label", lambda n: f"got:{n}")
plan = [
    {"kind": "double", "input": 21},
    {"kind": "label", "input": "done"},
]
print(coord.run_plan(plan))
# Output:
# [42, 'got:done']

An unknown kind fails loudly instead of silently

class Coordinator:
    def __init__(self):
        self.agents = {}
    def dispatch(self, task):
        kind = task["kind"]
        if kind not in self.agents:
            raise KeyError(f"no agent for: {kind}")
        return self.agents[kind](task["input"])

coord = Coordinator()
try:
    coord.dispatch({"kind": "ghost", "input": 1})
except KeyError as e:
    print("KeyError:", e)
# Output:
# KeyError: "no agent for: ghost"

Common mistakes & gotchas

Routing on a kind that was never registered

If dispatch just did self.agents[kind](...) with no guard, an unknown kind raises a bare KeyError with only the missing key — hard to debug. Check if kind not in self.agents and raise a clear message (no agent for: <kind>) so the failure tells you what's missing.

Assuming agents run in parallel

run_plan here dispatches in order, one after another — task 2 can depend on task 1's output. If you genuinely want parallelism you need a concurrency model (next-but-two entry) or a DAG (next entry). Don't assume a list of tasks runs concurrently just because it's a list.

Putting the routing logic inside the agents

The point of the coordinator is that agents don't know about each other — each just does its job on an input. If an agent starts deciding which other agent to call, you've smeared the routing across the system and lost the single place to reason about flow. Keep routing in the coordinator.

Why it matters

Almost every non-trivial AI product is multi-agent under the hood — a planner that delegates to specialists. Owning the coordinator/registry pattern means you can add a new capability by registering one function, swap a stub for a real model call without touching the flow, and reason about the whole system in one place.