Supervisor-Worker Agent Pattern for Cost Optimization

Before you start

• Familiarity with the tool-use pattern
• API access to at least two model tiers (frontier + cheap)
• Python 3.10+, async experience helpful

Written in April 2026. Newer models have since shipped, so check current model names and prices before you copy them.

The economics

Frontier models (GPT-5, Claude Opus 4) cost many times more per token than cheap models (Haiku 4.5, Gemini Flash, GPT-4o-mini). Most agent steps don't need frontier quality -- summarization, extraction, formatting, classification all run fine on cheap models. If the frontier plans and dispatches, the cheap models execute, you pay frontier rates only for the hard thinking.

Step 1: Define the supervisor interface

Supervisor receives a goal, emits a list of sub-tasks with the right model for each.

from typing import Literal
from pydantic import BaseModel

class SubTask(BaseModel):
    id: str
    tool: Literal["summarize", "extract_entities", "classify", "draft_response", "final_synthesis"]
    input: str
    model_tier: Literal["cheap", "mid", "frontier"]
    rationale: str

class Plan(BaseModel):
    tasks: list[SubTask]

SUPERVISOR_PROMPT = """
You plan agent workflows. Given a user goal, emit a list of sub-tasks.

Available tools and their cost tiers:
- summarize (cheap): collapse text to N sentences
- extract_entities (cheap): pull structured facts
- classify (cheap): assign a label
- draft_response (mid): write a candidate response
- final_synthesis (frontier): combine, judge, polish

Constraints:
- Use the cheapest tier that plausibly works
- Reserve frontier for steps requiring judgment or high-stakes output
- Reserve mid for drafting
- Default to cheap for extraction/classification/summarization
"""

Step 2: Worker pool keyed by tier

WORKERS = {
  "cheap":    "claude-haiku-4-5",
  "mid":      "claude-sonnet-4-6",
  "frontier": "claude-opus-4-6",
}

async def run_worker(task: SubTask) -> dict:
    model = WORKERS[task.model_tier]
    result = await call_llm(
        model=model,
        prompt=TOOL_PROMPTS[task.tool].format(input=task.input),
    )
    return {"task_id": task.id, "output": result, "tokens": result.usage}

Step 3: Cost accounting

PRICING = {...}  # from the token-budget playbook

def cost(model: str, input_tokens: int, output_tokens: int) -> float:
    p = PRICING[model]
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

class RunLedger:
    def __init__(self):
        self.entries = []
    def log(self, step: str, model: str, usage):
        self.entries.append({
            "step": step,
            "model": model,
            "input_tokens": usage.input_tokens,
            "output_tokens": usage.output_tokens,
            "cost": cost(model, usage.input_tokens, usage.output_tokens),
        })
    def total(self) -> float:
        return sum(e["cost"] for e in self.entries)
    def report(self):
        return {
            "total_usd": self.total(),
            "by_step": self.entries,
            "frontier_cost": sum(e["cost"] for e in self.entries
                                 if "opus" in e["model"].lower()),
            "cheap_cost": sum(e["cost"] for e in self.entries
                              if "haiku" in e["model"].lower()),
        }

Step 4: Orchestrate

import asyncio

async def orchestrate(goal: str) -> tuple[str, dict]:
    ledger = RunLedger()

    # Supervisor
    plan_resp = await call_llm(
        model=WORKERS["frontier"],
        prompt=SUPERVISOR_PROMPT + f"\nGoal: {goal}",
        response_format=Plan,
    )
    ledger.log("supervisor", WORKERS["frontier"], plan_resp.usage)
    plan: Plan = plan_resp.parsed

    # Workers (run parallelizable tasks concurrently)
    worker_results = await asyncio.gather(*[run_worker(t) for t in plan.tasks])
    for t, r in zip(plan.tasks, worker_results):
        ledger.log(f"worker:{t.tool}", WORKERS[t.model_tier], r["tokens"])

    # Final synthesis on frontier
    synthesis = await call_llm(
        model=WORKERS["frontier"],
        prompt=f"Goal: {goal}\nSub-results: {worker_results}\nSynthesize a final answer.",
    )
    ledger.log("final_synthesis", WORKERS["frontier"], synthesis.usage)

    return synthesis.text, ledger.report()

Step 5: Fallback logic

Workers on cheap models sometimes produce garbage on ambiguous inputs. Detect by validating worker output; on failure, retry on mid-tier once, then frontier.

async def run_worker_with_fallback(task: SubTask, validator) -> dict:
    for tier in ["cheap", "mid", "frontier"]:
        task.model_tier = tier
        r = await run_worker(task)
        if validator(r["output"]):
            return r
    raise RuntimeError(f"Task {task.id} failed on all tiers")

Expected savings

• Pure frontier baseline: $X per run
• Supervisor/worker with fallback: a fraction of X per run; measure your own baseline
• Latency: often FASTER despite more calls, because workers run concurrently

Failure modes

• Supervisor hallucinates tool names: use structured output (Pydantic schema) to enforce valid tools
• Worker output drift: cheap models drift on long instructions. Keep worker prompts short and narrow.
• Fallback cost spiral: if 50% of tasks need to fall back to frontier, you've picked wrong worker for the task. Reclassify.

Variations

• Model routing matrix: expand beyond 3 tiers to vendor-specific best picks per tool
• Cached supervisor plans: the same user goal often produces the same plan -- cache and replay
• Human review on frontier spend: flag any run where frontier cost > $X for human review