模型路由策略：智能调度降低LLM成本的核心技术

在 AI 应用的生产部署中，一个核心难题是：如何在保证输出质量的同时最小化成本？如果所有请求都使用 GPT-4o，成本高昂；如果全用 GPT-4o-mini，复杂任务效果差。模型路由（Model Routing）正是解决这一矛盾的关键技术。

一、为什么需要模型路由？

1.1 成本与质量的矛盾

策略一：全用顶级模型
┌─────────────────────────────┐
│ 所有请求 → GPT-4o           │
│                             │
│ 月成本: $50,000             │
│ 质量: ★★★★★                │
│ 简单任务: 90% 浪费           │
└─────────────────────────────┘

策略二：全用性价比模型
┌─────────────────────────────┐
│ 所有请求 → GPT-4o-mini      │
│                             │
│ 月成本: $2,500              │
│ 质量: ★★★☆☆                │
│ 复杂任务: 40% 不合格          │
└─────────────────────────────┘

策略三：智能路由（目标）
┌─────────────────────────────┐
│ 简单请求 → 便宜模型 (70%)    │
│ 复杂请求 → 旗舰模型 (30%)    │
│                             │
│ 月成本: $7,500              │
│ 质量: ★★★★★                │
└─────────────────────────────┘

1.2 收益测算

二、路由策略分类

2.1 基于规则的路由

最简单的实现方式：通过显式规则决定使用哪个模型。

class RuleBasedRouter:
    """
    基于规则的路由器
    
    规则类型：
    - 关键词匹配
    - 输入长度
    - 任务类型标记
    """
    
    RULES = [
        {
            'name': 'greeting',
            'condition': lambda msg: any(
                kw in msg.lower() 
                for kw in ['你好', 'hi', 'hello', '早上好']
            ),
            'model': 'gpt-4o-mini',
            'params': {'temperature': 0.3, 'max_tokens': 100}
        },
        {
            'name': 'code_generation',
            'condition': lambda msg: any(
                kw in msg.lower()
                for kw in ['写代码', '代码', 'function', '实现']
            ),
            'model': 'claude-3-sonnet',
            'params': {'temperature': 0.2, 'max_tokens': 2000}
        },
        {
            'name': 'financial_analysis',
            'condition': lambda msg: any(
                kw in msg.lower()
                for kw in ['财报', '股票', '投资', '分析数据']
            ),
            'model': 'gpt-4o',
            'params': {'temperature': 0.1, 'max_tokens': 1000}
        },
    ]
    
    def route(self, user_message: str) -> dict:
        for rule in self.RULES:
            if rule['condition'](user_message):
                return {
                    'model': rule['model'],
                    'params': rule['params'],
                    'reason': rule['name']
                }
        
        # 默认路由
        return {
            'model': 'gpt-4o-mini',
            'params': {'temperature': 0.7},
            'reason': 'default'
        }

2.2 基于难度的路由

通过评估问题的复杂度自动选择模型。

class DifficultyRouter:
    """
    基于任务难度的路由
    
    难度评估维度：
    1. 输入长度（长输入通常更复杂）
    2. 语言复杂度（词汇多样性、句式复杂度）
    3. 推理步骤（需要多少步推理）
    4. 领域专业性
    """
    
    def assess_difficulty(self, message: str) -> float:
        """评估请求复杂度 (0-1)"""
        scores = []
        
        # 1. 长度维度
        length = len(message)
        length_score = min(length / 2000, 1.0)  # 超过2000字算复杂
        scores.append(length_score * 0.2)
        
        # 2. 推理关键词
        reasoning_keywords = ['为什么', '如果', '比较', '分析', 
                             '推理', '推导', '假设', '因果']
        reasoning_count = sum(1 for kw in reasoning_keywords 
                            if kw in message)
        reasoning_score = min(reasoning_count / 3, 1.0)
        scores.append(reasoning_score * 0.3)
        
        # 3. 结构化要求
        structure_keywords = ['表格', 'JSON', '结构化', '格式要求',
                             'CSV', '分类', '排序']
        structure_count = sum(1 for kw in structure_keywords 
                            if kw in message)
        structure_score = min(structure_count / 2, 1.0)
        scores.append(structure_score * 0.2)
        
        # 4. 领域专业性
        domain_keywords = ['代码', '算法', '数学', '公式', '法律',
                          '金融', '医疗', '科学']
        domain_count = sum(1 for kw in domain_keywords 
                          if kw in message)
        domain_score = min(domain_count / 2, 1.0)
        scores.append(domain_score * 0.3)
        
        return sum(scores)
    
    def route(self, message: str) -> dict:
        difficulty = self.assess_difficulty(message)
        
        if difficulty < 0.2:
            model = 'gpt-4o-mini'
            params = {'temperature': 0.7, 'max_tokens': 300}
        elif difficulty < 0.5:
            model = 'claude-3-haiku'
            params = {'temperature': 0.5, 'max_tokens': 500}
        elif difficulty < 0.8:
            model = 'claude-3-sonnet'
            params = {'temperature': 0.3, 'max_tokens': 1500}
        else:
            model = 'gpt-4o'
            params = {'temperature': 0.2, 'max_tokens': 2000}
        
        return {
            'model': model,
            'params': params,
            'difficulty': difficulty
        }

