Multi-Model Orchestration: Combining GPT, Claude, Llama, and Open Source Models

from dataclasses import dataclass
from enum import Enum
from typing import Optional

class ModelProvider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    OPEN_SOURCE = "open_source"
    TOGETHER = "together"

@dataclass
class ModelProfile:
    """Profile of an LLM for routing decisions"""
    name: str
    provider: ModelProvider
    cost_per_1k_tokens: float  # Input tokens
    latency_ms: int  # Average latency
    max_context_tokens: int
    reasoning_capability: float  # 0-1 scale
    coding_capability: float  # 0-1 scale
    reliability: float  # 0-1 scale (uptime)
    max_concurrent_requests: int
    
    def calculate_score(self, task_type: str, 
                       budget_priority: float = 0.3,
                       latency_priority: float = 0.3,
                       quality_priority: float = 0.4) -> float:
        """Calculate model suitability score"""
        
        # Normalize metrics
        cost_score = 1 - (self.cost_per_1k_tokens / 0.1)  # Normalize to $0.1
        latency_score = 1 - (self.latency_ms / 5000)  # Normalize to 5s
        
        # Task-specific quality scores
        if task_type == "reasoning":
            quality_score = self.reasoning_capability
        elif task_type == "coding":
            quality_score = self.coding_capability
        else:
            quality_score = (self.reasoning_capability + self.coding_capability) / 2
        
        # Weighted score
        total_score = (
            cost_score * budget_priority +
            latency_score * latency_priority +
            quality_score * quality_priority
        )
        
        return total_score

# Define model profiles
MODELS = {
    'gpt-4': ModelProfile(
        name='GPT-4',
        provider=ModelProvider.OPENAI,
        cost_per_1k_tokens=0.03,
        latency_ms=2000,
        max_context_tokens=128000,
        reasoning_capability=0.95,
        coding_capability=0.92,
        reliability=0.999,
        max_concurrent_requests=10000
    ),
    'gpt-3.5': ModelProfile(
        name='GPT-3.5 Turbo',
        provider=ModelProvider.OPENAI,
        cost_per_1k_tokens=0.0005,
        latency_ms=800,
        max_context_tokens=16000,
        reasoning_capability=0.75,
        coding_capability=0.80,
        reliability=0.999,
        max_concurrent_requests=50000
    ),
    'claude-3-opus': ModelProfile(
        name='Claude 3 Opus',
        provider=ModelProvider.ANTHROPIC,
        cost_per_1k_tokens=0.015,
        latency_ms=1500,
        max_context_tokens=200000,
        reasoning_capability=0.92,
        coding_capability=0.88,
        reliability=0.998,
        max_concurrent_requests=5000
    ),
    'claude-3-sonnet': ModelProfile(
        name='Claude 3 Sonnet',
        provider=ModelProvider.ANTHROPIC,
        cost_per_1k_tokens=0.003,
        latency_ms=1000,
        max_context_tokens=200000,
        reasoning_capability=0.85,
        coding_capability=0.82,
        reliability=0.998,
        max_concurrent_requests=10000
    ),
    'llama-2-70b': ModelProfile(
        name='Llama 2 70B',
        provider=ModelProvider.OPEN_SOURCE,
        cost_per_1k_tokens=0.0008,
        latency_ms=1200,
        max_context_tokens=4096,
        reasoning_capability=0.70,
        coding_capability=0.75,
        reliability=0.95,
        max_concurrent_requests=100
    ),
    'mixtral-8x7b': ModelProfile(
        name='Mixtral 8x7B',
        provider=ModelProvider.OPEN_SOURCE,
        cost_per_1k_tokens=0.0005,
        latency_ms=800,
        max_context_tokens=32000,
        reasoning_capability=0.72,
        coding_capability=0.78,
        reliability=0.94,
        max_concurrent_requests=200
    ),
}

from typing import Dict, List
import json

