LLM Routing: How to Pick the Cheapest Model That Works

response = openai.chat.completions.create(
    model="gpt-4o",  # $2.50 input / $10 output per 1M tokens
    messages=messages,
)

Quality
  │
  │     ┌──── Frontier (Opus, o1)
  │    ╱
  │   ╱  ← Mid (GPT-4o, Sonnet) — sweet spot for most tasks
  │  ╱
  │ ╱──── Nano (mini, Haiku) — great for simple tasks
  │╱
  └────────────────── Complexity

ROUTE_THRESHOLDS = {
    "classification": {"min_accuracy": 0.95, "model": "gpt-4o-mini"},
    "rag_qa": {"min_helpfulness": 0.85, "model": "gpt-4o-mini"},
    "reasoning": {"min_accuracy": 0.75, "model": "gpt-4o"},
    "code": {"min_pass_rate": 0.80, "model": "gpt-4o"},
}

User query → Classifier (mini/Haiku) → Route label → Execution model → Response

from enum import Enum
from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

class TaskComplexity(str, Enum):
    SIMPLE = "simple"       # → gpt-4o-mini
    MODERATE = "moderate"   # → gpt-4o
    COMPLEX = "complex"     # → gpt-4o (or o1 for reasoning)

class RouteDecision(BaseModel):
    complexity: TaskComplexity
    reasoning: str

ROUTE_MAP = {
    TaskComplexity.SIMPLE: "gpt-4o-mini",
    TaskComplexity.MODERATE: "gpt-4o",
    TaskComplexity.COMPLEX: "gpt-4o",
}

CLASSIFIER_PROMPT = """Classify this user request by complexity:
- simple: lookup, classification, formatting, single-fact Q&A
- moderate: multi-step Q&A, summarization, structured extraction
- complex: multi-hop reasoning, code generation, analysis, creative writing

Request: {query}"""

def classify_query(query: str) -> RouteDecision:
    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": CLASSIFIER_PROMPT.format(query=query)}],
        response_format=RouteDecision,
    )
    return response.choices[0].message.parsed

def routed_completion(query: str, system: str = "You are a helpful assistant.") -> str:
    decision = classify_query(query)
    model = ROUTE_MAP[decision.complexity]

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": query},
        ],
    )
    return response.choices[0].message.content

Query → Cheap model → Confidence check → OK? → Return
                              ↓ No
                         Mid model → Confidence check → OK? → Return
                                              ↓ No
                                         Expensive model → Return

import json
from dataclasses import dataclass

@dataclass
class CascadeResult:
    response: str
    model_used: str
    confidence: float
    cascades: int  # how many models were tried

CASCADE_CHAIN = [
    {"model": "gpt-4o-mini", "min_confidence": 0.85},
    {"model": "gpt-4o", "min_confidence": 0.80},
    {"model": "gpt-4o", "min_confidence": 0.0},  # final fallback, always accept
]

CONFIDENCE_PROMPT = """Answer the question. Then rate your confidence 0.0-1.0.

Question: {query}

Respond as JSON: {{"answer": "...", "confidence": 0.0-1.0}}"""

async def cascade_complete(query: str) -> CascadeResult:
    cascades = 0

    for tier in CASCADE_CHAIN:
        cascades += 1
        response = await client.chat.completions.create(
            model=tier["model"],
            messages=[{"role": "user", "content": CONFIDENCE_PROMPT.format(query=query)}],
            response_format={"type": "json_object"},
        )

        parsed = json.loads(response.choices[0].message.content)
        confidence = parsed["confidence"]

        if confidence >= tier["min_confidence"]:
            return CascadeResult(
                response=parsed["answer"],
                model_used=tier["model"],
                confidence=confidence,
                cascades=cascades,
            )
    raise RuntimeError("Cascade exhausted without result")

async def self_consistency_cascade(query: str, n_samples: int = 3) -> CascadeResult:
    answers = []
    for _ in range(n_samples):
        result = await cheap_complete(query)
        answers.append(result)

    if len(set(answers)) == 1:
        return CascadeResult(response=answers[0], model_used="gpt-4o-mini", confidence=1.0, cascades=1)

    # Disagreement → escalate
    return await cascade_complete(query)

Query → Embed → Compare to route embeddings → Best match → Execution model

import numpy as np
from openai import OpenAI

client = OpenAI()

ROUTES = [
    {
        "name": "faq",
        "description": "Simple FAQ lookup, hours, pricing, contact info",
        "model": "gpt-4o-mini",
        "examples": [
            "What are your business hours?",
            "How much does the pro plan cost?",
            "What's your refund policy?",
        ],
    },
    {
        "name": "technical_support",
        "description": "Technical troubleshooting, error messages, configuration",
        "model": "gpt-4o",
        "examples": [
            "I'm getting a 403 error on the API endpoint",
            "How do I configure SSO with SAML?",
            "The webhook isn't firing after payment",
        ],
    },
    {
        "name": "analysis",
        "description": "Data analysis, comparisons, recommendations",
        "model": "gpt-4o",
        "examples": [
            "Compare our Q3 vs Q4 revenue trends",
            "Analyze this customer feedback and suggest improvements",
            "What pricing strategy would maximize conversion?",
        ],
    },
]

