💡 LLM 成本优化 / 模型路由

🔀 模型路由

80% 的请求用小模型就够了——简单问题 GPT-4o-mini/DeepSeek，复杂推理才上 o3/Opus。路由分类器帮你自动分配，60-80% 成本节省。

核心思想

用户请求 → 路由分类器 → 判断复杂度 → 选择模型

┌─────────────┐     ┌──────────┐     ┌───────────────┐
│ 简单问题     │ ──→ │ 小模型    │ ──→ │ GPT-4o-mini   │  $0.75/$4.50 per 1M
│ (FAQ/闲聊)   │     │ (80%请求) │     │ DeepSeek V3   │  ¥0.50/¥2.00 per 1M
└─────────────┘     └──────────┘     └───────────────┘

┌─────────────┐     ┌──────────┐     ┌───────────────┐
│ 中等问题     │ ──→ │ 中模型    │ ──→ │ Claude Sonnet │  $3/$15 per 1M
│ (写作/分析)  │     │ (15%请求) │     │ GPT-5.4       │  $2.50/$15 per 1M
└─────────────┘     └──────────┘     └───────────────┘

┌─────────────┐     ┌──────────┐     ┌───────────────┐
│ 复杂推理     │ ──→ │ 大模型    │ ──→ │ Claude Opus   │  $15/$75 per 1M
│ (代码/数学)  │     │ (5%请求)  │     │ o3            │  高定价
└─────────────┘     └──────────┘     └───────────────┘

成本计算（假设1000次请求）：
- 全用大模型:  1000 × $0.15 = $150
- 路由后:       800 × $0.005 + 150 × $0.03 + 50 × $0.15 = $14.5
- 节省: 90%!

路由分类器实现

方案1: 基于规则的分类器

import re
from dataclasses import dataclass
from enum import Enum

class Complexity(Enum):
    SIMPLE = "simple"       # FAQ、闲聊、简单翻译
    MEDIUM = "medium"       # 写作、分析、总结
    COMPLEX = "complex"     # 代码、数学、多步推理
    CRITICAL = "critical"   # 安全相关、高精度需求

@dataclass
class RoutingDecision:
    complexity: Complexity
    model: str
    reason: str
    confidence: float

class RuleBasedRouter:
    """基于规则的路由分类器"""
    
    MODEL_MAP = {
        Complexity.SIMPLE: "deepseek-chat",        # 最便宜
        Complexity.MEDIUM: "gpt-4o-mini",          # 便宜+够用
        Complexity.COMPLEX: "claude-sonnet-4-20250514",  # 强力
        Complexity.CRITICAL: "claude-opus-4-20250514",   # 最强
    }
    
    # 简单问题的特征
    SIMPLE_PATTERNS = [
        r'^(hi|hello|hey|你好|嗨)[\s!！。]*$',
        r'(what is|define|什么是|解释一下)\s+\w+\s*[?？]',
        r'(translate|翻译)\s+',
        r'(天气|weather|时间|time)',
        r'^(yes|no|是|否|好的|ok)[\s.。]*$',
    ]
    
    # 复杂问题的特征
    COMPLEX_PATTERNS = [
        r'(debug|fix|refactor|重构|调试)',
        r'(prove|证明|derive|推导)',
        r'(implement|实现|编写代码|write code)',
        r'(analyze|分析|compare|比较)\s+.{20,}',
        r'(step.by.step|分步|逐步)',
    ]
    
    def route(self, prompt: str, context: dict = None) -> RoutingDecision:
        """路由决策"""
        # 1. 检查简单模式
        for pattern in self.SIMPLE_PATTERNS:
            if re.search(pattern, prompt, re.IGNORECASE):
                return RoutingDecision(
                    complexity=Complexity.SIMPLE,
                    model=self.MODEL_MAP[Complexity.SIMPLE],
                    reason="匹配简单问题模式",
                    confidence=0.8
                )
        
        # 2. 检查复杂模式
        for pattern in self.COMPLEX_PATTERNS:
            if re.search(pattern, prompt, re.IGNORECASE):
                return RoutingDecision(
                    complexity=Complexity.COMPLEX,
                    model=self.MODEL_MAP[Complexity.COMPLEX],
                    reason="匹配复杂问题模式",
                    confidence=0.7
                )
        
        # 3. 基于长度和结构判断
        word_count = len(prompt.split())
        has_code = '```' in prompt or 'def ' in prompt or 'function ' in prompt
        has_math = any(c in prompt for c in ['∫', '∑', '∂', '√', '方程', 'equation'])
        
        if has_code or has_math:
            return RoutingDecision(
                complexity=Complexity.COMPLEX,
                model=self.MODEL_MAP[Complexity.COMPLEX],
                reason="包含代码或数学内容",
                confidence=0.85
            )
        
        if word_count > 100:
            return RoutingDecision(
                complexity=Complexity.MEDIUM,
                model=self.MODEL_MAP[Complexity.MEDIUM],
                reason="长提示词，需要中等处理能力",
                confidence=0.5
            )
        
        # 默认：中等
        return RoutingDecision(
            complexity=Complexity.MEDIUM,
            model=self.MODEL_MAP[Complexity.MEDIUM],
            reason="默认路由",
            confidence=0.3
        )

方案2: LLM 路由分类器