class ModelRouter:
    """Intelligent router for multi-model orchestration"""
    
    def __init__(self, models: Dict[str, ModelProfile]):
        self.models = models
        self.request_history = []
        self.model_metrics = {name: {
            'requests': 0,
            'errors': 0,
            'total_latency': 0,
            'total_cost': 0
        } for name in models}
    
    def classify_task(self, prompt: str, context_length: int) -> Dict:
        """Classify task complexity and requirements"""
        
        # Analyze prompt characteristics
        is_reasoning = any(keyword in prompt.lower() 
                          for keyword in ['explain', 'why', 'analyze', 'reason'])
        is_coding = any(keyword in prompt.lower() 
                       for keyword in ['code', 'function', 'implement', 'debug'])
        is_long_context = context_length > 50000
        
        complexity = 'simple'
        if is_reasoning or is_coding:
            complexity = 'complex'
        if is_long_context:
            complexity = 'very_complex'
        
        return {
            'is_reasoning': is_reasoning,
            'is_coding': is_coding,
            'is_long_context': is_long_context,
            'complexity': complexity,
            'context_length': context_length
        }
    
    def select_model(self, prompt: str, context_length: int,
                    budget_priority: float = 0.3,
                    latency_priority: float = 0.3,
                    quality_priority: float = 0.4) -> str:
        """Select best model for request"""
        
        # Classify task
        task = self.classify_task(prompt, context_length)
        
        # Filter models by requirements
        candidates = []
        for model_name, profile in self.models.items():
            # Check context length requirement
            if context_length > profile.max_context_tokens:
                continue
            
            # Calculate suitability score
            task_type = 'reasoning' if task['is_reasoning'] else 'coding' if task['is_coding'] else 'general'
            score = profile.calculate_score(
                task_type,
                budget_priority=budget_priority,
                latency_priority=latency_priority,
                quality_priority=quality_priority
            )
            
            candidates.append((model_name, score))
        
        if not candidates:
            # Fallback to most capable model
            return 'gpt-4'
        
        # Select highest scoring model
        best_model = max(candidates, key=lambda x: x[1])[0]
        return best_model
    
    def route_request(self, prompt: str, context_length: int,
                     budget_priority: float = 0.3,
                     latency_priority: float = 0.3,
                     quality_priority: float = 0.4) -> Dict:
        """Route request to best model"""
        
        selected_model = self.select_model(
            prompt, context_length,
            budget_priority, latency_priority, quality_priority
        )
        
        return {
            'model': selected_model,
            'provider': self.models[selected_model].provider.value,
            'cost_estimate': self.models[selected_model].cost_per_1k_tokens,
            'latency_estimate': self.models[selected_model].latency_ms
        }
    
    def record_request(self, model_name: str, latency_ms: int,
                      tokens_used: int, cost: float, success: bool):
        """Record request metrics"""
        
        metrics = self.model_metrics[model_name]
        metrics['requests'] += 1
        metrics['total_latency'] += latency_ms
        metrics['total_cost'] += cost
        
        if not success:
            metrics['errors'] += 1
    
    def get_model_stats(self, model_name: str) -> Dict:
        """Get statistics for model"""
        
        metrics = self.model_metrics[model_name]
        
        if metrics['requests'] == 0:
            return {}
        
        return {
            'total_requests': metrics['requests'],
            'error_rate': metrics['errors'] / metrics['requests'],
            'avg_latency': metrics['total_latency'] / metrics['requests'],
            'total_cost': metrics['total_cost'],
            'avg_cost_per_request': metrics['total_cost'] / metrics['requests']
        }

# Usage
router = ModelRouter(MODELS)

# Route simple task (prioritize cost)
result = router.route_request(
    "What is 2+2?",
    context_length=100,
    budget_priority=0.7,
    latency_priority=0.2,
    quality_priority=0.1
)
print(f"Simple task: {result['model']}")  # Likely GPT-3.5 or Mixtral

