You're running a customer support platform that handles everything from "what's your return policy?" to "can you help me debug this Python exception in my API integration?" Routing both questions through GPT-4o costs you a fortune and introduces unnecessary latency. Routing both through a lightweight model produces embarrassing results on the complex ones. What you actually need is a system that makes the right call at runtime — cheaply, correctly, and fast enough that users never notice the machinery behind it.
That system is an LLM router. Not a prompt template. Not a chain. A production-grade routing layer that evaluates each incoming request against a set of configurable policies and dispatches it to the most appropriate model. Done well, routing can cut your inference spend by 40–70% while maintaining or even improving quality on the requests that matter. Done poorly, it introduces subtle failure modes that are harder to debug than just using the wrong model in the first place.
By the end of this lesson you'll have built a complete router from scratch — including complexity classification, cost and latency modeling, a policy engine, fallback logic, and an observability layer. We're going to get into the internals, the edge cases, and the production failure modes that most tutorials gloss over.
What you'll learn:
You should be comfortable with:
asyncio, aiohttp)Before writing a line of code, let's be precise about what a router is and isn't. A router is a decision-making layer that sits between your application and your LLM backends. It receives a request, evaluates it against a policy, and dispatches it to the most appropriate model — potentially with request transformation along the way.
What makes routing hard isn't the dispatch itself. It's the classification problem upstream of dispatch, and the failure handling downstream of it. You need to make a routing decision before you've seen the response, which means you're predicting task complexity from the request alone. You need to do this fast enough that it doesn't negate the latency savings of using a cheaper model. And you need to handle the cases where your prediction is wrong.
Let's establish the taxonomy of signals a router can use:
Static signals — Derived from the request structure without any model inference:
Lightweight inference signals — Derived from a small, fast classification model:
Policy signals — External to the request content:
The smartest routers combine all three. We'll build a system that does exactly that.
Good architecture starts with good data models. Let's define the core types that everything else will build on.
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import time
import uuid
class ModelTier(str, Enum):
MICRO = "micro" # e.g., gpt-4o-mini, claude-haiku
STANDARD = "standard" # e.g., gpt-4o, claude-sonnet
PREMIUM = "premium" # e.g., o1, claude-opus
class TaskComplexity(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
UNKNOWN = "unknown"
@dataclass
class ModelBackend:
model_id: str
tier: ModelTier
provider: str # "openai", "anthropic", "cohere", etc.
cost_per_input_token: float # USD per token
cost_per_output_token: float
p50_latency_ms: float # Empirically measured
p95_latency_ms: float
max_context_tokens: int
supports_functions: bool = False
supports_vision: bool = False
weight: float = 1.0 # For load balancing within tier
@dataclass
class RoutingRequest:
request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
messages: list[dict] = field(default_factory=list)
user_id: Optional[str] = None
tenant_id: Optional[str] = None
# Constraints the caller can express
max_cost_usd: Optional[float] = None
max_latency_ms: Optional[float] = None
required_capabilities: list[str] = field(default_factory=list)
preferred_tier: Optional[ModelTier] = None
# Computed during routing
estimated_input_tokens: int = 0
complexity: TaskComplexity = TaskComplexity.UNKNOWN
complexity_confidence: float = 0.0
created_at: float = field(default_factory=time.time)
@dataclass
class RoutingDecision:
request_id: str
selected_backend: ModelBackend
routing_reason: str
fallback_chain: list[ModelBackend] = field(default_factory=list)
estimated_cost_usd: float = 0.0
estimated_latency_ms: float = 0.0
complexity_used: TaskComplexity = TaskComplexity.UNKNOWN
decision_latency_ms: float = 0.0
Notice that RoutingDecision carries a fallback_chain. This is critical — you don't just pick a winner, you pre-compute the entire fallback sequence at decision time so that if the primary backend fails, the fallback logic doesn't need to re-run classification. We'll come back to why this matters under load.
The classifier is the heart of the router and the place where most implementations cut corners. The naive approach is to use a few heuristics — prompt length, keyword presence — and call it done. This works until it doesn't. A three-sentence prompt asking for a multi-step logical proof is "short," but it's also "hard."
We'll use a two-stage approach: fast heuristics that catch obvious cases, followed by a lightweight embedding-based classifier for the ambiguous middle.
import re
import tiktoken
from typing import NamedTuple
class ComplexitySignals(NamedTuple):
token_count: int
has_code_block: bool
has_structured_output_request: bool
conversation_turns: int
reasoning_keyword_count: int
domain_specificity_score: float
REASONING_KEYWORDS = {
"analyze", "compare", "evaluate", "synthesize", "derive", "prove",
"explain why", "step by step", "reason through", "debug", "refactor",
"architect", "optimize", "trade-off", "implications"
}
DOMAIN_SPECIFIC_PATTERNS = [
r'\b(def |class |import |async |await )\b', # Code patterns
r'\b(\d+\.\d+\.\d+)\b', # Version numbers
r'\b(API|SDK|OAuth|JWT|CORS|REST|gRPC)\b', # Technical acronyms
r'\$[A-Z_]{2,}', # Environment variables
r'```[\w]*\n', # Code fences
]
def extract_complexity_signals(messages: list[dict]) -> ComplexitySignals:
enc = tiktoken.get_encoding("cl100k_base")
full_text = " ".join(
m.get("content", "") for m in messages if isinstance(m.get("content"), str)
)
tokens = enc.encode(full_text)
token_count = len(tokens)
text_lower = full_text.lower()
reasoning_count = sum(
1 for kw in REASONING_KEYWORDS if kw in text_lower
)
domain_score = 0.0
for pattern in DOMAIN_SPECIFIC_PATTERNS:
matches = re.findall(pattern, full_text)
domain_score += len(matches) * 0.1
domain_score = min(domain_score, 1.0)
has_code = "```" in full_text or bool(re.search(r'def |class |import ', full_text))
has_structured = bool(re.search(
r'(json|yaml|csv|table|schema|format it as|return a list)',
text_lower
))
turns = sum(1 for m in messages if m.get("role") == "user")
return ComplexitySignals(
token_count=token_count,
has_code_block=has_code,
has_structured_output_request=has_structured,
conversation_turns=turns,
reasoning_keyword_count=reasoning_count,
domain_specificity_score=domain_score,
)
def heuristic_complexity(signals: ComplexitySignals) -> tuple[TaskComplexity, float]:
"""
Returns (complexity, confidence). High confidence means skip the embedding
classifier and use this result directly.
"""
# Clear low-complexity cases — high confidence
if (signals.token_count < 150
and not signals.has_code_block
and signals.reasoning_keyword_count == 0
and signals.domain_specificity_score < 0.2
and signals.conversation_turns <= 2):
return TaskComplexity.LOW, 0.92
# Clear high-complexity cases — high confidence
if (signals.token_count > 2000
or (signals.has_code_block and signals.reasoning_keyword_count >= 2)
or signals.reasoning_keyword_count >= 4):
return TaskComplexity.HIGH, 0.88
# Structured output with moderate complexity
if signals.has_structured_output_request and signals.token_count > 400:
return TaskComplexity.MEDIUM, 0.75
# Ambiguous — return low confidence to trigger embedding classifier
return TaskComplexity.MEDIUM, 0.45
The confidence score is doing real work here. When confidence is above a threshold (we'll use 0.80), we skip the embedding classifier entirely. This is what keeps routing latency in the single-digit milliseconds for the majority of requests.
For ambiguous cases, we use a small embedding model and a pre-labeled reference set. The idea is simple: embed the incoming prompt, find the nearest neighbors from our labeled examples, and vote on complexity.
import numpy as np
from openai import AsyncOpenAI
class EmbeddingClassifier:
def __init__(self, openai_client: AsyncOpenAI):
self.client = openai_client
self.reference_embeddings: np.ndarray | None = None
self.reference_labels: list[TaskComplexity] = []
self.embedding_model = "text-embedding-3-small" # Fast, cheap, good enough
async def load_reference_set(self, examples: list[tuple[str, TaskComplexity]]):
"""
examples: list of (prompt_text, complexity_label) tuples.
Build this from your production logs — label ~200 examples per tier.
"""
texts = [ex[0] for ex in examples]
self.reference_labels = [ex[1] for ex in examples]
# Batch embed — embedding models are fast, ~50ms for 200 texts
response = await self.client.embeddings.create(
model=self.embedding_model,
input=texts
)
self.reference_embeddings = np.array(
[item.embedding for item in response.data]
)
# L2-normalize for cosine similarity via dot product
norms = np.linalg.norm(self.reference_embeddings, axis=1, keepdims=True)
self.reference_embeddings = self.reference_embeddings / norms
async def classify(
self,
messages: list[dict],
top_k: int = 7
) -> tuple[TaskComplexity, float]:
# Extract the last user message for embedding — full context is overkill here
query_text = next(
(m["content"] for m in reversed(messages) if m.get("role") == "user"),
""
)
response = await self.client.embeddings.create(
model=self.embedding_model,
input=[query_text]
)
query_vec = np.array(response.data[0].embedding)
query_vec = query_vec / np.linalg.norm(query_vec)
# Cosine similarities via dot product (vectors are normalized)
similarities = self.reference_embeddings @ query_vec
top_indices = np.argsort(similarities)[-top_k:][::-1]
# Weighted vote by similarity score
votes: dict[TaskComplexity, float] = {}
for idx in top_indices:
label = self.reference_labels[idx]
score = float(similarities[idx])
votes[label] = votes.get(label, 0.0) + score
best_label = max(votes, key=lambda k: votes[k])
total_score = sum(votes.values())
confidence = votes[best_label] / total_score if total_score > 0 else 0.5
return best_label, confidence
Production tip: Don't embed the entire conversation. Embedding only the last user turn is usually sufficient for complexity classification and cuts embedding latency by 60–80% on long conversations. The exception is when the complexity is inherently multi-turn (e.g., "given everything we've discussed, write a formal proposal") — detect this with a simple pattern match on the latest message.
With complexity classified, we need to translate that into a model selection decision. This is where cost and latency constraints enter the picture.
from pydantic import BaseModel
from typing import Callable
class RoutingPolicy(BaseModel):
name: str
# Complexity → tier mapping
complexity_tier_map: dict[str, str] = {
"low": "micro",
"medium": "standard",
"high": "premium",
"unknown": "standard", # Safe default
}
# Hard cost/latency gates
max_cost_usd_override: Optional[float] = None
max_latency_ms_override: Optional[float] = None
# Classifier confidence threshold — below this, upgrade one tier
min_confidence_to_downgrade: float = 0.75
# Whether to respect caller-specified constraints
allow_caller_cost_override: bool = True
allow_caller_latency_override: bool = True
class PolicyEngine:
def __init__(
self,
backends: list[ModelBackend],
default_policy: RoutingPolicy,
tenant_policies: dict[str, RoutingPolicy] | None = None
):
self.backends = backends
self.default_policy = default_policy
self.tenant_policies = tenant_policies or {}
# Pre-index backends by tier for O(1) lookup
self._tier_index: dict[ModelTier, list[ModelBackend]] = {}
for b in backends:
self._tier_index.setdefault(b.tier, []).append(b)
def _get_policy(self, request: RoutingRequest) -> RoutingPolicy:
if request.tenant_id and request.tenant_id in self.tenant_policies:
return self.tenant_policies[request.tenant_id]
return self.default_policy
def _select_from_tier(
self,
tier: ModelTier,
request: RoutingRequest
) -> list[ModelBackend]:
"""
Returns ordered list of candidates from a tier, filtering by
capability requirements and cost/latency constraints.
"""
candidates = self._tier_index.get(tier, [])
# Filter by required capabilities
if "vision" in request.required_capabilities:
candidates = [c for c in candidates if c.supports_vision]
if "functions" in request.required_capabilities:
candidates = [c for c in candidates if c.supports_functions]
# Filter by context window
if request.estimated_input_tokens > 0:
candidates = [
c for c in candidates
if c.max_context_tokens >= request.estimated_input_tokens * 1.2
]
# Filter by cost constraint
effective_max_cost = None
if request.max_cost_usd is not None:
effective_max_cost = request.max_cost_usd
if effective_max_cost is not None:
# Estimate cost: assume output is ~40% of input tokens (conservative)
estimated_output = request.estimated_input_tokens * 0.4
candidates = [
c for c in candidates
if (c.cost_per_input_token * request.estimated_input_tokens +
c.cost_per_output_token * estimated_output) <= effective_max_cost
]
# Filter by latency constraint
effective_max_latency = request.max_latency_ms
if effective_max_latency is not None:
candidates = [
c for c in candidates
if c.p95_latency_ms <= effective_max_latency
]
# Sort by weight (allows you to do weighted random within tier)
candidates.sort(key=lambda c: c.weight, reverse=True)
return candidates
def decide(self, request: RoutingRequest) -> RoutingDecision | None:
policy = self._get_policy(request)
start_time = time.time()
# Determine target tier from complexity
tier_name = policy.complexity_tier_map.get(
request.complexity.value, "standard"
)
target_tier = ModelTier(tier_name)
# If classifier confidence is low, upgrade one tier to be safe
if request.complexity_confidence < policy.min_confidence_to_downgrade:
tier_order = [ModelTier.MICRO, ModelTier.STANDARD, ModelTier.PREMIUM]
current_idx = tier_order.index(target_tier)
if current_idx < len(tier_order) - 1:
target_tier = tier_order[current_idx + 1]
# Respect explicit caller preference
if request.preferred_tier is not None:
target_tier = request.preferred_tier
# Build fallback chain: try target tier, then upgrade, never downgrade
tier_order = [ModelTier.MICRO, ModelTier.STANDARD, ModelTier.PREMIUM]
target_idx = tier_order.index(target_tier)
fallback_tiers = tier_order[target_idx:]
primary: ModelBackend | None = None
fallback_chain: list[ModelBackend] = []
for tier in fallback_tiers:
candidates = self._select_from_tier(tier, request)
for candidate in candidates:
if primary is None:
primary = candidate
else:
fallback_chain.append(candidate)
if primary is None:
return None # No viable backend found
# Estimate cost
est_output = request.estimated_input_tokens * 0.4
est_cost = (
primary.cost_per_input_token * request.estimated_input_tokens +
primary.cost_per_output_token * est_output
)
decision_ms = (time.time() - start_time) * 1000
return RoutingDecision(
request_id=request.request_id,
selected_backend=primary,
routing_reason=f"complexity={request.complexity.value}, "
f"confidence={request.complexity_confidence:.2f}, "
f"tier={target_tier.value}",
fallback_chain=fallback_chain,
estimated_cost_usd=est_cost,
estimated_latency_ms=primary.p50_latency_ms,
complexity_used=request.complexity,
decision_latency_ms=decision_ms,
)
Routing decisions mean nothing if you can't execute them reliably. Production LLM APIs fail. They rate-limit, they time out, they return 503s during deployments. Your fallback chain needs to execute automatically, and you need circuit breakers to avoid hammering a backend that's already struggling.
import asyncio
from collections import deque
from enum import auto
class CircuitState(Enum):
CLOSED = auto() # Normal operation
OPEN = auto() # Backend failing, reject fast
HALF_OPEN = auto() # Probing to see if backend recovered
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
success_threshold: int = 2,
timeout_seconds: float = 60.0,
window_size: int = 20
):
self.failure_threshold = failure_threshold
self.success_threshold = success_threshold
self.timeout_seconds = timeout_seconds
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: float = 0.0
self.recent_results: deque[bool] = deque(maxlen=window_size)
def record_success(self):
self.recent_results.append(True)
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.success_threshold:
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
def record_failure(self):
self.recent_results.append(False)
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.success_count = 0
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
def is_available(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time > self.timeout_seconds:
self.state = CircuitState.HALF_OPEN
self.success_count = 0
return True # Allow one probe request through
return False
# HALF_OPEN: allow through but monitor
return True
@property
def error_rate(self) -> float:
if not self.recent_results:
return 0.0
return 1.0 - (sum(self.recent_results) / len(self.recent_results))
Now let's wire together the dispatch layer that actually executes the fallback chain:
import aiohttp
from openai import AsyncOpenAI, APIStatusError, APITimeoutError
class LLMDispatcher:
def __init__(
self,
openai_client: AsyncOpenAI,
circuit_breakers: dict[str, CircuitBreaker] | None = None,
request_timeout_seconds: float = 30.0,
):
self.openai_client = openai_client
self.circuit_breakers = circuit_breakers or {}
self.request_timeout = request_timeout_seconds
def _get_circuit_breaker(self, backend: ModelBackend) -> CircuitBreaker:
if backend.model_id not in self.circuit_breakers:
self.circuit_breakers[backend.model_id] = CircuitBreaker()
return self.circuit_breakers[backend.model_id]
async def _call_backend(
self,
backend: ModelBackend,
messages: list[dict],
**kwargs
) -> dict:
"""
Actual API call. Returns a normalized response dict.
Real implementation would have provider-specific adapters.
"""
if backend.provider == "openai":
response = await asyncio.wait_for(
self.openai_client.chat.completions.create(
model=backend.model_id,
messages=messages,
**kwargs
),
timeout=self.request_timeout
)
return {
"content": response.choices[0].message.content,
"model": response.model,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
}
}
raise ValueError(f"Unsupported provider: {backend.provider}")
async def dispatch(
self,
decision: RoutingDecision,
messages: list[dict],
**kwargs
) -> tuple[dict, ModelBackend]:
"""
Returns (response, actual_backend_used).
Tries primary, then fallback chain in order.
"""
backends_to_try = [decision.selected_backend] + decision.fallback_chain
last_error: Exception | None = None
for backend in backends_to_try:
cb = self._get_circuit_breaker(backend)
if not cb.is_available():
# Circuit is open — skip this backend, try next
continue
try:
result = await self._call_backend(backend, messages, **kwargs)
cb.record_success()
return result, backend
except APITimeoutError as e:
cb.record_failure()
last_error = e
# Timeouts are worth retrying on next backend immediately
continue
except APIStatusError as e:
if e.status_code in (429, 503, 502):
# Rate limit or service unavailable — try fallback
cb.record_failure()
last_error = e
continue
elif e.status_code in (400, 401, 403):
# Client errors — don't try fallbacks, fail fast
cb.record_success() # Not the backend's fault
raise
else:
cb.record_failure()
last_error = e
continue
raise RuntimeError(
f"All backends exhausted for request {decision.request_id}. "
f"Last error: {last_error}"
)
Warning: Notice that we don't record circuit breaker failures on 400/401/403 errors. These are client-side errors — the backend is healthy, your request is malformed. Naively recording all errors as backend failures will open circuit breakers incorrectly and route requests to backends that can't handle them either.
Now we bring everything together into a single LLMRouter class with a clean interface:
class LLMRouter:
def __init__(
self,
backends: list[ModelBackend],
policy: RoutingPolicy,
openai_client: AsyncOpenAI,
confidence_threshold: float = 0.80,
tenant_policies: dict[str, RoutingPolicy] | None = None,
):
self.classifier_embeddings = EmbeddingClassifier(openai_client)
self.policy_engine = PolicyEngine(backends, policy, tenant_policies)
self.dispatcher = LLMDispatcher(openai_client)
self.confidence_threshold = confidence_threshold
self._metrics: list[dict] = [] # In production, use a real metrics sink
async def route_and_complete(
self,
messages: list[dict],
user_id: str | None = None,
tenant_id: str | None = None,
max_cost_usd: float | None = None,
max_latency_ms: float | None = None,
required_capabilities: list[str] | None = None,
**completion_kwargs
) -> dict:
request = RoutingRequest(
messages=messages,
user_id=user_id,
tenant_id=tenant_id,
max_cost_usd=max_cost_usd,
max_latency_ms=max_latency_ms,
required_capabilities=required_capabilities or [],
)
# Stage 1: Token count (always fast)
signals = extract_complexity_signals(messages)
request.estimated_input_tokens = signals.token_count
# Stage 2: Heuristic classification
complexity, confidence = heuristic_complexity(signals)
# Stage 3: Embedding classification only if confidence is low
if confidence < self.confidence_threshold:
try:
complexity, confidence = await self.classifier_embeddings.classify(
messages
)
except Exception:
# Embedding call failed — use heuristic result, upgrade tier
confidence = min(confidence, 0.50)
request.complexity = complexity
request.complexity_confidence = confidence
# Stage 4: Policy decision
decision = self.policy_engine.decide(request)
if decision is None:
raise RuntimeError("No viable backend available given constraints")
# Stage 5: Dispatch with fallback
dispatch_start = time.time()
response, actual_backend = await self.dispatcher.dispatch(
decision, messages, **completion_kwargs
)
dispatch_ms = (time.time() - dispatch_start) * 1000
# Stage 6: Record metrics
self._record_metric(request, decision, actual_backend, dispatch_ms, response)
return response
def _record_metric(
self,
request: RoutingRequest,
decision: RoutingDecision,
actual_backend: ModelBackend,
dispatch_ms: float,
response: dict,
):
usage = response.get("usage", {})
actual_cost = (
actual_backend.cost_per_input_token * usage.get("prompt_tokens", 0) +
actual_backend.cost_per_output_token * usage.get("completion_tokens", 0)
)
self._metrics.append({
"request_id": request.request_id,
"tenant_id": request.tenant_id,
"complexity": request.complexity.value,
"confidence": request.complexity_confidence,
"planned_model": decision.selected_backend.model_id,
"actual_model": actual_backend.model_id,
"fallback_used": actual_backend.model_id != decision.selected_backend.model_id,
"estimated_cost": decision.estimated_cost_usd,
"actual_cost": actual_cost,
"dispatch_latency_ms": dispatch_ms,
"decision_latency_ms": decision.decision_latency_ms,
"prompt_tokens": usage.get("prompt_tokens", 0),
"completion_tokens": usage.get("completion_tokens", 0),
})
For multi-tenant SaaS applications, you need per-tenant budget enforcement, not just per-request constraints. Let's add a Redis-backed budget manager that tracks spend in a rolling window.
import redis.asyncio as aioredis
import json
class BudgetManager:
def __init__(self, redis_client: aioredis.Redis, window_seconds: int = 3600):
self.redis = redis_client
self.window = window_seconds
async def get_remaining_budget(
self,
tenant_id: str,
budget_usd: float
) -> float:
key = f"budget:{tenant_id}:{int(time.time() // self.window)}"
spent_raw = await self.redis.get(key)
spent = float(spent_raw) if spent_raw else 0.0
return max(0.0, budget_usd - spent)
async def record_spend(self, tenant_id: str, cost_usd: float):
key = f"budget:{tenant_id}:{int(time.time() // self.window)}"
pipe = self.redis.pipeline()
pipe.incrbyfloat(key, cost_usd)
pipe.expire(key, self.window * 2) # Keep for 2 windows for debugging
await pipe.execute()
async def check_and_reserve(
self,
tenant_id: str,
estimated_cost: float,
budget_usd: float
) -> bool:
"""
Atomic check-and-reserve. Returns True if budget is available.
Uses a Lua script to make this atomic in Redis.
"""
lua_script = """
local key = KEYS[1]
local budget = tonumber(ARGV[1])
local cost = tonumber(ARGV[2])
local ttl = tonumber(ARGV[3])
local current = tonumber(redis.call('GET', key)) or 0
if current + cost > budget then
return 0
end
redis.call('INCRBYFLOAT', key, cost)
redis.call('EXPIRE', key, ttl)
return 1
"""
key = f"budget:{tenant_id}:{int(time.time() // self.window)}"
result = await self.redis.eval(
lua_script, 1, key, budget_usd, estimated_cost, self.window * 2
)
return bool(result)
Architecture note: The atomic Lua script is not optional here. Without it, you have a race condition under concurrent requests for the same tenant — two requests can both read "budget available," both proceed, and together exceed the limit. Redis Lua scripts execute atomically, solving this at the cost of slightly higher Redis latency (~1-2ms). This is almost always the right trade-off.
A router you can't observe is a router you can't improve. The most valuable signal is the delta between your complexity prediction and actual quality outcomes — but that requires a feedback loop from your evaluation system. Here's what to instrument, even before you have that loop:
from dataclasses import dataclass, asdict
@dataclass
class RouterObservabilityEvent:
# Routing metadata
request_id: str
timestamp: float
tenant_id: str | None
# Classification
heuristic_complexity: str
heuristic_confidence: float
embedding_classifier_used: bool
final_complexity: str
final_confidence: float
# Decision
target_tier: str
selected_model: str
fallback_triggered: bool
actual_model_used: str
# Economics
estimated_cost_usd: float
actual_cost_usd: float
cost_estimation_error_pct: float
# Performance
decision_latency_ms: float
embedding_latency_ms: float
total_dispatch_latency_ms: float
# Quality signals (filled in async via eval pipeline)
user_satisfaction_score: float | None = None
quality_eval_score: float | None = None
required_retry: bool = False
async def emit_to_datawarehouse(event: RouterObservabilityEvent, sink):
"""
In production, this goes to BigQuery, Snowflake, or ClickHouse.
The key is that every routing decision is a row you can analyze.
"""
await sink.write(asdict(event))
The metric that tells you most about routing quality is cost_estimation_error_pct. If you're consistently underestimating output length for certain task types, you'll be routing them to cheaper models than intended. Track this by complexity class and you'll quickly find systematic biases.
Another critical metric: track the fallback_triggered rate by backend. If a specific model is triggering fallbacks more than 5% of the time, you either have a reliability problem with that provider or your circuit breaker thresholds need tuning.
Set up a working router instance and run it against a varied test suite. Here's your exercise scaffold:
import asyncio
from openai import AsyncOpenAI
async def exercise_main():
client = AsyncOpenAI() # Set OPENAI_API_KEY in environment
backends = [
ModelBackend(
model_id="gpt-4o-mini",
tier=ModelTier.MICRO,
provider="openai",
cost_per_input_token=0.00000015,
cost_per_output_token=0.0000006,
p50_latency_ms=400,
p95_latency_ms=1200,
max_context_tokens=128000,
supports_functions=True,
),
ModelBackend(
model_id="gpt-4o",
tier=ModelTier.STANDARD,
provider="openai",
cost_per_input_token=0.0000025,
cost_per_output_token=0.00001,
p50_latency_ms=800,
p95_latency_ms=3000,
max_context_tokens=128000,
supports_functions=True,
supports_vision=True,
),
]
policy = RoutingPolicy(name="default")
router = LLMRouter(
backends=backends,
policy=policy,
openai_client=client,
)
# Test cases — vary dramatically in complexity
test_cases = [
# Should route to micro
{"messages": [{"role": "user", "content": "What is the capital of France?"}]},
# Should route to micro
{"messages": [{"role": "user", "content": "Translate 'good morning' to Spanish."}]},
# Should route to standard
{"messages": [{"role": "user", "content":
"Compare the trade-offs between PostgreSQL and MongoDB for "
"a multi-tenant SaaS application with variable schema requirements "
"and high read throughput. Consider indexing, consistency, and "
"operational complexity."}]},
# Should route to standard/premium
{"messages": [{"role": "user", "content":
"Here's a Python function that's causing intermittent deadlocks "
"in our async worker pool:\n```python\n"
"async def process_batch(items):\n"
" lock = asyncio.Lock()\n"
" async with lock:\n"
" results = await asyncio.gather(*[process_item(i) for i in items])\n"
" return results\n```\n"
"Analyze why this causes deadlocks, explain the exact failure mode, "
"and provide a corrected implementation with explanation."}]},
]
for i, test in enumerate(test_cases):
print(f"\n--- Test {i+1} ---")
print(f"Prompt: {test['messages'][0]['content'][:80]}...")
response = await router.route_and_complete(**test)
metric = router._metrics[-1]
print(f"Routed to: {metric['actual_model']}")
print(f"Complexity: {metric['complexity']} (confidence: {metric['confidence']:.2f})")
print(f"Actual cost: ${metric['actual_cost']:.6f}")
print(f"Dispatch latency: {metric['dispatch_latency_ms']:.0f}ms")
print(f"Response: {response['content'][:120]}...")
asyncio.run(exercise_main())
Exercise extensions to try:
RoutingPolicy to set min_confidence_to_downgrade = 0.95 and observe how more requests get upgraded to the next tier.circuit_breakers["gpt-4o-mini"].state = CircuitState.OPEN and verify that requests fall through to the standard tier.Token count is a proxy for complexity, not a measure of it. A 10-token prompt asking "Prove the Riemann Hypothesis" is trivially short but maximally hard. A 2000-token prompt that's a verbatim copy-paste of a document with the question "What's the title of this document?" is long but trivial. Always combine token count with semantic signals.
If your embedding classifier adds 80ms to every request and routing to a MICRO model saves 200ms over STANDARD, you're netting only 120ms — and that's before you account for the fact that the embedding call can fail. Profile your classifier end-to-end. The heuristic fast-path isn't optional for high-traffic systems.
Cost estimation based solely on input tokens is systematically wrong. A "low complexity" request can generate a very long output (e.g., "list all 50 US state capitals"). Track actual completion_tokens per complexity class in your observability data, compute a realistic output multiplier per task type, and feed it back into your cost model.
Your reference set for the embedding classifier needs to evolve with your traffic. If you built it six months ago and your user base has shifted toward more technical questions, your "medium complexity" cluster is miscalibrated. Schedule quarterly re-labeling sessions and track classification drift in your metrics.
Some implementations query the policy engine again on each fallback attempt. This is a latency disaster and introduces the risk of inconsistent decisions (a backend that was available when the primary call failed might now be the "primary" for the retry). Pre-compute the full fallback chain at decision time. If backend availability changes between primary and fallback calls, the circuit breaker handles it — don't re-run policy logic.
Rate limiting behavior is asymmetric across model tiers. Premium models typically have lower rate limits (requests per minute) but are less likely to throw 503s. Micro models have higher throughput limits but are more commonly rate-limited during peak hours. Tune your circuit breakers per-model, not with a single global configuration.
If you're seeing 70%+ of requests being routed to STANDARD when you expected 50%, check:
min_confidence_to_downgrade setting — lower it if the embedding classifier is consistently uncertainIf fallback_triggered rate exceeds 8% for a specific backend, investigate:
You've built a production-grade LLM router that:
The system you've built is genuinely useful as-is, but production hardening never stops. Here's where to invest next:
Immediate extensions:
quality_eval_score on observability events, then use this data to retrain the complexity thresholdsArchitecture evolutions:
The investment in routing infrastructure pays for itself quickly — at scale, the difference between "always use the best model" and "use the right model" is measured in tens of thousands of dollars per month. More importantly, it gives you a policy layer you can reason about and tune, rather than a single point of failure where one model's degradation takes down your entire product.
Learning Path: Building with LLMs