114：多模型Router智能路由策略：根据任务动态选模型节省成本

# 智能路由核心算法
import numpy as np
import time
from datetime import datetime

class ModelRouter:
    def __init__(self):
        # 模型池
        self.models = {
            "gpt-3.5-turbo": {
                "cost_per_1k_tokens": 0.0015,
                "max_context": 4096,
                "speed": 0.9,  # 0-1，1表示最快
                "quality": 0.7,  # 0-1，1表示最高
                "capabilities": ["text", "conversation", "simple_code"]
            },
            "gpt-3.5-turbo-16k": {
                "cost_per_1k_tokens": 0.003,
                "max_context": 16384,
                "speed": 0.8,
                "quality": 0.75,
                "capabilities": ["text", "conversation", "simple_code", "longer_context"]
            },
            "gpt-4": {
                "cost_per_1k_tokens": 0.03,
                "max_context": 8192,
                "speed": 0.6,
                "quality": 0.95,
                "capabilities": ["text", "conversation", "code", "reasoning", "complex_tasks"]
            },
            "gpt-4-turbo": {
                "cost_per_1k_tokens": 0.01,
                "max_context": 128000,
                "speed": 0.7,
                "quality": 0.9,
                "capabilities": ["text", "conversation", "code", "reasoning", "complex_tasks", "long_context"]
            },
            "gpt-4o": {
                "cost_per_1k_tokens": 0.005,
                "max_context": 128000,
                "speed": 0.85,
                "quality": 0.92,
                "capabilities": ["text", "conversation", "code", "reasoning", "complex_tasks", "long_context", "multimodal"]
            },
            "llama-3-8b": {
                "cost_per_1k_tokens": 0.0005,
                "max_context": 128000,
                "speed": 0.95,
                "quality": 0.65,
                "capabilities": ["text", "conversation", "simple_code"]
            },
            "llama-3-70b": {
                "cost_per_1k_tokens": 0.002,
                "max_context": 128000,
                "speed": 0.8,
                "quality": 0.8,
                "capabilities": ["text", "conversation", "code", "reasoning"]
            }
        }
        
        # 路由历史
        self.route_history = []
    
    def analyze_task(self, task):
        """分析任务特征"""
        # 任务类型识别
        task_type = self._identify_task_type(task)
        
        # 复杂度评估
        complexity = self._evaluate_complexity(task)
        
        # 上下文需求分析
        context_needs = self._analyze_context_needs(task)
        
        # 精度要求分析
        quality_requirement = self._analyze_quality_requirement(task)
        
        return {
            "task_type": task_type,
            "complexity": complexity,
            "context_needs": context_needs,
            "quality_requirement": quality_requirement
        }
    
    def select_model(self, task_analysis):
        """选择最优模型"""
        # 筛选符合要求的模型
        suitable_models = self._filter_models(task_analysis)
        
        if not suitable_models:
            return "No suitable model found"
        
        # 计算每个模型的得分
        model_scores = {}
        for model_name, model_info in suitable_models.items():
            score = self._calculate_model_score(model_name, model_info, task_analysis)
            model_scores[model_name] = score
        
        # 选择得分最高的模型
        best_model = max(model_scores, key=model_scores.get)
        
        # 记录路由决策
        self._record_route(best_model, task_analysis, model_scores)
        
        return best_model
    
    def _identify_task_type(self, task):
        """识别任务类型"""
        task_lower = task.lower()
        
        if any(keyword in task_lower for keyword in ["code", "program", "script", "function"]):
            return "code"
        elif any(keyword in task_lower for keyword in ["translate", "translation"]):
            return "translation"
        elif any(keyword in task_lower for keyword in ["analyze", "analysis", "data"]):
            return "analysis"
        elif any(keyword in task_lower for keyword in ["write", "create", "generate"]):
            return "generation"
        elif any(keyword in task_lower for keyword in ["answer", "question", "help"]):
            return "conversation"
        else:
            return "general"
    
    def _evaluate_complexity(self, task):
        """评估任务复杂度"""
        # 基于文本长度
        length_score = min(len(task) / 1000, 1.0)
        
        # 基于内容复杂度
        complex_keywords = ["complex", "difficult", "challenging", "advanced", "expert"]
        complexity_score = sum(1 for keyword in complex_keywords if keyword in task.lower()) / len(complex_keywords)
        
        # 综合评分
        return (length_score + complexity_score) / 2
    
    def _analyze_context_needs(self, task):
        """分析上下文需求"""
        # 基于文本长度
        context_needs = len(task) / 1000
        
        # 检查是否需要长上下文
        long_context_keywords = ["long", "detailed", "comprehensive", "entire"]
        if any(keyword in task.lower() for keyword in long_context_keywords):
            context_needs = max(context_needs, 0.8)
        
        return min(context_needs, 1.0)
    
    def _analyze_quality_requirement(self, task):
        """分析精度要求"""
        # 基于关键词
        high_quality_keywords = ["accurate", "precise", "exact", "correct", "professional"]
        quality_score = sum(1 for keyword in high_quality_keywords if keyword in task.lower()) / len(high_quality_keywords)
        
        # 基于任务类型
        task_type = self._identify_task_type(task)
        type_quality_map = {
            "code": 0.9,
            "translation": 0.8,
            "analysis": 0.85,
            "generation": 0.7,
            "conversation": 0.6,
            "general": 0.5
        }
        
        return max(quality_score, type_quality_map.get(task_type, 0.5))
    
    def _filter_models(self, task_analysis):
        """筛选符合要求的模型"""
        suitable_models = {}
        
        for model_name, model_info in self.models.items():
            # 检查上下文长度
            if task_analysis["context_needs"] > 0.5 and model_info["max_context"] < 16384:
                continue
            
            # 检查能力
            task_type = task_analysis["task_type"]
            if task_type == "code" and "code" not in model_info["capabilities"]:
                continue
            elif task_type == "translation" and "text" not in model_info["capabilities"]:
                continue
            elif task_type == "analysis" and "reasoning" not in model_info.get("capabilities", []):
                continue
            
            # 检查质量要求
            if task_analysis["quality_requirement"] > 0.8 and model_info["quality"] < 0.8:
                continue
            
            suitable_models[model_name] = model_info
        
        return suitable_models
    
    def _calculate_model_score(self, model_name, model_info, task_analysis):
        """计算模型得分"""
        # 成本因素（反向，成本越低得分越高）
        cost_score = 1.0 / (model_info["cost_per_1k_tokens"] * 1000)  # 归一化
        
        # 速度因素
        speed_score = model_info["speed"]
        
        # 质量因素
        quality_score = model_info["quality"]
        
        # 上下文因素
        context_score = 1.0 if model_info["max_context"] >= 16384 else 0.5
        
        # 任务匹配度
        task_match_score = self._calculate_task_match(model_name, model_info, task_analysis)
        
        # 权重
        weights = {
            "cost": 0.3,
            "speed": 0.2,
            "quality": 0.3,
            "context": 0.1,
            "task_match": 0.1
        }
        
        # 计算总分
        score = (
            cost_score * weights["cost"] +
            speed_score * weights["speed"] +
            quality_score * weights["quality"] +
            context_score * weights["context"] +
            task_match_score * weights["task_match"]
        )
        
        return score
    
    def _calculate_task_match(self, model_name, model_info, task_analysis):
        """计算任务匹配度"""
        task_type = task_analysis["task_type"]
        capabilities = model_info["capabilities"]
        
        match_map = {
            "code": ["code", "reasoning"],
            "translation": ["text"],
            "analysis": ["reasoning", "text"],
            "generation": ["text"],
            "conversation": ["conversation"],
            "general": ["text"]
        }
        
        required_capabilities = match_map.get(task_type, ["text"])
        match_count = sum(1 for cap in required_capabilities if cap in capabilities)
        
        return match_count / len(required_capabilities)
    
    def _record_route(self, model_name, task_analysis, model_scores):
        """记录路由决策"""
        route_record = {
            "timestamp": datetime.now().isoformat(),
            "model": model_name,
            "task_analysis": task_analysis,
            "model_scores": model_scores,
            "decision_time": time.time()
        }
        
        self.route_history.append(route_record)
        
        # 保持历史记录不超过1000条
        if len(self.route_history) > 1000:
            self.route_history = self.route_history[-1000:]
    
    def get_route_history(self):
        """获取路由历史"""
        return self.route_history
    
    def optimize_routing(self):
        """优化路由策略"""
        # 基于历史数据优化权重
        if len(self.route_history) < 10:
            return "Not enough data for optimization"
        
        # 这里可以实现更复杂的优化逻辑
        # 例如：基于历史性能调整模型评分权重
        
        return "Routing optimized based on historical data"

# 使用示例
if __name__ == "__main__":
    router = ModelRouter()
    
    # 测试不同任务
    tasks = [
        "Write a simple Python function to calculate Fibonacci numbers",
        "Translate this document from English to French: 'Hello, how are you today?'",
        "Analyze this complex financial data and provide insights",
        "Write a short story about a robot learning to paint",
        "Answer this question: What is the capital of France?"
    ]
    
    for task in tasks:
        print(f"\nTask: {task}")
        analysis = router.analyze_task(task)
        print(f"Task analysis: {analysis}")
        model = router.select_model(analysis)
        print(f"Selected model: {model}")
    
    # 优化路由
    optimization_result = router.optimize_routing()
    print(f"\nOptimization result: {optimization_result}")

总成本 = 总Token数 × 模型单价

总成本 = Σ(任务i的Token数 × 模型i单价)

成本节省 = 单一模型成本 - 智能路由成本
成本节省率 = (成本节省 / 单一模型成本) × 100%

pip install numpy scikit-learn flask