class LLMRouter:
    """用小模型做路由决策——更准确但有一点额外成本"""
    
    ROUTE_PROMPT = """Classify this user query by complexity. Respond with ONLY one word.

SIMPLE: Greetings, FAQs, simple lookups, translations, short answers
MEDIUM: Writing, analysis, summaries, explanations, comparisons  
COMPLEX: Coding, math proofs, multi-step reasoning, architecture design
CRITICAL: Security-sensitive, medical, legal, high-precision needs

User query: {query}

Complexity:"""
    
    def __init__(self, router_model: str = "deepseek-chat"):
        self.router_model = router_model
        self.model_map = {
            'SIMPLE': 'deepseek-chat',
            'MEDIUM': 'gpt-4o-mini',
            'COMPLEX': 'claude-sonnet-4-20250514',
            'CRITICAL': 'claude-opus-4-20250514',
        }
    
    async def route(self, prompt: str) -> RoutingDecision:
        """用 LLM 判断复杂度"""
        # 用最便宜的模型做路由
        route_response = await self._call_router(prompt)
        complexity = route_response.strip().upper()
        
        if complexity not in self.model_map:
            complexity = 'MEDIUM'  # 安全回退
        
        return RoutingDecision(
            complexity=Complexity(complexity.lower()),
            model=self.model_map[complexity],
            reason=f"LLM路由判断: {complexity}",
            confidence=0.8
        )
    
    async def _call_router(self, prompt: str) -> str:
        """调用路由模型"""
        # 使用最便宜的模型 + 最少的 token
        response = await call_llm(
            model=self.router_model,
            prompt=self.ROUTE_PROMPT.format(query=prompt[:200]),
            max_tokens=5,
            temperature=0.0
        )
        return response

# 成本分析：路由本身的开销
# DeepSeek 路由调用: ~50 tokens × ¥0.50/1M = ¥0.000025 (~$0.0000035)
# 如果路由决策能省 $0.05/请求（用小模型替代大模型），ROI = 14000x

方案3: 嵌入相似度路由

class EmbeddingRouter:
    """基于嵌入相似度的路由——适合 FAQ 场景"""
    
    def __init__(self):
        # 已知简单问题的嵌入库
        self.simple_embeddings = []  # FAQ 问题的嵌入
        self.complex_embeddings = [] # 复杂问题的嵌入
    
    async def route(self, prompt: str) -> RoutingDecision:
        """通过嵌入相似度判断问题类型"""
        prompt_embedding = await get_embedding(prompt)
        
        # 检查与简单问题的相似度
        max_simple_sim = max(
            cosine_similarity(prompt_embedding, e) 
            for e in self.simple_embeddings
        ) if self.simple_embeddings else 0
        
        # 检查与复杂问题的相似度
        max_complex_sim = max(
            cosine_similarity(prompt_embedding, e) 
            for e in self.complex_embeddings
        ) if self.complex_embeddings else 0
        
        if max_simple_sim > 0.85:
            return RoutingDecision(
                complexity=Complexity.SIMPLE,
                model="deepseek-chat",
                reason=f"与简单问题高度相似 ({max_simple_sim:.2f})",
                confidence=max_simple_sim
            )
        elif max_complex_sim > 0.80:
            return RoutingDecision(
                complexity=Complexity.COMPLEX,
                model="claude-sonnet-4-20250514",
                reason=f"与复杂问题相似 ({max_complex_sim:.2f})",
                confidence=max_complex_sim
            )
        else:
            return RoutingDecision(
                complexity=Complexity.MEDIUM,
                model="gpt-4o-mini",
                reason="未匹配已知模式",
                confidence=0.5
            )

级联回退策略

class CascadingRouter:
    """级联回退：先用小模型尝试，不满意再升级"""
    
    def __init__(self):
        self.models = [
            ("deepseek-chat", Complexity.SIMPLE),
            ("gpt-4o-mini", Complexity.MEDIUM),
            ("claude-sonnet-4-20250514", Complexity.COMPLEX),
        ]
    
    async def route_with_fallback(self, prompt: str, 
                                   quality_threshold: float = 0.8) -> dict:
        """级联回退路由"""
        for model, complexity in self.models:
            response = await call_llm(model=model, prompt=prompt)
            
            # 评估响应质量
            quality = await self._evaluate_quality(prompt, response)
            
            if quality >= quality_threshold:
                return {
                    'model': model,
                    'response': response,
                    'quality': quality,
                    'attempts': self.models.index((model, complexity)) + 1
                }
        
        # 所有模型都不满意，用最强模型
        final_response = await call_llm(
            model="claude-opus-4-20250514", prompt=prompt
        )
        return {
            'model': 'claude-opus-4-20250514',
            'response': final_response,
            'quality': 1.0,
            'attempts': len(self.models) + 1
        }
    
    async def _evaluate_quality(self, prompt: str, response: str) -> float:
        """评估响应质量（简化版）"""
        # 方法1: 基于启发式规则
        if len(response) < 10:
            return 0.2  # 太短
        if "I don't" in response or "我不" in response:
            return 0.4  # 拒绝回答
        if "```" in response and "error" in response.lower():
            return 0.3  # 代码有错误
        
        # 方法2: 用小模型快速评估
        # eval_response = await call_llm(
        #     model="deepseek-chat",
        #     prompt=f"Rate this response quality 0-1: Q:{prompt[:50]} A:{response[:100]}"
        # )
        
        return 0.7  # 默认中等质量

路由准确率 vs 成本权衡

← 上一篇：Prompt 压缩 | 下一篇：Batch API → | 返回首页