Most apps that call an LLM send every request to the same model. That works until the bill arrives. A classification job that returns one word costs the same as a multi-step reasoning job, because you picked one model at design time and never revisited it. The fix is an orchestrator-worker setup: a small central router reads the incoming task, decides which tier of model it needs, and delegates to a worker. The router sits at the hub, each model tier is a spoke, and no worker talks to another. This post builds that in Python using both the Anthropic and OpenAI SDKs, so you can mix providers in the same pipeline.
1. Define the tiers before writing any code
Three tiers cover almost everything. Fast handles extraction, classification, tagging, and format conversion. Balanced handles summarization, SQL generation, and short analysis. Deep handles multi-step reasoning, code review, and anything where a wrong answer is expensive. Map each tier to a model from each provider so you can fail over between them.
# tiers.py TIERS = { "fast": {"anthropic": "claude-haiku-4-5", "openai": "gpt-5.6-luna"}, "balanced": {"anthropic": "claude-sonnet-5", "openai": "gpt-5.6-terra"}, "deep": {"anthropic": "claude-opus-4-8", "openai": "gpt-5.6-sol"}, }
2. The orchestrator itself belongs on the cheap tier
The orchestrator runs on every request, before any real work happens. Whatever it costs gets multiplied across your entire traffic, and its latency lands in front of every response. Putting a deep model there means paying premium prices to decide what to do. Routing is short classification against a closed label set, which is precisely what the small models handle well.
The risk is that router errors are asymmetric. Sending a fast task to the deep tier wastes money and you see it in the bill. Sending a deep task to the fast tier produces a plausible wrong answer that flows downstream unnoticed. Three controls keep that in check: a closed label set instead of free-form output, a hard cap on max_tokens so the router cannot drift into explanation, and the validator gate covered further down, which catches the cases where the cheap tier was not enough.
import anthropic client = anthropic.Anthropic() ROUTER_PROMPT = """Classify the task into exactly one tier. fast: extraction, classification, tagging, formatting balanced: summarization, query generation, short analysis deep: multi-step reasoning, code review, planning Answer with the tier name only.""" def route(task_text): msg = client.messages.create( model="claude-haiku-4-5", max_tokens=8, system=ROUTER_PROMPT, messages=[{"role": "user", "content": task_text}], ) tier = msg.content[0].text.strip().lower() return tier if tier in TIERS else "balanced"
The fallback on the last line matters. Routers misfire, and an unrecognized label should land on a safe middle tier rather than raise. And when your app already knows the task type because the user clicked a specific button or the pipeline stage is fixed, skip the router entirely. A dictionary lookup costs nothing and never misclassifies.
3. Wrap each worker behind one function signature
The two SDKs differ in shape. Anthropic uses messages.create with a separate system argument and a required max_tokens. OpenAI uses responses.create with an input list and a reasoning effort setting. Normalize both to take a prompt and return a string, and the orchestrator stops caring which vendor answered. The Anthropic Python SDK and the OpenAI model catalog document the current identifiers for each family.
import anthropic from openai import OpenAI anthropic_client = anthropic.Anthropic() openai_client = OpenAI() def call_anthropic(model, system, prompt, max_tokens=1024): msg = anthropic_client.messages.create( model=model, max_tokens=max_tokens, system=system, messages=[{"role": "user", "content": prompt}], ) return msg.content[0].text def call_openai(model, system, prompt, effort="low"): resp = openai_client.responses.create( model=model, instructions=system, input=prompt, reasoning={"effort": effort}, ) return resp.output_text
4. Dispatch, and allow one escalation
The orchestrator is now small. Pick the tier, pick the provider, run the call. Add a single escalation path: if the fast tier returns something your validator rejects, retry once on the next tier up. That ladder is a model cascade, and it should stay short. One retry, not a loop. Unbounded escalation turns a cost optimization into a cost multiplier.
LADDER = ["fast", "balanced", "deep"] def run_task(prompt, system, validator, provider="anthropic"): tier = route(prompt) idx = LADDER.index(tier) for attempt_tier in LADDER[idx:idx + 2]: model = TIERS[attempt_tier][provider] if provider == "anthropic": out = call_anthropic(model, system, prompt) else: effort = "low" if attempt_tier == "fast" else "high" out = call_openai(model, system, prompt, effort) if validator(out): return {"tier": attempt_tier, "model": model, "output": out} raise ValueError("no tier produced a valid result")
The validator is the piece people skip, and it is what makes the whole design safe. For extraction, check that the JSON parses and the required keys exist. For SQL generation, run EXPLAIN against the target database. Without a validator you have no signal that the cheap tier was good enough, and the routing decision becomes a guess.
5. Log the tier, not just the answer
Store the tier, model, token counts, and validator result for every call. After a week you will see which task types actually escalate and which ones you over-provisioned from the start. That log is the only honest way to tune the routing prompt, and it usually reveals that a large share of traffic never needed anything above the fast tier.
6. A real pipeline: invoice ingestion
Consider a service that receives scanned invoices and has to produce clean rows in a database. Four distinct jobs happen per document, and only one of them is hard. Classifying the document type is a label from a short list. Field extraction is bounded and schema-driven. Totals reconciliation is arithmetic plus judgment about rounding and tax lines. Resolving an ambiguous line item against a product catalog is where reasoning actually pays for itself. Hardcode the first three, and let the orchestrator decide only on the last one.
# pipeline.py STAGE_TIERS = { "classify_doc": "fast", "extract_fields": "fast", "reconcile": "balanced", } def run_stage(stage, system, prompt, validator, provider="anthropic"): # known stage, no router call needed tier = STAGE_TIERS[stage] model = TIERS[tier][provider] ... # same call plus escalation logic as run_task def process_invoice(ocr_text): doc_type = run_stage( "classify_doc", system="Answer with one of: invoice, receipt, credit_note, other.", prompt=ocr_text[:2000], validator=is_known_doc_type, ) if doc_type != "invoice": return route_to_manual_queue(ocr_text, doc_type) fields = run_stage( "extract_fields", system=EXTRACTION_SCHEMA_PROMPT, prompt=ocr_text, validator=parses_as_invoice_json, ) if not totals_match(fields): fields = run_stage( "reconcile", system=RECONCILE_PROMPT, prompt=describe_mismatch(fields), validator=totals_match, ) # only unresolved items reach the orchestrator and the expensive tiers for item in unmatched_items(fields): item["sku"] = run_task( prompt=catalog_match_prompt(item), system=CATALOG_PROMPT, validator=sku_exists, )["output"] return persist(fields)
Notice what the structure buys you. Every invoice pays for two fast calls. The balanced tier only runs when the arithmetic disagrees, and the deep tier only touches line items the catalog lookup could not resolve on its own. On a clean batch, most documents never leave the cheap tier, and the expensive models are reserved for the handful of cases that genuinely earn them.
Wrapping Up
Orchestrator-worker does not have to mean an elaborate framework. A tier table, a cheap orchestrator, one wrapper per SDK, and a validator gate get you most of the benefit in under a hundred lines. Both providers expose a low-cost tier that handles far more than people expect once the prompt is narrow enough.
One practical note before you ship: pin the versioned model identifiers in production rather than the family aliases. Aliases move when a vendor releases a new generation, and a routing table that silently changes tiers under you is a debugging session nobody enjoys.