2.3 ML 预测路由

通过训练分类器预测哪个模型对当前请求效果最好。

import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_extraction.text import TfidfVectorizer

class MLRouter:
    """
    基于机器学习预测的路由器
    
    训练流程：
    1. 收集历史请求数据
    2. 使用多个模型处理同一请求
    3. 评估哪个模型效果最好（质量评分）
    4. 训练分类器预测最优模型
    """
    
    def __init__(self):
        self.vectorizer = TfidfVectorizer(max_features=5000)
        self.classifier = RandomForestClassifier(
            n_estimators=100,
            max_depth=10,
            random_state=42
        )
        self.is_trained = False
        self.model_order = ['cheap', 'balanced', 'premium']
    
    def _extract_features(self, message: str) -> np.ndarray:
        """提取文本特征"""
        return self.vectorizer.transform([message])
    
    def train(self, queries, best_models):
        """
        训练路由分类器
        queries: 历史查询文本列表
        best_models: 每个查询的最佳模型标签
        """
        X = self.vectorizer.fit_transform(queries)
        self.classifier.fit(X, best_models)
        self.is_trained = True
    
    def route(self, message: str) -> str:
        """预测最佳模型"""
        if not self.is_trained:
            return 'balanced'  # 未训练时使用默认
        
        features = self._extract_features(message)
        prediction = self.classifier.predict(features)[0]
        confidence = max(self.classifier.predict_proba(features)[0])
        
        # 置信度过低时回退到高配模型
        if confidence < 0.6:
            return 'premium'
        
        return prediction

三、生产级路由架构

3.1 整体架构

┌─────────────┐    ┌──────────────┐    ┌──────────────┐
│  用户请求    │───▶│  入口网关    │───▶│  请求分析    │
└─────────────┘    └──────────────┘    └──────┬───────┘
                                              │
                                              ▼
                                     ┌────────────────┐
                                     │  路由决策引擎   │
                                     │                │
                                     │ 规则匹配？─────▶ L1 路由
                                     │ ML预测？───────▶ L2 路由
                                     │ 自适应？───────▶ L3 路由
                                     └────────┬───────┘
                                              │
                    ┌─────────────────────────┼──────────┐
                    │                         │          │
                    ▼                         ▼          ▼
              ┌──────────┐            ┌────────────┐  ┌──────────┐
              │ GPT-4o   │            │ Claude 3   │  │ 自部署    │
              │ -mini    │            │ Sonnet     │  │ Llama    │
              │ (便宜)   │            │ (均衡)     │  │ (内部)   │
              └──────────┘            └────────────┘  └──────────┘

3.2 完整实现

from datetime import datetime, timedelta
from collections import defaultdict

class ProductionRouter:
    """
    生产级模型路由器
    
    功能：
    - 多级路由决策
    - 自动降级 / 升级
    - 效果反馈收集
    - 成本配额管理
    """
    
    def __init__(self):
        self.rules_router = RuleBasedRouter()
        self.difficulty_router = DifficultyRouter()
        self.ml_router = MLRouter()
        
        # 成本配额
        self.daily_quota = {
            'gpt-4o': 1000,      # 每天最多 1000 次
            'claude-3-sonnet': 3000,
            'claude-3-haiku': 10000,
            'gpt-4o-mini': 50000,
        }
        self.usage_today = defaultdict(int)
        self.last_reset = datetime.now()
    
    def _check_quota(self, model: str) -> bool:
        """检查模型配额"""
        # 每天重置
        now = datetime.now()
        if now.date() != self.last_reset.date():
            self.usage_today.clear()
            self.last_reset = now
        
        if self.usage_today[model] >= self.daily_quota.get(model, float('inf')):
            return False
        return True
    
    def route(self, messages, user_feedback=None):
        """
        主路由方法
        
        user_feedback: 用户对上次回复的评分 (1-5)
        """
        user_message = messages[-1]['content']
        
        # Step 1: 尝试规则路由（最快）
        rule_result = self.rules_router.route(user_message)
        if self._check_quota(rule_result['model']):
            rule_result['method'] = 'rule'
            return rule_result
        
        # Step 2: 难度路由
        diff_result = self.difficulty_router.route(user_message)
        if self._check_quota(diff_result['model']):
            diff_result['method'] = 'difficulty'
            return diff_result
        
        # Step 3: ML 预测（如果已训练）
        if self.ml_router.is_trained:
            ml_pred = self.ml_router.route(user_message)
            if self._check_quota(ml_pred):
                return {
                    'model': ml_pred,
                    'method': 'ml'
                }
        
        # Step 4: 配额耗尽时的降级
        # 尝试找到还有配额的模型（从便宜到贵）
        fallback_order = ['gpt-4o-mini', 'claude-3-haiku', 
                         'claude-3-sonnet', 'gpt-4o']
        for model in fallback_order:
            if self._check_quota(model):
                return {
                    'model': model,
                    'method': 'fallback',
                    'note': f'{model} 配额已用尽，降级'
                }
        
        # 所有配额都用完了
        return {
            'model': None,
            'error': 'all_quotas_exhausted',
            'retry_after': self._next_quota_reset()
        }

