适用场景与目标

当你的 AI 应用开始用多个模型时，一定会遇到这个问题：所有请求都打到一个最强最贵的模型上，导致账单爆表但输出质量没有相应提升。

智能模型路由（Model Routing）就是解决这个问题的：让对的请求去对的模型。

典型场景：

• 客服机器人：简单问答 → 小模型；复杂分析 → 大模型
• 代码助手：代码补全 → 小模型；代码审查 → 大模型
• 文档处理：摘要 → 小模型；全文翻译 → 大模型

目标：在不降低用户体验的前提下，将 API 成本降低 40-85%。

---

最小可行方案（MVP）

架构概览

请求 → 路由层（Router） → 任务分类 → 模型匹配 → 执行 → 返回

工具选型

工具	适用场景	推荐度
LiteLLM	生产级路由，支持 20+ 模型	⭐⭐⭐⭐⭐
Prompt Routing（自定义）	轻量原型，快速验证	⭐⭐⭐⭐
Portkey / Helicone	带观测的路由平台	⭐⭐⭐⭐
自研路由层	深度定制需求	⭐⭐⭐

MVP 步骤

Step 1：量化任务复杂度

先收集上线后真实请求的 token 分布和质量反馈：

# 埋点记录：记录每次请求的任务类型、token 数、模型、成本
import time
from dataclasses import dataclass

@dataclass
class RequestLog:
    task_type: str          # "classify" | "summarize" | "reason" | "generate"
    input_tokens: int
    output_tokens: int
    model: str
    latency_ms: int
    quality_score: float   # 1-5，可通过人工抽检或自动化评估获取

Step 2：建立路由规则表

基于实际数据，制定手工规则路由表——这是最稳的方式：

ROUTING_RULES = {
    "classify": {
        "model": "gpt-4o-mini",
        "max_tokens": 32000,
        "complexity_threshold": 0.3,
    },
    "summarize": {
        "model": "gpt-4o-mini",
        "max_tokens": 64000,
        "complexity_threshold": 0.5,
    },
    "reason": {
        "model": "claude-sonnet-4",
        "max_tokens": 128000,
        "complexity_threshold": 0.7,
    },
    "generate": {
        "model": "gpt-4o",
        "max_tokens": 128000,
        "complexity_threshold": 0.6,
    },
}

Step 3：用 LiteLLM 实现路由（生产级推荐）

pip install litellm

# litellm_router.py
from litellm import Router
from litellm.router_config import RouterConfig

# 定义模型组和路由策略
model_list = [
    {
        "model_name": "fast-model",      # 别名
        "litellm_params": {
            "model": "gpt-4o-mini",
            "api_key": os.getenv("OPENAI_API_KEY"),
        },
        "rpm_limit": 1000,               # 每分钟请求数限制
    },
    {
        "model_name": "smart-model",     # 别名
        "litellm_params": {
            "model": "claude-sonnet-4",
            "api_key": os.getenv("ANTHROPIC_API_KEY"),
        },
        "rpm_limit": 500,
    },
]

router = Router(
    model_list=model_list,
    routing_strategy="latency-defined",  # 优先选最低延迟
    # routing_strategy="cost-defined",  # 或：优先选最低成本
    allowed_fails=3,                    # 连续失败 3 次后切换模型
    timeout=60,
)

# 简单任务 → 自动路由到 fast-model
response = router.acompletion(
    model="fast-model",
    messages=[{"role": "user", "content": "请分类这条用户反馈：产品很好用"}],
)

# 复杂任务 → 路由到 smart-model
response = router.acompletion(
    model="smart-model",
    messages=[{"role": "user", "content": "分析以下代码的技术债并给出重构方案"}],
)

Step 4：实现基于任务复杂度的动态路由

# dynamic_router.py
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def classify_task_complexity(prompt: str, history: list[str] = None) -> float:
    """
    简单规则评估任务复杂度（0.0 ~ 1.0）
    实际生产中可用专门的评估模型或启发式规则
    """
    complexity = 0.0

    # 信号1：关键词检测
    complex_keywords = ["分析", "比较", "推理", "设计", "优化", "架构", "重构"]
    simple_keywords = ["分类", "总结", "翻译", "补全", "查找"]

    prompt_lower = prompt.lower()
    if any(kw in prompt_lower for kw in complex_keywords):
        complexity += 0.4
    if any(kw in prompt_lower for kw in simple_keywords):
        complexity -= 0.2

    # 信号2：上下文长度
    token_count = len(prompt) // 4  # 粗估
    if token_count > 32000:
        complexity += 0.3
    elif token_count > 8000:
        complexity += 0.1

    # 信号3：多轮历史
    if history and len(history) > 3:
        complexity += 0.2

    return min(1.0, max(0.0, complexity))


def route_request(prompt: str, history: list[str] = None) -> str:
    score = classify_task_complexity(prompt, history)

    if score < 0.35:
        return "gpt-4o-mini"
    elif score < 0.65:
        return "gpt-4o"
    else:
        return "claude-opus-4"


# 使用示例
model = route_request("请将这段中文翻译成英文", [])
# → "gpt-4o-mini" （低成本，简单翻译任务）