def embed_texts(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return [item.embedding for item in response.data]

def build_route_index():
    route_embeddings = []
    for route in ROUTES:
        example_embeddings = embed_texts(route["examples"])
        centroid = np.mean(example_embeddings, axis=0)
        route_embeddings.append(centroid)
    return route_embeddings

ROUTE_EMBEDDINGS = build_route_index()

def semantic_route(query: str, threshold: float = 0.75) -> dict:
    query_embedding = embed_texts([query])[0]

    similarities = [
        np.dot(query_embedding, route_emb) /
        (np.linalg.norm(query_embedding) * np.linalg.norm(route_emb))
        for route_emb in ROUTE_EMBEDDINGS
    ]

    best_idx = int(np.argmax(similarities))
    best_score = similarities[best_idx]

    if best_score < threshold:
        return {"name": "fallback", "model": "gpt-4o", "score": best_score}

    return {
        "name": ROUTES[best_idx]["name"],
        "model": ROUTES[best_idx]["model"],
        "score": best_score,
    }

import re
from dataclasses import dataclass

@dataclass
class Rule:
    name: str
    pattern: re.Pattern
    model: str
    priority: int

RULES = sorted([
    Rule("code", re.compile(r"```|def |class |import |function |debug", re.I), "gpt-4o", 1),
    Rule("sql", re.compile(r"\b(SELECT|INSERT|UPDATE|DELETE|JOIN)\b", re.I), "gpt-4o-mini", 2),
    Rule("short_query", re.compile(r"^.{1,50}$"), "gpt-4o-mini", 3),
    Rule("long_document", re.compile(r".{5000,}", re.S), "gpt-4o", 4),
], key=lambda r: r.priority)

DEFAULT_MODEL = "gpt-4o-mini"

def rule_route(query: str) -> str:
    for rule in RULES:
        if rule.pattern.search(query):
            return rule.model
    return DEFAULT_MODEL

async def production_route(query: str, metadata: dict) -> str:
    # Layer 1: Hard rules (free, instant)
    if metadata.get("user_tier") == "enterprise":
        return "gpt-4o"  # SLA guarantee
    if metadata.get("task_type") == "classification":
        return "gpt-4o-mini"  # known-simple task

    rule_model = rule_route(query)
    if rule_model != DEFAULT_MODEL:
        return rule_model

    # Layer 2: Semantic router (cheap, fast)
    semantic = semantic_route(query)
    if semantic["score"] > 0.85:
        return semantic["model"]

    # Layer 3: Classifier (small LLM call)
    decision = classify_query(query)
    return ROUTE_MAP[decision.complexity]

from fastapi import FastAPI
from pydantic import BaseModel
import time
import structlog

logger = structlog.get_logger()
app = FastAPI()

class RouteRequest(BaseModel):
    query: str
    user_tier: str = "standard"
    task_type: str | None = None
    max_model: str | None = None  # cost cap

class RouteResponse(BaseModel):
    response: str
    model_used: str
    route_path: str
    latency_ms: float
    estimated_cost_usd: float

MODEL_COSTS = {
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},  # per 1M tokens
    "gpt-4o": {"input": 2.50, "output": 10.00},
}

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

@app.post("/v1/completions")
async def routed_completion(req: RouteRequest) -> RouteResponse:
    start = time.monotonic()
    route_path = "unknown"

    # Route selection
    if req.task_type == "classification":
        model = "gpt-4o-mini"
        route_path = "rule:task_type"
    elif req.user_tier == "enterprise":
        model = "gpt-4o"
        route_path = "rule:enterprise"
    else:
        semantic = semantic_route(req.query)
        if semantic["score"] > 0.85:
            model = semantic["model"]
            route_path = f"semantic:{semantic['name']}"
        else:
            decision = classify_query(req.query)
            model = ROUTE_MAP[decision.complexity]
            route_path = f"classifier:{decision.complexity}"

    if req.max_model and model_tier(model) > model_tier(req.max_model):
        model = req.max_model
        route_path += ":capped"

    # Execute
    response = await client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": req.query}],
    )
    content = response.choices[0].message.content
    usage = response.usage

    latency = (time.monotonic() - start) * 1000
    cost = estimate_cost(model, usage.prompt_tokens, usage.completion_tokens)

    logger.info(
        "routed_completion",
        model=model,
        route_path=route_path,
        latency_ms=latency,
        cost_usd=cost,
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
    )

    return RouteResponse(
        response=content,
        model_used=model,
        route_path=route_path,
        latency_ms=latency,
        estimated_cost_usd=cost,
    )

AGENT_STEP_ROUTES = {
    "plan": "gpt-4o",           # reasoning-heavy
    "execute_tool": "gpt-4o-mini",  # structured, constrained
    "summarize": "gpt-4o-mini",     # simple synthesis
    "reflect": "gpt-4o",        # quality check
}

from prometheus_client import Counter, Histogram

ROUTE_COUNTER = Counter("llm_route_total", "Routes by path", ["route_path", "model"])
ROUTE_COST = Histogram("llm_route_cost_usd", "Cost per request", ["model"])
ROUTE_LATENCY = Histogram("llm_route_latency_ms", "End-to-end latency", ["route_path"])

import hashlib

def cache_key(query: str, model: str) -> str:
    return hashlib.sha256(f"{model}:{query}".encode()).hexdigest()

async def cached_routed_complete(query: str) -> RouteResponse:
    model = select_model(query)
    key = cache_key(query, model)

    cached = await redis.get(key)
    if cached:
        return RouteResponse(**json.loads(cached), route_path="cache:hit")

    result = await routed_completion(query, model)
    await redis.setex(key, 3600, result.model_dump_json())
    return result