# Route complex reasoning (prioritize quality)
result = router.route_request(
    "Explain quantum entanglement and its implications for cryptography",
    context_length=5000,
    budget_priority=0.1,
    latency_priority=0.2,
    quality_priority=0.7
)
print(f"Complex task: {result['model']}")  # Likely GPT-4 or Claude

import asyncio
from typing import Optional, Callable

class ResilientMultiModelClient:
    """Client with fallback strategies for reliability"""
    
    def __init__(self, router: ModelRouter, 
                 model_clients: Dict[str, Callable]):
        self.router = router
        self.model_clients = model_clients
        self.fallback_chain = [
            'gpt-4',
            'claude-3-opus',
            'gpt-3.5',
            'claude-3-sonnet',
            'mixtral-8x7b'
        ]
    
    async def call_with_fallback(self, prompt: str,
                                context_length: int,
                                max_retries: int = 3) -> Optional[str]:
        """Call model with automatic fallback"""
        
        # Get primary model
        primary_model = self.router.select_model(prompt, context_length)
        
        # Build fallback chain starting with primary
        fallback_chain = [primary_model] + [
            m for m in self.fallback_chain if m != primary_model
        ]
        
        for attempt, model_name in enumerate(fallback_chain):
            if attempt >= max_retries:
                break
            
            try:
                client = self.model_clients[model_name]
                response = await client(prompt)
                
                # Record success
                self.router.record_request(
                    model_name, 
                    latency_ms=100,  # Actual latency
                    tokens_used=len(prompt.split()),
                    cost=0.001,  # Actual cost
                    success=True
                )
                
                return response
            
            except Exception as e:
                print(f"Model {model_name} failed: {e}")
                
                # Record failure
                self.router.record_request(
                    model_name,
                    latency_ms=100,
                    tokens_used=len(prompt.split()),
                    cost=0.001,
                    success=False
                )
                
                # Try next model
                continue
        
        return None
    
    async def call_parallel(self, prompt: str,
                           num_models: int = 2) -> str:
        """Call multiple models in parallel for consensus"""
        
        # Select top N models
        candidates = []
        for model_name, profile in self.router.models.items():
            score = profile.calculate_score('general')
            candidates.append((model_name, score))
        
        top_models = sorted(candidates, key=lambda x: x[1], reverse=True)[:num_models]
        
        # Call models in parallel
        tasks = [
            self.model_clients[model_name](/ai/prompt)
            for model_name, _ in top_models
        ]
        
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Return first successful response
        for response in responses:
            if not isinstance(response, Exception):
                return response
        
        return None

# Usage
client = ResilientMultiModelClient(router, model_clients)

# Call with automatic fallback
response = await client.call_with_fallback(
    "Explain quantum computing",
    context_length=1000
)

# Call multiple models for consensus
response = await client.call_parallel(
    "Is this code secure?",
    num_models=2
)

class CostOptimizer:
    """Optimize costs through intelligent model selection"""
    
    def __init__(self, router: ModelRouter, budget_per_month: float):
        self.router = router
        self.budget_per_month = budget_per_month
        self.current_spend = 0
        self.requests_processed = 0
    
    def get_budget_remaining(self) -> float:
        """Get remaining budget"""
        return self.budget_per_month - self.current_spend
    
    def should_use_expensive_model(self, task_complexity: str) -> bool:
        """Decide if expensive model is justified"""
        
        budget_remaining = self.get_budget_remaining()
        budget_percentage = budget_remaining / self.budget_per_month
        
        # Use expensive models only if budget allows
        if task_complexity == 'simple':
            return False
        elif task_complexity == 'complex':
            return budget_percentage > 0.3
        else:  # very_complex
            return budget_percentage > 0.1
    
    def select_cost_optimized_model(self, prompt: str,
                                   context_length: int) -> str:
        """Select model optimizing for cost"""
        
        task = self.router.classify_task(prompt, context_length)
        
        if not self.should_use_expensive_model(task['complexity']):
            # Use cheapest model
            cheapest = min(
                self.router.models.items(),
                key=lambda x: x[1].cost_per_1k_tokens
            )
            return cheapest[0]
        
        # Use quality-optimized selection
        return self.router.select_model(
            prompt, context_length,
            budget_priority=0.6,
            latency_priority=0.2,
            quality_priority=0.2
        )
    
    def estimate_monthly_cost(self, requests_per_day: int,
                             avg_tokens_per_request: int) -> float:
        """Estimate monthly cost"""
        
        total_requests = requests_per_day * 30
        total_tokens = total_requests * avg_tokens_per_request
        
        # Estimate using average model cost
        avg_cost = sum(m.cost_per_1k_tokens for m in self.router.models.values()) / len(self.router.models)
        
        return (total_tokens / 1000) * avg_cost

