# 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)
```
```python
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
```
```python
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)
```python
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
```python
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
```python
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
```python
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)
```python
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

| Factor | Measurement | Weight |
|---|---|---|
| **Task Type** | Classification accuracy | High |
| **Complexity** | Token count, reasoning depth | High |
| **Domain** | Keywords, embeddings | High |
| **Latency Budget** | SLA requirements | Medium |
| **Cost Budget** | $ per query | Medium |
| **Quality Threshold** | Minimum acceptable | High |
| **Availability** | Model health, rate limits | Medium |
| **Privacy** | Data sensitivity | High |

---

## Routing Patterns

### 1. Mixture of Agents (MoA)
```python
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)
```python
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)
```python
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
| Metric | Description |
|---|---|
| **Routing Accuracy** | % queries sent to optimal model |
| **Cost Savings** | % reduction vs always-large |
| **Latency Reduction** | % reduction vs always-large |
| **Quality Preservation** | Quality vs always-large baseline |
| **Abstention Rate** | % queries router abstains |

### A/B Testing
```python
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
```python
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
```python
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
```python
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

- [[ModelArchitectures/Mixture-of-Experts.md]] — MoE as internal routing
- [[ModelArchitectures/Small-Language-Models.md]] — Small models for routing
- [[ModelArchitectures/Inference-Optimization.md]] — Routing for latency
- [[ModelArchitectures/Reasoning-Models.md]] — Routing to reasoning models
- [[Production/Cost-Optimization.md]] — Cost-aware routing
- [[Production/Scaling-Strategies.md]] — Routing for scaling
- [[Production/Reliability.md]] — Circuit breakers, health checks
- [[Evaluation/Agent-Benchmarks.md]] — Router benchmarks
- [[Frontiers/Research-Directions.md]] — 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).*