"Multi-Model Orchestration Patterns: Route AI Requests Like a Pro"

from openai import OpenAI

client = OpenAI(
    api_key="your-crazyrouter-api-key",
    base_url="https://crazyrouter.com/v1"
)

# Complexity classifier (can be rule-based or ML-based)
def classify_complexity(message: str) -> str:
    """Classify request complexity as low, medium, or high."""
    # Simple heuristics — replace with a classifier in production
    word_count = len(message.split())
    
    if word_count < 20 and "?" in message:
        return "low"
    elif any(kw in message.lower() for kw in ["analyze", "compare", "design", "architect", "refactor"]):
        return "high"
    elif word_count > 200:
        return "high"
    else:
        return "medium"

MODEL_MAP = {
    "low": "gpt-4.1-nano",       # $0.10/M input — simple Q&A
    "medium": "gpt-4.1-mini",    # $0.40/M input — standard tasks
    "high": "gpt-4.1",           # $2.00/M input — complex reasoning
}

def route_request(messages):
    user_message = messages[-1]["content"]
    complexity = classify_complexity(user_message)
    model = MODEL_MAP[complexity]
    
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    
    return {
        "response": response,
        "model_used": model,
        "complexity": complexity,
        "cost_tier": complexity
    }

import numpy as np
from sklearn.ensemble import RandomForestClassifier

# Train on historical data: (request_features) -> best_model
# Features: word_count, has_code, has_question, topic_embedding, etc.

class ModelRouter:
    def __init__(self):
        self.classifier = RandomForestClassifier()
        self.models = ["gpt-4.1-nano", "gpt-4.1-mini", "gpt-4.1", "claude-sonnet-4-5"]
    
    def extract_features(self, message):
        return [
            len(message.split()),           # word count
            int("```" in message),          # has code
            int("?" in message),            # is question
            len(message),                   # char count
            message.count("\n"),            # line count
            int(any(kw in message.lower()   # has complex keywords
                for kw in ["analyze", "design", "compare", "explain"]))
        ]
    
    def route(self, message):
        features = np.array([self.extract_features(message)])
        model_idx = self.classifier.predict(features)[0]
        return self.models[model_idx]

TASK_ROUTES = {
    "code": {
        "model": "gpt-4.1",
        "system": "You are an expert programmer. Write clean, efficient code.",
        "temperature": 0.2
    },
    "creative": {
        "model": "claude-sonnet-4-5",
        "system": "You are a creative writer with a vivid imagination.",
        "temperature": 0.8
    },
    "analysis": {
        "model": "claude-sonnet-4-5",
        "system": "You are a precise analyst. Be thorough and data-driven.",
        "temperature": 0.3
    },
    "translation": {
        "model": "gpt-4.1-mini",
        "system": "You are a professional translator.",
        "temperature": 0.1
    },
    "math": {
        "model": "o4-mini",
        "system": "Solve step by step.",
        "temperature": 0.0
    },
    "chat": {
        "model": "gpt-4.1-nano",
        "system": "You are a helpful assistant.",
        "temperature": 0.7
    }
}

def detect_task_type(message: str) -> str:
    """Detect the task type from the user message."""
    message_lower = message.lower()
    
    if any(kw in message_lower for kw in ["write code", "function", "implement", "debug", "```"]):
        return "code"
    elif any(kw in message_lower for kw in ["write a story", "creative", "poem", "imagine"]):
        return "creative"
    elif any(kw in message_lower for kw in ["analyze", "compare", "evaluate", "review"]):
        return "analysis"
    elif any(kw in message_lower for kw in ["translate", "翻译", "traduire"]):
        return "translation"
    elif any(kw in message_lower for kw in ["calculate", "solve", "equation", "math"]):
        return "math"
    else:
        return "chat"

def route_by_task(user_message):
    task_type = detect_task_type(user_message)
    config = TASK_ROUTES[task_type]
    
    response = client.chat.completions.create(
        model=config["model"],
        messages=[
            {"role": "system", "content": config["system"]},
            {"role": "user", "content": user_message}
        ],
        temperature=config["temperature"]
    )
    
    return response, task_type

import re

COST_CASCADE = [
    {"model": "gpt-4.1-nano", "cost_per_1k": 0.0001},
    {"model": "gpt-4.1-mini", "cost_per_1k": 0.0004},
    {"model": "gpt-4.1", "cost_per_1k": 0.002},
]

def quality_check(response_text: str, task_type: str) -> bool:
    """Check if the response meets quality thresholds."""
    # Basic quality heuristics
    if len(response_text.strip()) < 20:
        return False
    if "I don't know" in response_text or "I'm not sure" in response_text:
        return False
    if task_type == "code" and "```" not in response_text:
        return False  # Code task should contain code blocks
    return True

