LLM Performance Series Part 3: Reducing Costs with Routing, Fallbacks, and Ensembles

from sklearn.linear_model import LogisticRegression
import numpy as np

class QueryRouter:
 """ """

    COMPLEXITY_LEVELS = {
 "simple": "claude-haiku-4-5", # , FAQ
 "medium": "claude-sonnet-4-6", #
 "complex": "claude-opus-4-7", # ,
    }

    def __init__(self):
        self.classifier = LogisticRegression()
 # : , , ,
        self.feature_extractors = [
            lambda q: len(q.split()),
 lambda q: int(any(kw in q for kw in ["", "", "", "sql"])),
            lambda q: q.count("?"),
            lambda q: int(len(q) > 200),
        ]

    def extract_features(self, query: str) -> np.ndarray:
        return np.array([f(query) for f in self.feature_extractors]).reshape(1, -1)

    def route(self, query: str) -> str:
        if not hasattr(self.classifier, "classes_"):
 return "medium" # when
        level = self.classifier.predict(self.extract_features(query))[0]
        return self.COMPLEXITY_LEVELS[level]

    async def query(self, user_input: str) -> str:
        model = self.route(user_input)
        return await llm_call(model, user_input)

def simple_router(query: str) -> str:
    tokens = len(query.split())
 has_code = any(kw in query.lower for kw in ["", "sql", "", ""])
    has_numbers = sum(c.isdigit() for c in query) > 5
    is_long = tokens > 100

    if has_code or (is_long and has_numbers):
        return "claude-opus-4-7"
    elif tokens > 30 or has_numbers:
        return "claude-sonnet-4-6"
    else:
        return "claude-haiku-4-5"

async def escalating_query(query: str) -> dict:
    models = [
 ("claude-haiku-4-5", 0.85), # (, )
        ("claude-sonnet-4-6", 0.90),
 ("claude-opus-4-7", None), # —
    ]

    for model, threshold in models:
        response = await llm_call(model, query)
        confidence = await estimate_confidence(response)

        if threshold is None or confidence >= threshold:
            return {
                "answer": response,
                "model_used": model,
                "confidence": confidence
            }

 print(f"{model} {confidence:.2f} < {threshold} → ")

async def ensemble_query(query: str, models: list, weights: list = None) -> str:
    responses = await asyncio.gather(*[llm_call(m, query) for m in models])

    if weights is None:
        weights = [1.0] * len(models)

 # reply after
    embeddings = await asyncio.gather(*[embed(r) for r in responses])
    centroid = np.average(embeddings, axis=0, weights=weights)

 # reply
    distances = [np.linalg.norm(e - centroid) for e in embeddings]
    best_idx = np.argmin(distances)
    return responses[best_idx]

SELECT route, count(*) AS n,
       avg(CASE WHEN regenerated THEN 1 ELSE 0 END) AS regen_rate
FROM llm_requests WHERE created_at > now() - interval '7 days'
GROUP BY route;