Multi-Model AI Workflows: How to Use Different AI Models for Different Tasks

# Cascading Model Implementation

class CascadingAIWorkflow:
    def __init__(self):
        self.models = [
            {'name': 'gpt-3.5-turbo', 'cost': 0.001, 'capability': 0.7},
            {'name': 'gpt-4', 'cost': 0.015, 'capability': 0.9},
            {'name': 'gpt-4-turbo', 'cost': 0.020, 'capability': 0.95}
        ]
    
    def cascade_process(self, prompt: str, quality_threshold: float = 0.8):
        """Try cheaper models first, escalate if quality insufficient"""
        
        for model in self.models:
            response = self.query_model(model['name'], prompt)
            quality_score = self.assess_quality(response)
            
            if quality_score >= quality_threshold:
                return {
                    'response': response,
                    'model_used': model['name'],
                    'cost': model['cost'],
                    'quality': quality_score
                }
                
        # If we get here, even the best model didn't meet threshold
        return {
            'error': 'Quality threshold not met',
            'best_attempt': response,
            'model_used': self.models[-1]['name']
        }
    
    def assess_quality(self, response: str) -> float:
        """Simple quality assessment (implement with your criteria)"""
        
        factors = {
            'completeness': len(response) > 100,  # Basic length check
            'coherence': not any(word in response.lower() for word in ['error', 'sorry', 'cannot']),
            'specificity': len([w for w in response.split() if len(w) > 6]) > 5
        }
        
        return sum(factors.values()) / len(factors)

# Parallel Processing for Quality and Speed

import asyncio
import aiohttp

class ParallelModelProcessor:
    def __init__(self):
        self.model_configs = {
            'speed': {'model': 'claude-haiku', 'timeout': 1.0},
            'quality': {'model': 'gpt-4-turbo', 'timeout': 5.0},
            'creative': {'model': 'claude-3-opus', 'timeout': 3.0}
        }
    
    async def parallel_process(self, prompt: str, approaches: list):
        """Process same prompt with multiple models in parallel"""
        
        tasks = []
        for approach in approaches:
            config = self.model_configs[approach]
            task = self.query_model_async(
                config['model'], 
                prompt, 
                timeout=config['timeout']
            )
            tasks.append((approach, task))
        
        results = {}
        completed_tasks = await asyncio.gather(*[task for _, task in tasks], return_exceptions=True)
        
        for (approach, _), result in zip(tasks, completed_tasks):
            if isinstance(result, Exception):
                results[approach] = {'error': str(result)}
            else:
                results[approach] = result
                
        return results
    
    def select_best_result(self, parallel_results: dict) -> dict:
        """Select best result based on criteria"""
        
        # Priority order: quality > creative > speed
        priority_order = ['quality', 'creative', 'speed']
        
        for approach in priority_order:
            if approach in parallel_results and 'error' not in parallel_results[approach]:
                return {
                    'selected_approach': approach,
                    'result': parallel_results[approach],
                    'all_results': parallel_results
                }
        
        # Fallback to first successful result
        for approach, result in parallel_results.items():
            if 'error' not in result:
                return {
                    'selected_approach': approach,
                    'result': result,
                    'all_results': parallel_results
                }
                
        return {'error': 'All approaches failed', 'all_results': parallel_results}

# Dynamic Cost Optimization

class CostOptimizedRouter:
    def __init__(self):
        # Current pricing (updates from API)
        self.current_pricing = {
            'gpt-3.5-turbo': {'input': 0.0010, 'output': 0.0020},
            'gpt-4': {'input': 0.0300, 'output': 0.0600},
            'gpt-4-turbo': {'input': 0.0100, 'output': 0.0300},
            'claude-3-haiku': {'input': 0.0008, 'output': 0.0024},
            'claude-3.5-sonnet': {'input': 0.0030, 'output': 0.0150},
            'gemini-pro': {'input': 0.0005, 'output': 0.0015}
        }
        
        # Performance benchmarks for different task types
        self.performance_scores = {
            'simple_qa': {
                'gpt-3.5-turbo': 0.85,
                'claude-3-haiku': 0.83,
                'gemini-pro': 0.81
            },
            'code_generation': {
                'gpt-4-turbo': 0.92,
                'claude-3.5-sonnet': 0.89,
                'gpt-4': 0.88
            },
            'creative_writing': {
                'gpt-4': 0.94,
                'claude-3-opus': 0.93,
                'gpt-4-turbo': 0.90
            }
        }
    
    def calculate_value_score(self, task_type: str, model: str) -> float:
        """Calculate value = performance / cost ratio"""
        
        performance = self.performance_scores.get(task_type, {}).get(model, 0)
        # Assume average of 1000 input + 500 output tokens
        cost = (1000 * self.current_pricing[model]['input'] + 
                500 * self.current_pricing[model]['output']) / 1000
                
        return performance / cost if cost > 0 else 0
    
    def optimize_model_selection(self, task_type: str, budget_limit: float = None) -> str:
        """Select model with best value for task type"""
        
        available_models = self.performance_scores.get(task_type, {}).keys()
        
        if budget_limit:
            # Filter models by budget
            available_models = [
                model for model in available_models
                if self.estimate_cost(model) <= budget_limit
            ]
        
        if not available_models:
            return 'gpt-3.5-turbo'  # Fallback to cheapest
            
        # Select model with highest value score
        best_model = max(available_models, 
                        key=lambda m: self.calculate_value_score(task_type, m))
        
        return best_model

