Orchestrator-worker patterns for multi-model agent teams

Building multi-model agent teams demands clear separation of planning and execution. The orchestrator worker agent pattern puts a single coordinator in charge of task decomposition while specialized workers run each subtask on the model best suited for it. This guide gives you an ordered path to implement that pattern without coupling your codebase to a specific provider.

Why the orchestrator worker agent pattern wins for multi-model teams

A flat prompt that tries to reason, call tools, and format output in one shot breaks down as soon as you need a different model for each job. The orchestrator worker agent pattern fixes this by splitting responsibilities: the orchestrator thinks, the workers do. You get independent scaling, per-task model selection, and fault isolation.

In a multi-model setup, the orchestrator is typically a strong reasoning model that emits a structured plan. Workers are thin wrappers around cheaper or specialized models—vision, codegen, embedding, or long-context summarization. The pattern also makes cost visible: each worker call is a discrete, metered unit.

Step 1: Define the orchestrator’s contract

Treat the orchestrator as a typed function, not a free-form chat. It accepts a task and returns a list of worker jobs. Use a strict schema so downstream code never guesses.

from pydantic import BaseModel, Field

class WorkerJob(BaseModel):
    job_id: str
    worker_type: str  # "extract", "summarize", "codegen"
    model_hint: str   # e.g. "anthropic/claude-3.5-sonnet"
    payload: dict

class OrchestratorPlan(BaseModel):
    jobs: list[WorkerJob] = Field(default_factory=list)
    final_synthesis_hint: str = "openai/gpt-4o"

The orchestrator itself is just a chat completion with response format forced to OrchestratorPlan. If you skip the schema, you will spend the next month writing parsers for slightly different JSON.

Step 2: Map subtasks to models

Do not hardcode model names inside worker logic. Keep a routing table that the orchestrator can reference via model_hint. This decouples model upgrades from code changes.

{
  "extract": "anthropic/claude-3.5-sonnet",
  "summarize": "google/gemini-flash-1.5",
  "codegen": "openai/gpt-4o",
  "embed": "text-embedding-3-small"
}

The orchestrator worker agent pattern stays clean when the orchestrator only emits hints and a central resolver turns hints into concrete endpoints. That resolver can also apply client routing directives—like “prefer provider X unless degraded”—without touching worker code.

Step 3: Implement worker isolation

Each worker should run as an independent async task with its own timeout and retry budget. Share nothing except the job payload and a result queue.

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(base_url="https://api.your-gateway.com/v1", api_key="sk-...")

async def run_worker(job: WorkerJob, route_table: dict) -> dict:
    model = route_table.get(job.worker_type, job.model_hint)
    try:
        resp = await client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": job.payload["text"]}],
            timeout=30,
        )
        return {"job_id": job.job_id, "ok": True, "text": resp.choices[0].message.content}
    except Exception as e:
        return {"job_id": job.job_id, "ok": False, "error": str(e)}

async def dispatch(plan: OrchestratorPlan, route_table: dict):
    tasks = [run_worker(j, route_table) for j in plan.jobs]
    return await asyncio.gather(*tasks)

Isolation matters because a slow vision model should not block a fast summarizer. Use asyncio.gather with return_exceptions=True if you want partial progress.

Step 4: Add fallback and cache hints

Workers fail. Providers rate-limit. Build explicit fallback into the resolver, and forward provider cache-control hints to cut cost on repeated context.

When you route through a gateway such as n4n.ai, the same OpenAI-compatible endpoint fronts 240+ models and automatically fails over if a provider is rate-limited or degraded. You still need application-level fallback for task-specific errors—like a codegen worker returning invalid syntax.

async def run_worker_with_fallback(job, route_table, fallbacks):
    primary = route_table.get(job.worker_type)
    models = [primary, *fallbacks.get(job.worker_type, [])]
    for m in models:
        try:
            resp = await client.chat.completions.create(
                model=m,
                messages=[{"role": "user", "content": job.payload["text"]}],
                extra_headers={"x-cache-control": "read-write"},  # forward cache hint
                timeout=30,
            )
            return {"job_id": job.job_id, "model": m, "text": resp.choices[0].message.content}
        except Exception:
            continue
    return {"job_id": job.job_id, "ok": False}

Cache hints are ignored by workers that don’t support them, but when honored they turn a 10k-token system prompt into a cached read. That is a real per-token saving.

Step 5: Meter usage and set guards

Every worker call returns token counts. Aggregate them per job type so you can spot runaway loops. The orchestrator worker agent pattern makes this trivial because each subtask is a discrete call.

def log_usage(results, plan):
    for res, job in zip(results, plan.jobs):
        if "usage" in res:
            print(f"{job.worker_type}: {res['usage']['total_tokens']} tok")

Set a hard cap on replanning cycles. If the orchestrator issues more than N plans for one user request, abort. Unbounded orchestration is the most common way to burn budget.

Common pitfalls and tradeoffs

Context fragmentation. Workers only see their payload. If the orchestrator strips too much context, workers produce inconsistent output. Tradeoff: send enough, but not the full 50k-token thread to a cheap model.

Latency multiplication. Sequential orchestrator→worker→synthesis adds round trips. Parallelize independent workers, but accept that the pattern adds at least one extra hop versus a monolith.

Error masking. A worker returning ok:false can be silently dropped by a lazy gather. Always reconcile the result count against the plan; missing jobs mean a partial failure that the synthesizer must handle.

Over-decomposition. New engineers split tasks into ten workers when two would do. Each worker is a network call, a billing event, and a potential failure. The orchestrator worker agent pattern is not a license to microservice your prompt.

Model drift. If you pin model_hint to a specific version and the provider deprecates it, workers break. Use the routing table as the single migration point.

Reference implementation sketch

A minimal service looks like:

1. POST /run → orchestrator generates OrchestratorPlan.
2. Resolver expands worker_type to model + fallback list.
3. asyncio.gather runs workers with cache headers.
4. Synthesizer (using final_synthesis_hint) merges results.
5. Usage middleware emits per-token metrics.

The orchestrator worker agent pattern pays off the moment you need to swap a $0.01/1k-token model for a $0.10 one on a single subtask without rewriting the whole agent. Build the seams first; the models will keep changing.