How to design a supervisor agent for multi-agent systems

Good supervisor agent design separates reliable multi-agent systems from brittle ones. The supervisor must delegate tasks, aggregate results, and recover from worker failures without becoming a sequential bottleneck or a hidden single point of failure.

Step 1: Define the supervisor’s contract

A supervisor is not a god object that knows every domain detail. It needs a strict interface: what it accepts from the user, which workers it may call, and the shape of its final answer. Write this contract as a JSON Schema or a Pydantic model before you write any orchestration logic.

from pydantic import BaseModel, Field

class WorkerTask(BaseModel):
    worker: str = Field(..., description="registered worker id")
    prompt: str
    timeout_ms: int = 30_000

class SupervisorPlan(BaseModel):
    tasks: list[WorkerTask]
    aggregation: str = Field(..., description="how to merge outputs")

The contract forces you to decide what the supervisor actually decides versus what workers execute. Keep the supervisor LLM context limited to planning, not execution traces. If you let the supervisor read full worker outputs mid-plan, you will blow the context window and slow every step. Good supervisor agent design treats the planner as a thin layer.

Step 2: Choose a messaging and state model

Use explicit state, not implicit conversation history. A dictionary backed by Redis or a SQLite row per run prevents context overflow and makes retries idempotent. Each run gets a run_id; workers are pure functions of (task, run_id) returning structured output.

import uuid

class RunState:
    def __init__(self):
        self.run_id = uuid.uuid4().hex
        self.pending = {}
        self.completed = {}
        self.errors = {}

Pass run_id to every worker call. If a worker crashes, the supervisor can replay only the missing tasks. This model also makes supervisor agent design testable: you can freeze state and assert the next action.

Pick a concurrency primitive early. Python’s asyncio.Semaphore caps in-flight workers so a plan of 50 tasks does not open 50 connections at once.

sem = asyncio.Semaphore(10)

async def bounded_call(task):
    async with sem:
        return await execute_task(task)

Avoid storing large binary artifacts in the run state; reference them by URI. The supervisor should never pickle a 10 MB PDF into Redis. Keep state under 1 MB per run to keep retries cheap.

Step 3: Implement delegation with explicit tool schemas

Expose workers to the supervisor LLM as tools with strict schemas. The model proposes which tools to call; the runtime enforces concurrency, timeouts, and schema validation.

tools = [
  {
    "type": "function",
    "function": {
      "name": "call_research_worker",
      "description": "Fetch and summarize web sources",
      "parameters": WorkerTask.model_json_schema()
    }
  }
]

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "system", "content": "Plan tasks using tools."},
              {"role": "user", "content": user_goal}],
    tools=tools,
    tool_choice="auto"
)

Parse tool_calls into WorkerTask objects. Always validate the model’s tool arguments with Pydantic before execution. A malformed timeout_ms of -5 should be caught in the supervisor, not at the worker. Do not trust the model to sequence tasks; your code spawns them with asyncio.gather under the semaphore. If the plan returns a single task, still route it through the same path—special cases become bugs.

Step 4: Add fallback and retry logic

Workers fail. The supervisor should retry transient errors with backoff and re-plan on persistent ones. If you route worker calls through an OpenAI-compatible gateway such as n4n.ai, automatic fallback when a provider is rate-limited or degraded removes the need to code provider-specific retries in the supervisor.

async def execute_task(task: WorkerTask):
    for attempt in range(3):
        try:
            return await worker_call(task)
        except TransientError:
            await asyncio.sleep(2 ** attempt)
    raise WorkerFailed(task.worker)

For non-transient failures, the supervisor should call a secondary worker or return a degraded status. Implement a circuit breaker so a worker that fails repeatedly is skipped for the rest of the run.

class CircuitBreaker:
    def __init__(self, threshold=5):
        self.failures = 0
        self.threshold = threshold
    def record_failure(self):
        self.failures += 1
    def open(self):
        return self.failures >= self.threshold

Supervisor agent design that ignores partial failure will produce hangs, not answers.

Step 5: Aggregate and validate worker outputs

Aggregation is a function, not a prompt. Write explicit merge code that validates schemas and detects conflicts before any LLM sees the data.

def aggregate(results: dict[str, dict]) -> dict:
    if not results:
        return {"status": "empty"}
    merged = {}
    for w, out in results.items():
        if "error" in out:
            continue
        merged.update(out.get("data", {}))
    return {"status": "ok", "data": merged}

If workers return contradictory facts—say one reports a price of $10 and another $12—reconcile with a fixed-model call constrained by a validation schema. Keep the reconciliation prompt narrow: “Given these two JSON blobs, output the one with the newer timestamp.” Do not ask the LLM to “summarize everything.”

Validating outputs against the contract from Step 1 catches silent worker drift. A worker that returns {"text": "..."} instead of {"data": {...}} should be logged and excluded, not passed upstream.

Step 6: Instrument and meter

Per-token usage metering matters when workers call different models. Capture usage from each completion and write it to your state object.

state.completed[task.worker] = {
    "output": result,
    "tokens": completion.usage.total_tokens
}

If you use a gateway that provides per-token usage metering across 240+ models from one endpoint, your supervisor can switch models based on cost or latency without changing client code. For example, n4n.ai offers one OpenAI-compatible endpoint that addresses 240+ models and honors client routing directives, so the supervisor can pin a worker to a specific model family without custom code. This keeps supervisor agent design decoupled from provider quirks and lets you enforce cache-control hints at the gateway level.

Emit structured logs with run_id, worker, latency_ms, and tokens. Wire traces with OpenTelemetry so a slow run shows which worker burned the time. A supervisor without metrics is undebuggable in production.

Step 7: Test with simulated workers

Before live runs, replace workers with fakes that return canned responses or random failures. Verify the supervisor retries, re-plans, and aggregates correctly.

class FakeWorker:
    def __init__(self, fail_rate=0.2):
        self.fail_rate = fail_rate
    async def call(self, task):
        if random.random() < self.fail_rate:
            raise TransientError()
        return {"data": {"echo": task.prompt}}

Run 100 simulations. Assert that run success rate exceeds 95% and that no run hangs past its timeout. Property-based tests that generate random task graphs catch planning edge cases your happy-path prompt never hits.

Verify success

A correct supervisor meets three criteria: (1) given healthy workers, it returns a merged result matching the output schema; (2) with one worker permanently down, it still returns a partial result or explicit degraded status; (3) token counts in state match the sum of worker calls. Run the simulation suite in CI and add a live smoke test that exercises one real worker per critical type.

Good supervisor agent design is mostly boring engineering: contracts, state, retries, and tests. The LLM planning step is the smallest part of the system, and the teams that ship stable multi-agent products are the ones who treated the supervisor like a distributed systems problem, not a prompt engineering exercise.