Model Routing

Dynamic selection and cascading between models for cost, quality, and latency optimization.

Routing Strategies

1. Cascade Routing (Sequential)

Query → Small Model → Confidence Check → Large Model (if needed)

class CascadeRouter:
    def __init__(self, models: list[Model], thresholds: list[float]):
        self.models = models  # Smallest to largest
        self.thresholds = thresholds  # Confidence thresholds

    async def route(self, query: str) -> Response:
        for model, threshold in zip(self.models, self.thresholds):
            response = await model.generate(query)
            confidence = self.estimate_confidence(response)

            if confidence >= threshold or model == self.models[-1]:
                return response

            # Escalate to next model
            logger.info(f"Escalating: {model.name} confidence {confidence:.2f} < {threshold}")

        return response

2. Parallel Routing (Consensus)

Query → [Model A, Model B, Model C] → Aggregate → Best Answer

class ParallelRouter:
    def __init__(self, models: list[Model], aggregator: Aggregator):
        self.models = models
        self.aggregator = aggregator

    async def route(self, query: str) -> Response:
        responses = await asyncio.gather(*[
            m.generate(query) for m in self.models
        ])
        return self.aggregator.aggregate(responses)

3. Semantic Routing (Content-Based)

class SemanticRouter:
    def __init__(self):
        self.classifier = RouterClassifier()  # Small model
        self.routes = {
            "coding": CodeModel(),
            "math": MathModel(),
            "creative": CreativeModel(),
            "factual": FactualModel(),
            "reasoning": ReasoningModel()
        }

    async def route(self, query: str) -> Response:
        category = await self.classifier.classify(query)
        model = self.routes[category]
        return await model.generate(query)

4. Cost-Aware Routing

class CostAwareRouter:
    def __init__(self, models: list[Model], budget: CostBudget):
        self.models = models
        self.budget = budget

    async def route(self, query: str, priority: Priority) -> Response:
        # Filter by budget
        affordable = [m for m in self.models if m.cost_per_token <= self.budget.max_per_query]

        # Select based on priority
        if priority == Priority.SPEED:
            model = min(affordable, key=lambda m: m.latency_p50)
        elif priority == Priority.QUALITY:
            model = max(affordable, key=lambda m: m.quality_score)
        else:  # BALANCED
            model = max(affordable, key=lambda m: m.quality_score / m.cost_per_token)

        return await model.generate(query)

Router Implementation

LLM-Based Router

class LLMRouter:
    def __init__(self, router_model: Model, routes: dict[str, Model]):
        self.router = router_model
        self.routes = routes

    async def route(self, query: str) -> Response:
        prompt = f"""
        Route this query to the best model:

        Available models:
        {self.format_routes()}

        Query: {query}

        Respond with JSON: {{"model": "model_name", "reasoning": "..."}}
        """

        decision = await self.router.generate_json(prompt)
        model = self.routes[decision["model"]]
        return await model.generate(query)

Embedding-Based Router

class EmbeddingRouter:
    def __init__(self, routes: dict[str, Model], embed_model):
        self.routes = routes
        self.embed = embed_model
        self.route_embeddings = {
            name: self.embed.embed(desc) 
            for name, desc in ROUTE_DESCRIPTIONS.items()
        }

    def route(self, query: str) -> Model:
        query_emb = self.embed.embed(query)
        similarities = {
            name: cosine(query_emb, emb) 
            for name, emb in self.route_embeddings.items()
        }
        best_route = max(similarities, key=similarities.get)
        return self.routes[best_route]

Learned Router (Trainable)

class LearnedRouter(nn.Module):
    def __init__(self, embed_dim: int, num_routes: int):
        super().__init__()
        self.encoder = TransformerEncoder(embed_dim)
        self.router_head = nn.Linear(embed_dim, num_routes)

    def forward(self, query_emb):
        encoded = self.encoder(query_emb)
        logits = self.router_head(encoded.mean(dim=1))
        return F.softmax(logits, dim=-1)

# Training: reward = quality - λ * cost
def router_loss(router_logits, target_route, quality, cost):
    ce_loss = F.cross_entropy(router_logits, target_route)
    cost_penalty = cost.mean()
    return ce_loss + λ * cost_penalty

