Multi-Model Strategies for Production AI

multi_model_benefits:
  reliability:
    - Provider outages don't stop you
    - Rate limits manageable
    - Redundancy for critical paths

  cost_optimization:
    - Right-size model to task
    - Expensive models for complex only
    - Cheaper models for volume

  capability_matching:
    - Different models excel at different tasks
    - Specialized models for specific domains
    - Best tool for each job

  risk_reduction:
    - No single vendor dependency
    - Pricing changes manageable
    - API changes isolated

class ComplexityRouter:
    """Route requests based on task complexity."""

    def __init__(self):
        self.models = {
            "simple": "gpt-3.5-turbo",
            "medium": "claude-3-sonnet",
            "complex": "claude-3-opus"
        }
        self.complexity_classifier = self._load_classifier()

    async def route(self, request: str) -> str:
        complexity = self._assess_complexity(request)

        if complexity < 0.3:
            return self.models["simple"]
        elif complexity < 0.7:
            return self.models["medium"]
        else:
            return self.models["complex"]

    def _assess_complexity(self, request: str) -> float:
        # Heuristics-based assessment
        signals = {
            "length": len(request) > 2000,
            "reasoning_words": any(w in request.lower() for w in
                ["analyze", "compare", "evaluate", "synthesize"]),
            "code": "```" in request or "function" in request,
            "multi_step": "then" in request.lower() or "steps" in request.lower()
        }
        return sum(signals.values()) / len(signals)

class CostTierRouter:
    """Route based on cost sensitivity and value."""

    PRICING = {
        "gpt-3.5-turbo": 0.002,
        "claude-3-haiku": 0.001,
        "claude-3-sonnet": 0.01,
        "gpt-4-turbo": 0.02,
        "claude-3-opus": 0.05
    }

    def route(self, request: dict) -> str:
        user_tier = request.get("user_tier", "free")
        task_value = request.get("task_value", "low")

        if user_tier == "enterprise" and task_value == "high":
            return "claude-3-opus"

        if user_tier in ["pro", "enterprise"]:
            return "claude-3-sonnet"

        # Free tier or low value
        return "claude-3-haiku"

class FallbackChain:
    """Try models in order until one succeeds."""

    def __init__(self, models: list[tuple[str, callable]]):
        self.chain = models

    async def generate(self, prompt: str) -> dict:
        for model_name, model_fn in self.chain:
            try:
                response = await model_fn(prompt)
                return {
                    "content": response,
                    "model": model_name,
                    "fallback_used": model_name != self.chain[0][0]
                }
            except Exception as e:
                logger.warning(f"{model_name} failed: {e}")
                continue

        raise AllModelsFailed("All models in chain failed")

# Usage
chain = FallbackChain([
    ("claude-3-sonnet", anthropic_client.generate),
    ("gpt-4-turbo", openai_client.generate),
    ("gpt-3.5-turbo", openai_client_35.generate),
])

class TaskSpecializedRouter:
    """Route to models based on task type."""

    TASK_MODELS = {
        "code_generation": "claude-3-sonnet",
        "code_review": "claude-3-opus",
        "summarization": "gpt-3.5-turbo",
        "translation": "gpt-4-turbo",
        "analysis": "claude-3-opus",
        "chat": "claude-3-haiku",
        "creative_writing": "claude-3-sonnet"
    }

    def route(self, task_type: str) -> str:
        return self.TASK_MODELS.get(task_type, "claude-3-sonnet")

class ABTestRouter:
    """Route for A/B testing model performance."""

    def __init__(self, experiment_config: dict):
        self.config = experiment_config
        self.results_store = ResultsStore()

    def route(self, request_id: str) -> str:
        # Deterministic assignment based on request_id
        bucket = hash(request_id) % 100

        if bucket < self.config["treatment_percentage"]:
            return self.config["treatment_model"]
        else:
            return self.config["control_model"]

    def record_outcome(self, request_id: str, outcome: dict):
        model = self.route(request_id)
        self.results_store.record(model, outcome)

    def get_results(self) -> dict:
        return self.results_store.analyze()

class UnifiedLLMClient:
    """Single interface for multiple LLM providers."""

    def __init__(self):
        self.providers = {
            "anthropic": AnthropicProvider(),
            "openai": OpenAIProvider(),
            "local": OllamaProvider()
        }
        self.router = SmartRouter()

    async def generate(
        self,
        prompt: str,
        model: str = None,
        **kwargs
    ) -> LLMResponse:
        # Auto-route if no model specified
        if model is None:
            model = await self.router.route(prompt, kwargs)

        provider = self._get_provider(model)
        response = await provider.generate(prompt, model, **kwargs)

        # Track for analytics
        await self._track(model, prompt, response)

        return response

    def _get_provider(self, model: str) -> Provider:
        if model.startswith("claude"):
            return self.providers["anthropic"]
        elif model.startswith("gpt"):
            return self.providers["openai"]
        else:
            return self.providers["local"]

# routing_config.yml
routing:
  default_model: "claude-3-sonnet"

  rules:
    - condition:
        task_type: "code"
        complexity: "high"
      model: "claude-3-opus"

    - condition:
        user_tier: "free"
      model: "claude-3-haiku"

    - condition:
        requires_vision: true
      model: "gpt-4-vision"

  fallbacks:
    - "claude-3-sonnet"
    - "gpt-4-turbo"
    - "gpt-3.5-turbo"

  experiments:
    new_model_test:
      treatment: "claude-3-opus"
      control: "gpt-4-turbo"
      percentage: 10

multi_model_metrics:
  per_model:
    - Success rate
    - Latency (p50, p95, p99)
    - Cost per request
    - Quality scores

  routing:
    - Model distribution
    - Fallback frequency
    - Routing accuracy

  overall:
    - Combined success rate
    - Total cost
    - Quality maintenance