# Intelligent Batch Processing

class BatchOptimizer:
    def __init__(self, daily_budget: float = 100.0):
        self.daily_budget = daily_budget
        self.current_spend = 0.0
        self.task_queue = []
        
    def add_task(self, prompt: str, priority: int, deadline: str, context: dict = None):
        """Add task to optimization queue"""
        
        task = {
            'prompt': prompt,
            'priority': priority,
            'deadline': datetime.fromisoformat(deadline),
            'context': context or {},
            'estimated_cost': self.estimate_task_cost(prompt)
        }
        
        self.task_queue.append(task)
        
    def optimize_batch(self) -> list:
        """Optimize entire batch for cost and performance"""
        
        # Sort by priority and deadline
        self.task_queue.sort(key=lambda t: (t['priority'], t['deadline']), reverse=True)
        
        optimized_batch = []
        remaining_budget = self.daily_budget - self.current_spend
        
        # Greedy optimization with budget constraint
        for task in self.task_queue:
            if remaining_budget <= 0:
                # Switch all remaining tasks to cheapest model
                task['model'] = 'gpt-3.5-turbo'
                task['reason'] = 'Budget exhausted'
            else:
                # Optimal model selection within budget
                optimal_model = self.select_optimal_model(task, remaining_budget)
                task['model'] = optimal_model
                remaining_budget -= task['estimated_cost']
            
            optimized_batch.append(task)
            
        return optimized_batch
        
    def schedule_processing(self, batch: list) -> dict:
        """Schedule batch processing for optimal resource utilization"""
        
        # Group by model for efficient batching
        model_groups = {}
        for task in batch:
            model = task['model']
            if model not in model_groups:
                model_groups[model] = []
            model_groups[model].append(task)
            
        # Schedule processing times
        schedule = {}
        current_time = datetime.now()
        
        for model, tasks in model_groups.items():
            # Estimate processing time for batch
            batch_time = len(tasks) * self.get_avg_processing_time(model)
            schedule[model] = {
                'tasks': tasks,
                'start_time': current_time,
                'estimated_completion': current_time + timedelta(seconds=batch_time),
                'parallel_possible': len(tasks) > 5  # Worth parallelizing if >5 tasks
            }
            current_time += timedelta(seconds=batch_time)
            
        return schedule

{
  "model_performance": {
    "gpt-4-turbo": {
      "tasks_completed": 1247,
      "avg_response_time": "2.3s",
      "quality_score": 0.92,
      "cost_per_task": "$0.045",
      "user_satisfaction": 4.6
    },
    "claude-3.5-sonnet": {
      "tasks_completed": 856,
      "avg_response_time": "1.8s", 
      "quality_score": 0.89,
      "cost_per_task": "$0.032",
      "user_satisfaction": 4.4
    },
    "gpt-3.5-turbo": {
      "tasks_completed": 3421,
      "avg_response_time": "0.9s",
      "quality_score": 0.78,
      "cost_per_task": "$0.008",
      "user_satisfaction": 4.1
    }
  },
  
  "routing_efficiency": {
    "correct_routing_rate": 0.87,
    "cost_savings_vs_single_model": 0.63,
    "performance_improvement": 0.23,
    "user_override_rate": 0.12
  },
  
  "business_impact": {
    "monthly_cost_reduction": "$2,340",
    "productivity_improvement": "34%",
    "quality_consistency": "91%",
    "response_time_improvement": "45%"
  }
}