# Usage
optimizer = CostOptimizer(router, budget_per_month=1000)

# Select cost-optimized model
model = optimizer.select_cost_optimized_model(
    "What is the capital of France?",
    context_length=100
)

# Estimate costs
estimated_cost = optimizer.estimate_monthly_cost(
    requests_per_day=10000,
    avg_tokens_per_request=500
)
print(f"Estimated monthly cost: ${estimated_cost:.2f}")

class IntelligentModelSelector:
    """Select best model for each task"""
    
    def __init__(self):
        self.models = {
            'gpt-4': {'cost': 0.06, 'speed': 8, 'quality': 9.5},
            'gpt-3.5': {'cost': 0.0015, 'speed': 9, 'quality': 8.5},
            'claude-3': {'cost': 0.015, 'speed': 8, 'quality': 9},
            'llama-2': {'cost': 0.001, 'speed': 7, 'quality': 8}
        }
        self.task_requirements = {
            'summarization': {'quality': 0.8, 'speed': 0.9, 'cost': 0.3},
            'coding': {'quality': 0.95, 'speed': 0.7, 'cost': 0.2},
            'translation': {'quality': 0.9, 'speed': 0.8, 'cost': 0.3},
            'chat': {'quality': 0.8, 'speed': 0.95, 'cost': 0.4}
        }
    
    def select_model(self, task: str) -> str:
        """Select best model for task"""
        
        if task not in self.task_requirements:
            return 'gpt-3.5'  # Default
        
        requirements = self.task_requirements[task]
        best_model = None
        best_score = -1
        
        for model_name, specs in self.models.items():
            # Calculate score based on requirements
            score = (
                (specs['quality'] / 10) * requirements['quality'] +
                (specs['speed'] / 10) * requirements['speed'] +
                (1 - specs['cost'] / 0.06) * requirements['cost']
            )
            
            if score > best_score:
                best_score = score
                best_model = model_name
        
        return best_model

class FallbackOrchestrator:
    """Handle model failures with fallbacks"""
    
    def __init__(self):
        self.primary_model = 'gpt-4'
        self.fallback_models = ['claude-3', 'gpt-3.5', 'llama-2']
        self.max_retries = 3
    
    async def query_with_fallback(self, prompt: str) -> str:
        """Query with automatic fallback"""
        
        models_to_try = [self.primary_model] + self.fallback_models
        
        for model in models_to_try:
            for attempt in range(self.max_retries):
                try:
                    response = await self._call_model(model, prompt)
                    return response
                except Exception as e:
                    print(f"Error with {model}: {e}")
                    if attempt == self.max_retries - 1:
                        break
        
        return "Unable to generate response"
    
    async def _call_model(self, model: str, prompt: str) -> str:
        """Call specific model"""
        # Implementation
        pass