Routing Decision Factors

Routing Patterns

1. Mixture of Agents (MoA)

class MixtureOfAgents:
    def __init__(self, agents: list[Agent], aggregator: Agent):
        self.agents = agents
        self.aggregator = aggregator

    async def run(self, query: str):
        # Layer 1: Parallel diverse responses
        layer1 = await asyncio.gather(*[a.run(query) for a in self.agents])

        # Layer 2: Aggregate with context
        layer2 = await self.aggregator.run(
            f"Query: {query}\n\nResponses:\n{layer1}\n\nSynthesize best answer:"
        )

        # Layer 3: Refine (optional)
        return await self.aggregator.run(
            f"Refine: {layer2}"
        )

2. Adaptive Compute (Early Exit)

class EarlyExitModel:
    def __init__(self, backbone, exits: list[ExitLayer]):
        self.backbone = backbone
        self.exits = exits

    def forward(self, x, threshold=0.9):
        for i, layer in enumerate(self.backbone.layers):
            x = layer(x)

            if i in self.exit_points:
                logits = self.exits[i](x)
                probs = F.softmax(logits, dim=-1)
                confidence = probs.max(dim=-1).values

                if confidence > threshold:
                    return logits, i  # Early exit

        return self.final_head(x), len(self.backbone.layers)

3. Speculative Decoding (Model Cascade)

class SpeculativeCascade:
    def __init__(self, draft_model: Model, target_model: Model):
        self.draft = draft_model
        self.target = target_model

    async def generate(self, prompt: str) -> str:
        # Draft generates k tokens
        draft_tokens = await self.draft.generate(prompt, max_tokens=10)

        # Target verifies in parallel
        target_logits = await self.target.score(prompt + draft_tokens)

        # Accept/reject
        accepted = self.verify(draft_tokens, target_logits)

        if accepted:
            return prompt + accepted
        else:
            # Fallback to target
            return await self.target.generate(prompt)

Evaluation

Router Metrics

A/B Testing

class RouterABTest:
    def __init__(self, router_a, router_b, traffic_split=0.5):
        self.router_a = router_a
        self.router_b = router_b
        self.split = traffic_split

    async def route(self, query, user_id):
        if hash(user_id) % 100 < self.split * 100:
            return await self.router_a.route(query)
        return await self.router_b.route(query)

Production Patterns

1. Health-Aware Routing

class HealthAwareRouter:
    def __init__(self, models, health_checker):
        self.models = models
        self.health = health_checker

    async def route(self, query):
        healthy = await self.health.get_healthy_models()
        if not healthy:
            raise AllModelsUnhealthy()
        return await self.route_to_healthy(query, healthy)

2. Circuit Breaker

class CircuitBreakerRouter:
    def __init__(self, models, failure_threshold=5, timeout=60):
        self.models = models
        self.failures = defaultdict(int)
        self.last_failure = defaultdict(float)
        self.threshold = failure_threshold
        self.timeout = timeout

    def is_available(self, model):
        if self.failures[model] >= self.threshold:
            if time.time() - self.last_failure[model] > self.timeout:
                self.failures[model] = 0  # Reset
                return True
            return False
        return True

3. Canary Routing

class CanaryRouter:
    def __init__(self, stable_model, canary_model, canary_rate=0.05):
        self.stable = stable_model
        self.canary = canary_model
        self.rate = canary_rate

    async def route(self, query, user_id):
        if hash(user_id) % 100 < self.rate * 100:
            return await self.canary.generate(query)
        return await self.stable.generate(query)

Related Notes

• Mixture of Experts — MoE as internal routing
• Small Language Models — Small models for routing
• Inference Optimization — Routing for latency
• Reasoning Models — Routing to reasoning models
• Cost Optimization — Cost-aware routing
• Scaling Strategies — Routing for scaling
• Reliability — Circuit breakers, health checks
• Agent Benchmarks — Router benchmarks
• Research Directions — Learned routing research

Sources: "Model Routing" surveys, "Cascade" papers, "Mixture of Agents" (Wang et al.), "Early Exit" (Schwartz et al.), "Speculative Decoding" (Leviathan et al.), OpenRouter, LiteLLM, LangChain router, production routing systems (Shopify, Stripe, Notion).