四、效果评估与回传

4.1 反馈收集

class RouterFeedback:
    """路由效果反馈系统"""
    
    def __init__(self):
        self.feedback_store = []
    
    def collect_feedback(self, request_id, model_used, 
                         user_rating, success, latency):
        """收集路由决策的效果反馈"""
        self.feedback_store.append({
            'request_id': request_id,
            'model': model_used,
            'rating': user_rating,
            'success': success,
            'latency': latency,
            'timestamp': datetime.now()
        })
    
    def get_model_performance(self, days=7):
        """获取各模型的效果统计"""
        cutoff = datetime.now() - timedelta(days=days)
        recent = [f for f in self.feedback_store 
                  if f['timestamp'] > cutoff]
        
        stats = defaultdict(list)
        for fb in recent:
            stats[fb['model']].append({
                'rating': fb['rating'],
                'success': fb['success']
            })
        
        return {
            model: {
                'avg_rating': mean([s['rating'] for s in scores]),
                'success_rate': sum(s['success'] for s in scores) / len(scores),
                'sample_size': len(scores)
            }
            for model, scores in stats.items()
        }

4.2 持续优化

def optimize_rules(router, feedback, min_samples=100):
    """
    基于反馈数据自动优化路由规则
    """
    for rule in router.rules_router.RULES:
        # 计算该规则的模型选择是否最优
        rule_feedback = [
            f for f in feedback 
            if f.get('rule_matched') == rule['name']
        ]
        
        if len(rule_feedback) < min_samples:
            continue
        
        # 计算当前模型的效果
        current_ratings = [
            f['rating'] for f in rule_feedback 
            if f['model'] == rule['model']
        ]
        current_score = mean(current_ratings) if current_ratings else 0
        
        # 尝试推荐更好的模型
        model_scores = defaultdict(list)
        for fb in rule_feedback:
            model_scores[fb['model']].append(fb['rating'])
        
        best_model = max(
            model_scores, 
            key=lambda m: mean(model_scores[m])
        )
        
        if best_model != rule['model'] and current_score < 4.0:
            print(f"⚠️ 规则 '{rule['name']}' 建议从 "
                  f"{rule['model']} 改为 {best_model}")

五、高级路由策略

5.1 多模型投票

对关键请求使用多个模型，取多数意见：

def consensus_routing(messages, min_models=3):
    """
    多模型投票：多个模型独立生成，取一致结果
    用于金融、医疗等高质量要求场景
    """
    models = ['gpt-4o', 'claude-3-sonnet', 'gemini-1.5-pro']
    results = []
    
    for model in models:
        response = call_model(model, messages)
        results.append(response)
    
    # 一致性分析
    if all(r == results[0] for r in results):
        return results[0]  # 完全一致
    else:
        # 使用最昂贵的模型，或要求重新生成
        return results[0]  # 或用投票机制

5.2 级联路由（Cascade）

先用便宜模型，效果不好时升级：

def cascade_routing(messages, max_attempts=3):
    """
    级联路由：先尝试便宜模型，评分不够再升级
    """
    tiers = [
        {'model': 'gpt-4o-mini', 'cost': 0.001},
        {'model': 'claude-3-haiku', 'cost': 0.002},
        {'model': 'gpt-4o', 'cost': 0.015},
    ]
    
    for attempt, tier in enumerate(tiers[:max_attempts]):
        response = call_model(tier['model'], messages)
        
        # 自评分：评估回复质量
        quality = self_evaluate_quality(messages, response)
        
        if quality > 0.8 or attempt == max_attempts - 1:
            return response
        # 质量不够，升级到更好的模型
    
    return response  # 兜底

六、生产实践建议

6.1 实施路径

6.2 关键经验

1. 从简单开始：规则路由足以覆盖 80% 的场景
2. 成本与质量的天平：宁可偶尔浪费，不要让用户体验下降
3. 配额是最后防线：路由策略优先，配额是兜底
4. 持续收集反馈：没有反馈就没有优化方向
5. 灰度发布：新路由策略先覆盖 10% 的请求

总结

模型路由是 AI 工程化中投入产出比最高的优化手段之一。通过合理的设计，可以在不降低用户体验的前提下，将 API 成本降低 60-80%。关键在于选择与自身场景匹配的路由策略，逐步从简单规则演进到智能自适应路由。