class MultiModelCostOptimizer:
    """Optimize costs across multiple models"""
    
    def __init__(self):
        self.model_costs = {
            'gpt-4': 0.06,
            'gpt-3.5': 0.0015,
            'claude-3': 0.015,
            'llama-2': 0.001
        }
    
    def estimate_cost(self, task: str, tokens: int) -> dict:
        """Estimate cost for task"""
        
        costs = {}
        for model, cost_per_1k in self.model_costs.items():
            total_cost = (tokens * cost_per_1k) / 1000
            costs[model] = total_cost
        
        return costs
    
    def get_cheapest_model(self, task: str, tokens: int) -> str:
        """Get cheapest model for task"""
        
        costs = self.estimate_cost(task, tokens)
        return min(costs, key=costs.get)

import time
import json
from typing import Dict, List

class ModelBenchmark:
    """Benchmark LLM models"""
    
    def __init__(self):
        self.results = {}
        self.test_queries = [
            "Explain quantum computing",
            "Write a Python function for binary search",
            "Summarize the history of AI",
            "Translate 'Hello world' to Spanish",
            "Solve: 2x + 5 = 15"
        ]
    
    def benchmark_models(self, models: Dict[str, any]) -> Dict:
        """Benchmark multiple models"""
        
        benchmark_results = {}
        
        for model_name, model_client in models.items():
            print(f"Benchmarking {model_name}...")
            
            model_results = {
                'latency': [],
                'quality': [],
                'cost': [],
                'tokens': []
            }
            
            for query in self.test_queries:
                # Measure latency
                start = time.time()
                response = model_client.complete(query)
                latency = time.time() - start
                
                model_results['latency'].append(latency)
                model_results['quality'].append(self._score_quality(response))
                model_results['cost'].append(self._estimate_cost(response))
                model_results['tokens'].append(len(response.split()))
            
            # Calculate averages
            benchmark_results[model_name] = {
                'avg_latency': sum(model_results['latency']) / len(model_results['latency']),
                'avg_quality': sum(model_results['quality']) / len(model_results['quality']),
                'avg_cost': sum(model_results['cost']) / len(model_results['cost']),
                'avg_tokens': sum(model_results['tokens']) / len(model_results['tokens'])
            }
        
        return benchmark_results
    
    def _score_quality(self, response: str) -> float:
        """Score response quality (0-1)"""
        # Simple heuristic: longer, more detailed responses score higher
        return min(1.0, len(response) / 1000)
    
    def _estimate_cost(self, response: str) -> float:
        """Estimate cost of response"""
        tokens = len(response.split())
        # Rough estimate: $0.0015 per 1K tokens
        return (tokens * 0.0015) / 1000
    
    def print_comparison(self, results: Dict):
        """Print benchmark comparison"""
        
        print("\n" + "="*60)
        print("Model Benchmark Results")
        print("="*60)
        
        for model, metrics in results.items():
            print(f"\n{model}:")
            print(f"  Avg Latency: {metrics['avg_latency']:.3f}s")
            print(f"  Avg Quality: {metrics['avg_quality']:.2f}/1.0")
            print(f"  Avg Cost: ${metrics['avg_cost']:.4f}")
            print(f"  Avg Tokens: {metrics['avg_tokens']:.0f}")