model = route_request("分析这段微服务代码的架构问题并提供优化建议", [])
# → "claude-opus-4" （高复杂度，需要深度推理）

---

关键实现细节

1. Prompt Caching 配合路由，效果叠加

2026 年主流 API 均已支持 Prompt Caching，路由 + 缓存叠加可再省 45-80%：

# 以 Anthropic 为例，使用缓存减少重复上下文开销
messages = [
    {
        "role": "user",
        "content": {
            "type": "text",
            "text": "你是一个代码审查助手..."
        }
    },
    {
        "role": "user",
        "content": {
            "type": "text",
            "text": prompt,
            "cache_control": {"type": "ephemeral"}  # 缓存这段 system prompt
        }
    }
]

response = client.messages.create(
    model="claude-sonnet-4",
    messages=messages,
    max_tokens=1024,
)

2. 基于一致性的路由（共识机制）

对于高风险任务，可将同一请求发给两个模型，比较输出取优：

async def consensus_route(prompt: str, threshold: float = 0.7) -> str:
    """共识路由：两个模型结果一致则返回，不同则上放大模型"""
    fast_response = await router.acompletion(
        model="fast-model",
        messages=[{"role": "user", "content": prompt}]
    )
    smart_response = await router.acompletion(
        model="smart-model",
        messages=[{"role": "user", "content": prompt}]
    )

    # 简单一致性检测（生产中可用嵌入相似度）
    if fast_response.choices[0].message.content == smart_response.choices[0].message.content:
        return fast_response

    # 不一致，上升至更大模型重新评估
    final_response = await router.acompletion(
        model="claude-opus-4",
        messages=[{"role": "user", "content": f"对比以下两个答案，给出最优解：\nA: {fast_response}\nB: {smart_response}"}]
    )
    return final_response

3. 路由观测：必须埋点

# 每一次路由决策都要记录，用于后续优化
def log_routing_decision(
    request_id: str,
    task_type: str,
    input_tokens: int,
    routed_model: str,
    final_model: str,      # 可能被 fallback 修改
    latency_ms: int,
    cost_usd: float,
    success: bool,
):
    # 发送到你的监控服务（DataDog / Prometheus / 自建）
    metric = {
        "request_id": request_id,
        "task_type": task_type,
        "input_tokens": input_tokens,
        "routed_model": routed_model,
        "final_model": final_model,
        "latency_ms": latency_ms,
        "cost_usd": cost_usd,
        "success": success,
        "timestamp": time.time(),
    }
    # push to metrics pipeline

---

常见坑与规避清单

坑	描述	规避方案
路由到能力不足的模型	简单任务用小模型是对的，但小模型产生幻觉或错误格式导致业务受损	设置 quality gate：输出结构校验 + 人工抽检率
路由抖动	同样任务两次，路由到不同模型，结果不一致	相同 prompt 应路由到同一模型，或在路由层加入一致性 hash
Token 估算错误	用字符数估算 token 有误差，长文本路由失准	用 tiktoken 或等效库做精确 token 计数
没有 fallback	目标模型 API 限流或宕机时请求失败	路由层必须配置 fallback 链：Opus → Sonnet → Haiku
只看成本不顾质量	过度压缩成本导致用户投诉上升	设立质量指标（CSAT、任务成功率），成本节省不能以质量跌破基线为代价
冷启动无数据	上线初期没有真实的 task_type 分布，只能拍脑袋	上线前用影子模式（shadow mode）跑真实流量，收集数据后再开启路由
缓存键粒度太细	相同 system prompt 但不同任务场景被当作不同缓存	缓存键设计应包含 task_type，而不是只看 prompt 相似度

---

成本 / 性能 / 维护权衡

成本对比（以 100 万请求 / 月为例）

策略	模型组合	预估月成本	节省比例
全量 GPT-4o	单模型	~$8,000	基准
路由（GPT-4o-mini + GPT-4o + Claude）	3 模型	~$2,400	70%
路由 + Prompt Caching	3 模型 + 缓存	~$1,200	85%

数字为示例，实际按 token 单价和请求分布计算

性能影响

• 路由本身延迟：<5ms（本地规则判断）
• 模型间延迟差异：GPT-4o-mini 的 TTFT（Time To First Token）通常比 Opus 快 2-3x
• 一致性路由延迟：×2（需发两个模型），适合离线批处理

维护成本

• 路由规则需要随着模型能力迭代更新（每年模型能力变化显著）
• 建议每季度 review 一次路由规则，基于真实数据调整 complexity_threshold

---

一周内可执行行动清单

• Day 1：审计现有 AI 请求，统计 task_type 分布和 token 消耗（用日志或 LangSmith / Helicone）
• Day 2：实现基础路由规则（基于关键词 + token 数的手工规则）
• Day 3：接入 LiteLLM，配置 2-3 个模型组成模型组，开启影子模式验证
• Day 4：配置 fallback 链 + 路由日志埋点
• Day 5：用历史请求数据回放，对比路由前后成本和质量指标
• Day 6：灰度开启 10% 流量，观察 48 小时
• Day 7：扩量至 100%，设置质量告警（CSAT 下降 >5% 触发人工 review）

---

核心原则记住：路由的目标不是最便宜，而是最合适。把对的模型用在对的任务上，成本自然降，质量不掉线。