API 成本降低 73%！多模型智能路由，简单问题走小模型、复杂问题走大模型

#!/usr/bin/env python3
"""
多模型智能路由引擎
根据任务复杂度自动选择合适的模型
"""
import re
from dataclasses import dataclass
from enum import Enum

class Complexity(Enum):
    """任务复杂度"""
    SIMPLE = "simple"       # 简单：摘要、格式化、分类
    MODERATE = "moderate"   # 中等：配置审查、告警聚类
    COMPLEX = "complex"     # 复杂：根因分析、架构设计

class Model(Enum):
    """可用模型"""
    QWEN_7B_LOCAL = "qwen2.5:7b"          # 本地 Ollama
    QWEN_32B_CLOUD = "qwen2.5:32b"        # 云端
    QWEN_72B_CLOUD = "qwen2.5:72b"        # 云端

@dataclass
class RoutingResult:
    """路由结果"""
    complexity: Complexity
    model: Model
    reason: str
    estimated_tokens: int
    estimated_cost: float

# 关键词 -> 复杂度映射规则
COMPLEXITY_RULES = {
    Complexity.SIMPLE: [
        r"摘要|总结|概括",
        r"格式化|整理|清洗",
        r"分类|归类|打标签",
        r"翻译|转换",
        r"提取.*?关键字",
    ],
    Complexity.MODERATE: [
        r"审查|检查|校验",
        r"对比|比较|差异",
        r"聚类|聚合|关联",
        r"优化建议|改进",
    ],
    Complexity.COMPLEX: [
        r"根因|根本原因|为什么",
        r"架构|设计|方案",
        r"排查|诊断|定位",
        r"生成.*?策略|制定.*?计划",
        r"分析.*?链路|完整.*?流程",
    ],
}

# 复杂度 -> 模型映射
COMPLEXITY_MODEL_MAP = {
    Complexity.SIMPLE: Model.QWEN_7B_LOCAL,
    Complexity.MODERATE: Model.QWEN_32B_CLOUD,
    Complexity.COMPLEX: Model.QWEN_72B_CLOUD,
}

# 模型单 token 成本（元/千 token）
MODEL_COST = {
    Model.QWEN_7B_LOCAL: 0.0,
    Model.QWEN_32B_CLOUD: 0.002,
    Model.QWEN_72B_CLOUD: 0.01,
}

def classify_complexity(prompt: str) -> Complexity:
    """根据 Prompt 内容判断任务复杂度"""
    for complexity in [Complexity.COMPLEX,
                       Complexity.MODERATE,
                       Complexity.SIMPLE]:
        patterns = COMPLEXITY_RULES[complexity]
        for pattern in patterns:
            if re.search(pattern, prompt):
                return complexity
    return Complexity.MODERATE

def route(prompt: str, input_length: int = 0) -> RoutingResult:
    """智能路由：选择合适的模型"""
    complexity = classify_complexity(prompt)

    # 长上下文自动升级模型
    if input_length > 30000 and complexity != Complexity.COMPLEX:
        complexity = Complexity.COMPLEX
        reason = "输入超 30K token，升级至大模型"
    else:
        reason = f"任务匹配 {complexity.value} 级别"

    model = COMPLEXITY_MODEL_MAP[complexity]
    estimated_tokens = max(input_length, 500) + 300
    estimated_cost = (estimated_tokens / 1000) * MODEL_COST[model]

    return RoutingResult(
        complexity=complexity,
        model=model,
        reason=reason,
        estimated_tokens=estimated_tokens,
        estimated_cost=estimated_cost
    )

# 使用示例
test_prompts = [
    "帮我总结这条告警的核心信息",
    "审查这个 Nginx 配置有没有安全隐患",
    "分析这次故障的根因，给出完整的推理链路",
]

for p in test_prompts:
    result = route(p)
    print(f"Prompt: {p}")
    print(f"  → 复杂度: {result.complexity.value}")
    print(f"  → 模型: {result.model.value}")
    print(f"  → 预估成本: {result.estimated_cost:.4f} 元\n")

Prompt: 帮我总结这条告警的核心信息
  → 复杂度: simple
  → 模型: qwen2.5:7b
  → 预估成本: 0.0000 元

Prompt: 审查这个 Nginx 配置有没有安全隐患
  → 复杂度: moderate
  → 模型: qwen2.5:32b
  → 预估成本: 0.0016 元

Prompt: 分析这次故障的根因，给出完整的推理链路
  → 复杂度: complex
  → 模型: qwen2.5:72b
  → 预估成本: 0.0080 元

def refine_routing(prompt: str, context: dict) -> RoutingResult:
    """精细化路由：结合输入特征调整"""
    base_result = route(prompt, context.get("input_length", 0))

    # 输入过长，升级模型
    if context.get("input_length", 0) > 2000:
        if base_result.complexity == Complexity.SIMPLE:
            base_result.complexity = Complexity.MODERATE
            base_result.model = Model.QWEN_32B_CLOUD
            base_result.reason += " + 输入较长自动升级"

    # 多数据源关联，升级模型
    if context.get("multi_source", False):
        base_result.complexity = Complexity.MODERATE
        base_result.model = Model.QWEN_32B_CLOUD
        base_result.reason += " + 多源数据自动升级"

    return base_result

import time

class ModelHealthChecker:
    """模型健康检查器"""

    def __init__(self):
        self.health_status = {}

    def check(self, model: Model) -> bool:
        """检查模型是否可用"""
        status = self.health_status.get(model.value)
        if status and time.time() - status["last_check"] < 30:
            return status["healthy"]

        try:
            if model == Model.QWEN_7B_LOCAL:
                resp = requests.post(
                    "http://localhost:11434/api/generate",
                    json={"model": "qwen2.5:7b",
                          "prompt": "ping", "stream": False},
                    timeout=5
                )
                healthy = resp.status_code == 200
            else:
                healthy = True
        except requests.RequestException:
            healthy = False

        self.health_status[model.value] = {
            "healthy": healthy,
            "last_check": time.time(),
            "fail_count": 0 if healthy
                else self.health_status.get(model.value, {}).get("fail_count", 0) + 1
        }
        return healthy

    def get_fallback(self, model: Model) -> Model:
        """获取降级模型"""
        fallback_map = {
            Model.QWEN_7B_LOCAL: Model.QWEN_32B_CLOUD,
            Model.QWEN_32B_CLOUD: Model.QWEN_72B_CLOUD,
            Model.QWEN_72B_CLOUD: None,
        }
        return fallback_map.get(model)

from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_classify(prompt_prefix: str) -> Complexity:
    """缓存分类结果，相同前缀不重复计算"""
    return classify_complexity(prompt_prefix)

def fast_route(prompt: str) -> RoutingResult:
    """快速路由：用前 50 字符做缓存 key"""
    prefix = prompt[:50].strip()
    complexity = cached_classify(prefix)
    model = COMPLEXITY_MODEL_MAP[complexity]
    return RoutingResult(
        complexity=complexity,
        model=model,
        reason="cached routing",
        estimated_tokens=500,
        estimated_cost=0
    )

# Crontab 配置：路由优化定时任务
0 4 * * * /opt/scripts/routing_optimizer.sh       # 每天凌晨优化路由策略
0 */6 * * * /opt/scripts/model_cost_review.sh      # 每6小时成本回顾
*/5 * * * * /opt/scripts/routing_health_check.sh   # 每5分钟路由健康检查
0 8 * * 1 /opt/scripts/classifier_retrain.sh       # 每周一重训练分类器