class PerformanceMonitor:
    """Monitor model performance in real-time"""
    
    def __init__(self):
        self.metrics = {
            'requests': 0,
            'errors': 0,
            'total_latency': 0,
            'total_tokens': 0,
            'total_cost': 0
        }
        self.model_metrics = {}
    
    def record_request(self, model: str, latency: float, 
                      tokens: int, cost: float, success: bool = True):
        """Record request metrics"""
        
        self.metrics['requests'] += 1
        self.metrics['total_latency'] += latency
        self.metrics['total_tokens'] += tokens
        self.metrics['total_cost'] += cost
        
        if not success:
            self.metrics['errors'] += 1
        
        # Track per-model metrics
        if model not in self.model_metrics:
            self.model_metrics[model] = {
                'requests': 0,
                'errors': 0,
                'total_latency': 0,
                'total_cost': 0
            }
        
        self.model_metrics[model]['requests'] += 1
        self.model_metrics[model]['total_latency'] += latency
        self.model_metrics[model]['total_cost'] += cost
        
        if not success:
            self.model_metrics[model]['errors'] += 1
    
    def get_stats(self) -> Dict:
        """Get performance statistics"""
        
        total_requests = self.metrics['requests']
        
        stats = {
            'total_requests': total_requests,
            'error_rate': self.metrics['errors'] / total_requests if total_requests > 0 else 0,
            'avg_latency': self.metrics['total_latency'] / total_requests if total_requests > 0 else 0,
            'avg_tokens': self.metrics['total_tokens'] / total_requests if total_requests > 0 else 0,
            'total_cost': self.metrics['total_cost'],
            'models': {}
        }
        
        for model, metrics in self.model_metrics.items():
            model_requests = metrics['requests']
            stats['models'][model] = {
                'requests': model_requests,
                'error_rate': metrics['errors'] / model_requests if model_requests > 0 else 0,
                'avg_latency': metrics['total_latency'] / model_requests if model_requests > 0 else 0,
                'avg_cost': metrics['total_cost'] / model_requests if model_requests > 0 else 0
            }
        
        return stats
    
    def print_report(self):
        """Print performance report"""
        
        stats = self.get_stats()
        
        print("\n" + "="*60)
        print("Performance Report")
        print("="*60)
        print(f"Total Requests: {stats['total_requests']}")
        print(f"Error Rate: {stats['error_rate']:.2%}")
        print(f"Avg Latency: {stats['avg_latency']:.3f}s")
        print(f"Total Cost: ${stats['total_cost']:.2f}")
        
        print("\nPer-Model Stats:")
        for model, metrics in stats['models'].items():
            print(f"\n{model}:")
            print(f"  Requests: {metrics['requests']}")
            print(f"  Error Rate: {metrics['error_rate']:.2%}")
            print(f"  Avg Latency: {metrics['avg_latency']:.3f}s")
            print(f"  Avg Cost: ${metrics['avg_cost']:.4f}")

class ContextAwareRouter:
    """Route requests based on context"""
    
    def __init__(self):
        self.model_specializations = {
            'gpt-4': ['complex_reasoning', 'code_generation', 'analysis'],
            'gpt-3.5': ['general_qa', 'summarization', 'chat'],
            'claude-3': ['writing', 'analysis', 'research'],
            'llama-2': ['general_qa', 'chat']
        }
    
    def route_request(self, query: str, context: Dict = None) -> str:
        """Route request to best model"""
        
        # Analyze query
        query_type = self._classify_query(query)
        
        # Get context
        if context is None:
            context = {}
        
        # Find best model
        best_model = self._find_best_model(query_type, context)
        
        return best_model
    
    def _classify_query(self, query: str) -> str:
        """Classify query type"""
        
        keywords = {
            'code': ['code', 'function', 'algorithm', 'program'],
            'writing': ['write', 'essay', 'article', 'story'],
            'analysis': ['analyze', 'explain', 'compare', 'evaluate'],
            'chat': ['hello', 'how are you', 'what is', 'tell me']
        }
        
        query_lower = query.lower()
        
        for query_type, words in keywords.items():
            if any(word in query_lower for word in words):
                return query_type
        
        return 'general_qa'
    
    def _find_best_model(self, query_type: str, context: Dict) -> str:
        """Find best model for query type"""
        
        # Check budget constraint
        if context.get('budget') == 'low':
            return 'gpt-3.5'
        
        # Check latency constraint
        if context.get('latency') == 'critical':
            return 'gpt-3.5'
        
        # Find model with specialization
        for model, specializations in self.model_specializations.items():
            if query_type in specializations:
                return model
        
        return 'gpt-3.5'  # Default