Model Routing Rules Engine: Production Architecture 2026

Overview

Model Routing Rules Engine solves the challenge of intelligently routing requests to optimal models. This guide covers the design decisions, implementation details, and trade-offs you need to know.

Why This Pattern Matters

As AI applications scale from prototype to production, you need systematic approaches to intelligently routing requests to optimal models. Without this pattern, teams face:

Architecture Diagram

┌─────────────────────────────────────┐
│          Client / Frontend          │
└────────────┬────────────────────────┘
             │ API Request
             ▼
┌─────────────────────────────────────┐
│         Model Routing Rules Layer       │
│                                     │
│  ┌──────────┐  ┌──────────────────┐ │
│  │  Router  │  │    Controller    │ │
│  └────┬─────┘  └────────┬─────────┘ │
│       │                 │           │
│  ┌────▼─────────────────▼─────────┐ │
│  │      Core Processing Engine     │ │
│  └────────────────┬───────────────┘ │
└───────────────────┼─────────────────┘
                    │
        ┌───────────┼───────────┐
        ▼           ▼           ▼
   ┌─────────┐ ┌─────────┐ ┌─────────┐
   │  LLM 1  │ │  LLM 2  │ │ Storage │
   └─────────┘ └─────────┘ └─────────┘

Implementation

Core Data Models

python
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
from enum import Enum
class AIRequest(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    input: str
    model: str = "gpt-4o-mini"
    priority: int = Field(default=5, ge=1, le=10)
    metadata: dict = {}
    created_at: datetime = Field(default_factory=datetime.utcnow)
    
class AIResponse(BaseModel):
    request_id: str
    output: str
    model_used: str
    latency_ms: float
    tokens_used: int
    cost_usd: float
    status: ProcessingStatus = ProcessingStatus.COMPLETED

Main Implementation

python
import asyncio
import time
from openai import AsyncOpenAI
from typing import Optional
class ModelRoutingRulesEngine:
    """Implements Model Routing Rules Engine for intelligently routing requests to optimal models."""
    
    def __init__(self, config: dict):
        self.config = config
        self.client = AsyncOpenAI()
        self._setup()
    
    def _setup(self):
        """Initialize all required components."""
        # Configure based on pattern requirements
        self.primary_model = self.config.get('primary_model', 'gpt-4o-mini')
        self.fallback_model = self.config.get('fallback_model', 'gpt-4o-mini')
        self.max_retries = self.config.get('max_retries', 3)
    
    async def process(self, request: AIRequest) -> AIResponse:
        """Process an AI request with Model Routing Rules Engine."""
        start_time = time.time()
        
        try:
            # Primary processing
            response = await self._call_llm(request)
            
            # Validate response
            validated = self._validate(response)
            
            # Record metrics
            latency = (time.time() - start_time) * 1000
            
            return AIResponse(
                request_id=request.id,
                output=validated,
                model_used=self.primary_model,
                latency_ms=latency,
                tokens_used=0,  # from actual response
                cost_usd=0.0    # calculate based on tokens
            )
            
        except Exception as e:
            # Fallback handling
            return await self._handle_fallback(request, e)
    
    async def _call_llm(self, request: AIRequest) -> str:
        response = await self.client.chat.completions.create(
            model=self.primary_model,
            messages=[{"role": "user", "content": request.input}],
            max_tokens=2048
        )
        return response.choices[0].message.content or ""
    
    async def _handle_fallback(self, request: AIRequest, error: Exception) -> AIResponse:
        """Handle failures with fallback strategy."""
        # Log the error
        print(f"Primary failed: {error}. Using fallback.")
        
        # Try fallback
        try:
            response = await self.client.chat.completions.create(
                model=self.fallback_model,
                messages=[{"role": "user", "content": request.input}]
            )
            return AIResponse(
                request_id=request.id,
                output=response.choices[0].message.content or "",
                model_used=self.fallback_model,
                latency_ms=0,
                tokens_used=0,
                cost_usd=0
            )
        except Exception as e2:
            return AIResponse(
                request_id=request.id,
                output="",
                model_used="none",
                latency_ms=0,
                tokens_used=0,
                cost_usd=0,
                status=ProcessingStatus.FAILED
            )
    
    def _validate(self, response: str) -> str:
        """Validate and sanitize AI response."""
        if not response:
            raise ValueError("Empty response received")
        if len(response) > 50000:
            response = response[:50000]
        return response.strip()
Usage
processor = ModelRoutingRulesEngine(config={
    "primary_model": "gpt-4o-mini",
    "fallback_model": "gpt-4o-mini",
    "max_retries": 3
})
async def main():
    request = AIRequest(input="Test the Model Routing Rules Engine implementation")
    response = await processor.process(request)
    print(f"Response: {response.output[:200]}")
    print(f"Latency: {response.latency_ms:.0f}ms")
    print(f"Status: {response.status}")
asyncio.run(main())

Production Deployment

yaml
kubernetes deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ai-service
  template:
    spec:
      containers:
        - name: ai-service
          image: your-registry/ai-service:latest
          env:
            - name: OPENAI_API_KEY
              valueFrom:
                secretKeyRef:
                  name: ai-secrets
                  key: openai-key
          resources:
            requests:
              memory: "512Mi"
              cpu: "250m"
            limits:
              memory: "1Gi"
              cpu: "500m"
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 30

Monitoring

python
from prometheus_client import Counter, Histogram, Gauge
Metrics for Model Routing Rules Engine
requests_total = Counter('ai_requests_total', 'Total AI requests', ['model', 'status'])
request_duration = Histogram('ai_request_duration_seconds', 'Request duration')
active_requests = Gauge('ai_active_requests', 'Currently active requests')
def monitor(func):
    async def wrapper(*args, **kwargs):
        active_requests.inc()
        with request_duration.time():
            try:
                result = await func(*args, **kwargs)
                requests_total.labels(model='gpt-4o-mini', status='success').inc()
                return result
            except Exception as e:
                requests_total.labels(model='gpt-4o-mini', status='error').inc()
                raise
            finally:
                active_requests.dec()
    return wrapper

Key Design Decisions

Conclusion

The Model Routing Rules Engine provides a solid foundation for intelligently routing requests to optimal models in production. By following these patterns, you'll build AI systems that are reliable, observable, and cost-effective.

---

*Model Routing Rules Engine architecture guide | May 2026*