def cascade_request(messages, task_type="general"):
    for tier in COST_CASCADE:
        response = client.chat.completions.create(
            model=tier["model"],
            messages=messages
        )
        
        content = response.choices[0].message.content
        
        if quality_check(content, task_type):
            return {
                "content": content,
                "model": tier["model"],
                "escalated": tier != COST_CASCADE[0]
            }
        
        print(f"{tier['model']} response insufficient, escalating...")
    
    # Return last response even if quality check failed
    return {
        "content": content,
        "model": COST_CASCADE[-1]["model"],
        "escalated": True
    }

import random
import hashlib
from datetime import datetime

class ModelABTest:
    def __init__(self, variants):
        """
        variants: [
            {"model": "gpt-4.1", "weight": 0.5},
            {"model": "claude-sonnet-4-5", "weight": 0.5}
        ]
        """
        self.variants = variants
        self.results = {v["model"]: {"count": 0, "latency_sum": 0, "errors": 0}
                       for v in variants}
    
    def select_variant(self, user_id: str = None):
        """Select a model variant. Consistent per user if user_id provided."""
        if user_id:
            # Deterministic assignment based on user ID
            hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
            threshold = 0
            for variant in self.variants:
                threshold += variant["weight"]
                if (hash_val % 100) / 100 < threshold:
                    return variant["model"]
        
        # Random assignment
        r = random.random()
        threshold = 0
        for variant in self.variants:
            threshold += variant["weight"]
            if r < threshold:
                return variant["model"]
        
        return self.variants[-1]["model"]
    
    def record(self, model, latency_ms, success=True):
        self.results[model]["count"] += 1
        self.results[model]["latency_sum"] += latency_ms
        if not success:
            self.results[model]["errors"] += 1
    
    def report(self):
        for model, stats in self.results.items():
            avg_latency = stats["latency_sum"] / max(stats["count"], 1)
            error_rate = stats["errors"] / max(stats["count"], 1)
            print(f"{model}: {stats['count']} calls, "
                  f"avg {avg_latency:.0f}ms, "
                  f"error rate {error_rate:.1%}")

# Usage
ab_test = ModelABTest([
    {"model": "gpt-4.1", "weight": 0.5},
    {"model": "claude-sonnet-4-5", "weight": 0.5}
])

model = ab_test.select_variant(user_id="user_123")

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key="your-crazyrouter-api-key",
    base_url="https://crazyrouter.com/v1"
)

async def ensemble_request(messages, models=None):
    """Query multiple models and return consensus."""
    models = models or ["gpt-4.1", "claude-sonnet-4-5", "gemini-2.5-flash"]
    
    async def query_model(model):
        try:
            response = await async_client.chat.completions.create(
                model=model,
                messages=messages
            )
            return {"model": model, "content": response.choices[0].message.content}
        except Exception as e:
            return {"model": model, "error": str(e)}
    
    # Query all models in parallel
    results = await asyncio.gather(*[query_model(m) for m in models])
    
    # Filter successful responses
    successful = [r for r in results if "content" in r]
    
    if not successful:
        raise Exception("All models failed")
    
    # For classification tasks: majority vote
    # For generation tasks: return all and let the application choose
    return {
        "responses": successful,
        "count": len(successful),
        "models_used": [r["model"] for r in successful]
    }

class AIOrchestrator:
    def __init__(self, api_key, base_url="https://crazyrouter.com/v1"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.router = ModelRouter()
        self.circuit_breaker = CircuitBreaker()
        self.ab_test = None  # Optional
    
    def complete(self, messages, strategy="auto", **kwargs):
        """
        Main entry point for AI completions.
        
        Strategies:
        - "auto": Complexity-based routing
        - "cheap": Always use cheapest model
        - "best": Always use best model
        - "cascade": Start cheap, escalate if needed
        - "specific": Use kwargs["model"]
        """
        if strategy == "specific":
            model = kwargs["model"]
        elif strategy == "cheap":
            model = "gpt-4.1-nano"
        elif strategy == "best":
            model = "gpt-4.1"
        elif strategy == "cascade":
            return self._cascade(messages, kwargs.get("task_type", "general"))
        else:  # auto
            model = self.router.route(messages[-1]["content"])
        
        return self._call_with_fallback(messages, model)
    
    def _call_with_fallback(self, messages, primary_model):
        fallback_models = self._get_fallbacks(primary_model)
        
        for model in [primary_model] + fallback_models:
            if not self.circuit_breaker.can_execute(model):
                continue
            try:
                response = self.client.chat.completions.create(
                    model=model, messages=messages
                )
                self.circuit_breaker.record_success(model)
                return response
            except Exception as e:
                self.circuit_breaker.record_failure(model)
        
        raise Exception("All models unavailable")
    
    def _get_fallbacks(self, model):
        FALLBACKS = {
            "gpt-4.1": ["claude-sonnet-4-5", "gemini-2.5-flash"],
            "claude-sonnet-4-5": ["gpt-4.1", "gemini-2.5-flash"],
            "gemini-2.5-flash": ["gpt-4.1-mini", "deepseek-v3"],
            "gpt-4.1-mini": ["deepseek-v3", "gpt-4.1-nano"],
            "gpt-4.1-nano": ["deepseek-v3"],
        }
        return FALLBACKS.get(model, ["gpt-4.1-mini"])