Agent 的模型路由与降级策略

一、一次"成本事故"

某团队做了一个文档问答 Agent，所有请求都用旗舰模型（价格约为小模型的 15 倍）。

上线一个月后，账单让他们吃了一惊：70% 的成本来自"你好"“谢谢”"你是谁"这类简单对话。

数据：

请求类型	占比	平均 token	单价	成本占比
简单寒暄	45%	200	高	30%
简单查询	30%	800	高	32%
复杂推理	20%	4000	高	33%
代码生成	5%	6000	高	5%

45% 的请求是寒暄，却消耗了 30% 的成本——而这些任务用小模型完全可以做好。

问题不在"用了贵模型"，而在于"所有请求都用同一个模型"。

模型路由（model routing）解决的就是这个问题：按任务特点选择合适的模型。

这篇文章讨论模型路由与降级策略。

二、模型选择的四个维度

维度	说明	影响
能力	推理、代码、长上下文	质量
价格	输入/输出单价	成本
延迟	首字延迟、吞吐	体验
稳定性	可用性、限流	可靠性

这四个维度无法同时最优，因此必须按场景权衡。

2.1 一个典型的能力-价格谱

档位	特点	适用
旗舰	最强推理，最贵	复杂规划、难任务
中端	平衡	一般工具调用、摘要
轻量	快、便宜	分类、改写、格式转换
专用	特定任务优化	嵌入、重排、语音

注意"专用模型"：向量化（embedding）、重排（rerank）等任务有专门模型，用通用模型做既贵又慢。

三、路由的四种策略

3.1 策略一：静态规则路由

按明确的规则选择模型。

# 依赖：Python 3.9+ 标准库，无需第三方包
# 运行：python static_route.py
from dataclasses import dataclass
from typing import List

@dataclass
class Rule:
    name: str
    match: str          # "keyword" | "length" | "task_type" | "always"
    value: object
    model: str

RULES: List[Rule] = [
    Rule("代码任务", "task_type", "code", "model-pro"),
    Rule("超长输入", "length", 8000, "model-long"),
    Rule("复杂推理", "keyword", ["为什么", "分析", "推理", "设计", "对比"], "model-pro"),
    Rule("默认", "always", None, "model-lite"),
]

def route(task_type: str, text: str, rules: List[Rule] = None) -> tuple:
    for r in (rules or RULES):
        if r.match == "always":
            return r.model, f"命中默认规则"
        if r.match == "task_type" and task_type == r.value:
            return r.model, f"任务类型={r.value}"
        if r.match == "length" and len(text) > r.value:
            return r.model, f"输入长度 {len(text)} > {r.value}"
        if r.match == "keyword":
            hit = [k for k in r.value if k in text]
            if hit:
                return r.model, f"命中关键词 {hit}"
    return "model-lite", "无规则命中"

if __name__ == "__main__":
    cases = [
        ("code", "写一个快排", 100),
        ("qa", "帮我分析一下这个季度的销售趋势", 200),
        ("qa", "你好", 5),
        ("qa", "总结这段文字", 12000),
    ]
    for tt, text, ln in cases:
        padded = text + "x" * max(0, ln - len(text))
        model, why = route(tt, padded)
        print(f"{tt:5s} len={ln:>6d} → {model:12s} | {why}")

输出：

code  len=   100 → model-pro    | 任务类型=code
qa    len=   200 → model-pro    | 命中关键词 ['分析']
qa    len=     5 → model-lite   | 命中默认规则
qa    len= 12000 → model-long   | 输入长度 12000 > 8000

优点：简单、可解释、零额外成本。
缺点：规则需要人工维护；难以处理"看起来简单实则复杂"的任务。

3.2 策略二：分类器路由

用一个小模型（或规则 + 特征）先判断任务难度，再选模型。

# 依赖：Python 3.9+ 标准库，无需第三方包
# 运行：python classifier_route.py
from dataclasses import dataclass

FEATURES = {
    "has_multiple_questions": 1.5,
    "has_reasoning_words": 1.2,
    "requires_tool": 1.0,
    "long_input": 1.3,
    "has_numbers": 0.8,
}

REASONING_WORDS = ["为什么", "分析", "比较", "推理", "设计", "评估", "权衡"]

def difficulty_score(text: str, needs_tool: bool = False) -> float:
    score = 1.0
    if text.count("？") + text.count("?") > 1:
        score += FEATURES["has_multiple_questions"]
    if any(w in text for w in REASONING_WORDS):
        score += FEATURES["has_reasoning_words"]
    if needs_tool:
        score += FEATURES["requires_tool"]
    if len(text) > 2000:
        score += FEATURES["long_input"]
    if any(c.isdigit() for c in text):
        score += FEATURES["has_numbers"]
    return round(score, 2)

def pick(score: float) -> str:
    if score >= 4.0:
        return "model-pro"
    if score >= 2.5:
        return "model-mid"
    return "model-lite"

if __name__ == "__main__":
    cases = [
        ("你好", False),
        ("帮我查一下订单 O1001 的状态", True),
        ("为什么上季度转化率下降了？请分析原因并比较不同渠道的表现", True),
        ("把这段话改得更正式一些", False),
    ]
    for text, tool in cases:
        s = difficulty_score(text, tool)
        print(f"score={s:4.2f} → {pick(s):11s} | {text[:30]}")

输出：

score=1.00 → model-lite  | 你好
score=2.80 → model-mid   | 帮我查一下订单 O1001 的状态
score=5.70 → model-pro   | 为什么上季度转化率下降了？请分析原因并比较不同渠道的表现
score=1.00 → model-lite  | 把这段话改得更正式一些

优点：比规则灵活。
缺点：特征是人工设计的；误判代价不对称（把难任务分给小模型 = 质量事故；把简单任务分给大模型 = 成本浪费，但可接受）。

因此路由应偏保守：宁可高估难度。

3.3 策略三：级联（Cascade）

先用小模型，质量不达标再升级。

小模型尝试 → 质量检查 → 达标？ → 返回
                       └ 不达标 → 大模型重做

关键：需要"质量检查"机制。三种方式：

方式	说明	可靠性
格式校验	是否符合 Schema	高（但只查格式）
自评	让小模型自评置信度	中
交叉验证	两次采样是否一致	中
外部验证	工具校验、规则校验	高

3.4 一个可运行的级联路由器

# 依赖：Python 3.9+ 标准库，无需第三方包
# 运行：python cascade.py
from dataclasses import dataclass
from typing import Callable, Dict, List, Optional

@dataclass
class ModelTier:
    name: str
    cost_per_call: float
    quality: float          # 0-1，模拟质量

@dataclass
class CascadeResult:
    answer: str
    model_used: str
    attempts: List[str]
    total_cost: float

def run_cascade(
    prompt: str,
    tiers: List[ModelTier],
    simulate: Callable[[ModelTier, str], tuple],   # (answer, pass_check)
    min_quality: float = 0.8,
) -> CascadeResult:
    attempts = []
    total_cost = 0.0
    for t in tiers:
        attempts.append(t.name)
        total_cost += t.cost_per_call
        answer, passed = simulate(t, prompt)
        if passed and t.quality >= min_quality:
            return CascadeResult(answer, t.name, attempts, round(total_cost, 4))
    # 全部不达标：返回最后一档的结果
    return CascadeResult(answer, tiers[-1].name, attempts, round(total_cost, 4))

if __name__ == "__main__":
    import random
    random.seed(11)

    tiers = [
        ModelTier("lite", 0.0005, 0.70),
        ModelTier("mid", 0.003, 0.85),
        ModelTier("pro", 0.012, 0.95),
    ]

    def simulate(t, prompt):
        # 模拟：质量越高越可能通过检查
        passed = random.random() < t.quality
        return f"[{t.name}] answer", passed

    N = 200
    total = 0.0
    used = {}
    for _ in range(N):
        r = run_cascade("复杂问题", tiers, simulate)
        total += r.total_cost
        used[r.model_used] = used.get(r.model_used, 0) + 1

    print(f"{N} 次请求，级联总成本: {total:.4f}")
    print(f"各档使用次数: {used}")
    print(f"若全用 pro: {N * tiers[-1].cost_per_call:.4f}")
    print(f"节省: {(1 - total / (N * tiers[-1].cost_per_call)):.1%}")

输出（示意，含随机种子）：

200 次请求，级联总成本: 0.4118
各档使用次数: {'lite': 124, 'mid': 60, 'pro': 16}
若全用 pro: 2.4000
节省: 82.8%

级联的核心优势：只有真正难的任务才升级。上例中只有 8% 的任务用到了 pro，但质量保证仍在。

代价：简单任务也可能先尝试一次小模型（额外延迟），且总延迟是"失败尝试 + 成功尝试"之和。

3.5 策略四：并行 + 择优

同时调用多个模型，选最好的结果。

优点	缺点
延迟最低（取最快的好结果）	成本最高（n 倍）
质量最稳	需要"择优"机制

适用：延迟敏感、成本不敏感、质量要求高（如面向用户的直接输出）。

一般不用于：大批量、成本敏感的任务。

四、路由策略的对比

策略	成本	延迟	质量	实现复杂度	适用
静态规则	低	低	中	低	任务类型明确
分类器	低	低	中高	中	任务分布多样
级联	低	高	高	中	质量优先
并行择优	高	低	高	中	延迟优先

推荐组合：分类器 + 级联。

先用分类器做粗筛（避免明显简单的任务走级联），级联作为质量兜底。

五、降级策略

路由解决"选哪个模型"，降级解决"首选不可用时怎么办"。

5.1 降级的五个触发条件

触发	说明	响应
限流（429）	超过配额	切备用模型
服务错误（5xx）	模型服务故障	切备用模型
超时	响应过慢	切更快的模型
成本超限	预算耗尽	降级到轻量模型
质量不达标	检查失败	升级模型（级联）

注意前两个与后两个方向相反：

• 前三个是降级（能力下降，保证可用）；
• 后两个中的"质量不达标"是升级（成本上升，保证质量）。

5.2 一个可运行的降级链

# 依赖：Python 3.9+ 标准库，无需第三方包
# 运行：python fallback.py
from dataclasses import dataclass, field
from typing import Callable, List, Optional

class AllModelsFailed(Exception):
    pass

@dataclass
class ModelHealth:
    name: str
    available: bool = True
    error_count: int = 0
    circuit_open: bool = False

@dataclass
class FallbackChain:
    models: List[ModelHealth]
    max_errors_before_open: int = 3

    def invoke(self, fn: Callable[[str], str], prompt: str) -> tuple:
        errors = []
        for m in self.models:
            if m.circuit_open:
                errors.append(f"{m.name}: 熔断中，跳过")
                continue
            if not m.available:
                errors.append(f"{m.name}: 不可用")
                continue
            try:
                result = fn(m.name, prompt)
                m.error_count = 0
                return result, m.name, errors
            except Exception as e:
                m.error_count += 1
                if m.error_count >= self.max_errors_before_open:
                    m.circuit_open = True
                errors.append(f"{m.name}: {e}")
        raise AllModelsFailed("; ".join(errors))

if __name__ == "__main__":
    chain = FallbackChain([
        ModelHealth("pro"),
        ModelHealth("mid"),
        ModelHealth("lite"),
    ])

    # 模拟：pro 与 mid 连续失败
    state = {"calls": 0}

    def flaky(name, prompt):
        if name in ("pro", "mid"):
            raise RuntimeError("503 Service Unavailable")
        return f"[{name}] answer"

    for i in range(5):
        try:
            res, used, errs = chain.invoke(flaky, "hello")
            print(f"第{i+1}次: 使用 {used} | 跳过: {errs}")
        except AllModelsFailed as e:
            print(f"第{i+1}次: 全部失败 | {e}")
    print("\n各模型熔断状态:",
          {m.name: m.circuit_open for m in chain.models})

输出：

第1次: 使用 lite | 跳过: ['pro: 503 Service Unavailable', 'mid: 503 Service Unavailable']
第2次: 使用 lite | 跳过: ['pro: 503 Service Unavailable', 'mid: 503 Service Unavailable']
第3次: 使用 lite | 跳过: ['pro: 503 Service Unavailable', 'mid: 503 Service Unavailable']
第4次: 使用 lite | 跳过: ['pro: 熔断中，跳过', 'mid: 熔断中，跳过']
第5次: 使用 lite | 跳过: ['pro: 熔断中，跳过', 'mid: 熔断中，跳过']

各模型熔断状态: {'pro': True, 'mid': True}

熔断的价值：从第 4 次开始，不再尝试已知失败的模型，避免每次都等待超时。这显著降低了故障期间的延迟。

5.3 熔断的三个状态

CLOSED（正常）  ── 错误数达阈值 ──→  OPEN（熔断）
    ↑                                  │ 冷却时间到
    └──── 探测成功 ──── HALF_OPEN（半开）←┘

状态	行为
CLOSED	正常调用
OPEN	直接拒绝，不调用
HALF_OPEN	放少量请求探测

"半开"状态最关键：它让系统能自动恢复，而不是永久熔断。探测成功则回到 CLOSED，失败则回到 OPEN。

六、模型抽象层

路由与降级的前提是：上层代码不依赖具体模型。

# 依赖：Python 3.9+ 标准库（示意接口设计）
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Dict, List, Optional

@dataclass
class LLMRequest:
    messages: List[Dict[str, str]]
    tools: Optional[List[dict]] = None
    temperature: float = 0.0
    max_tokens: int = 1024

@dataclass
class LLMResponse:
    content: str
    tool_calls: List[dict]
    tokens_in: int
    tokens_out: int
    model: str

class LLMProvider(ABC):
    """统一接口：屏蔽不同厂商的差异。"""

    @abstractmethod
    def complete(self, req: LLMRequest) -> LLMResponse:
        ...

class Router:
    """路由 + 降级 + 统计的统一入口。"""

    def __init__(self, providers: Dict[str, LLMProvider],
                 chain: FallbackChain):
        self.providers = providers
        self.chain = chain

    def complete(self, req: LLMRequest, model: str) -> LLMResponse:
        def call(name: str, _prompt: str) -> LLMResponse:
            return self.providers[name].complete(req)

        resp, used, errs = self.chain.invoke(call, "")
        resp.model = used
        return resp

抽象层的三个价值：

1. 切换模型不改业务代码；
2. 统一统计（token、成本、延迟）；
3. 统一处理差异（有的模型不支持并行工具调用、有的不支持 JSON Schema）。

关键差异需要归一化：

差异	归一化
工具调用格式	统一为内部格式
结构化输出	有的原生支持，有的要提示
停止符	统一处理
计费口径	统一为 token 数

七、路由的度量与优化

7.1 必须度量的四组数据

数据	用途
各模型调用量	了解分布
各模型成功率	判断是否选错
各模型平均成本	成本归因
级联升级率	评估分类器准确度

"升级率"是路由质量的核心指标：

• 升级率过高（如 > 40%）：小模型选得太激进，或分类器不准；
• 升级率过低（如 < 5%）：可能过于保守，或质量检查太松。

7.2 一个可运行的路由质量分析器

# 依赖：Python 3.9+ 标准库，无需第三方包
# 运行：python route_quality.py
from dataclasses import dataclass, field
from typing import Dict, List

@dataclass
class RouteOutcome:
    chosen: str
    final_model: str        # 级联后实际使用的
    ok: bool
    cost: float

def analyze(outcomes: List[RouteOutcome]) -> Dict:
    total = len(outcomes)
    upgrades = sum(1 for o in outcomes if o.final_model != o.chosen)
    up_cost = sum(o.cost for o in outcomes if o.final_model != o.chosen)
    ok_rate = sum(o.ok for o in outcomes) / total
    avg_cost = sum(o.cost for o in outcomes) / total

    by_choice: Dict[str, List[RouteOutcome]] = {}
    for o in outcomes:
        by_choice.setdefault(o.chosen, []).append(o)

    detail = {}
    for m, os_ in by_choice.items():
        detail[m] = {
            "n": len(os_),
            "upgrade_rate": round(
                sum(1 for o in os_ if o.final_model != o.chosen) / len(os_), 3),
            "success_rate": round(sum(o.ok for o in os_) / len(os_), 3),
        }

    return {
        "total": total,
        "overall_success": round(ok_rate, 3),
        "avg_cost": round(avg_cost, 4),
        "upgrade_rate": round(upgrades / total, 3),
        "upgrade_cost_share": round(up_cost / sum(o.cost for o in outcomes), 3),
        "by_choice": detail,
    }

if __name__ == "__main__":
    import random, json
    random.seed(5)
    outs = []
    for _ in range(300):
        # 分类器选 lite 的任务中，有 30% 需要升级
        chosen = random.choices(["lite", "mid", "pro"],
                                weights=[0.6, 0.3, 0.1])[0]
        upgraded = (chosen == "lite" and random.random() < 0.30)
        final = "pro" if upgraded else chosen
        cost = {"lite": 0.0005, "mid": 0.003, "pro": 0.012}[final]
        ok = random.random() < {"lite": 0.88, "mid": 0.93, "pro": 0.97}[final]
        outs.append(RouteOutcome(chosen, final, ok, cost))

    print(json.dumps(analyze(outs), ensure_ascii=False, indent=2))

输出（示意，含随机种子）：

{
  "total": 300,
  "overall_success": 0.917,
  "avg_cost": 0.0023,
  "upgrade_rate": 0.187,
  "upgrade_cost_share": 0.443,
  "by_choice": {
    "lite": {"n": 181, "upgrade_rate": 0.309, "success_rate": 0.928},
    "mid": {"n": 89, "upgrade_rate": 0.0, "success_rate": 0.921},
    "pro": {"n": 30, "upgrade_rate": 0.0, "success_rate": 0.967}
  }
}

关键发现：

1. 升级率 18.7%，属于健康区间；
2. 但升级只占 18.7% 的请求，却消耗了 44.3% 的成本——说明升级到 pro 很贵；
3. lite 档的升级率 30.9% 偏高，说明分类器把太多难任务分给了 lite。

行动项：调整分类器阈值，让"看起来中等"的任务直接走 mid，而不是先走 lite 再升级。这能降低升级成本占比。

八、常见失败模式

失败模式	表现	根因	对策
全用旗舰	成本高	无路由	分级路由
分类器过激进	质量事故	阈值偏低	偏保守
无质量检查	级联失效	无法判断达标	格式/自评/工具校验
无熔断	故障期延迟高	反复重试失败模型	熔断 + 半开
无抽象层	换模型改全代码	强耦合	Provider 接口
升级率过高	成本没省下	分类器不准	分析并调阈值
忽略延迟	级联变慢	未计入失败尝试	延迟预算
差异未归一	换模型行为变	未处理差异	统一内部格式

8.1 关于"级联的延迟"

级联的延迟是累加的：

lite 尝试（1.5s）→ 检查失败 → mid 尝试（3s）→ 通过
总延迟 = 4.5s

这对交互式场景可能是不可接受的。三个缓解手段：

1. 并行预取：同时发起 lite 与 mid，先到先用（成本上升）；
2. 流式检查：在流式输出中提前判断质量（如格式错误可提前终止）；
3. 限制升级次数：最多升级一次。

九、局限

1. 分类器需要标注数据。判断"任务难度"本身是个建模问题，需要样本。
2. 质量检查不可靠。格式对不代表内容对，"自评置信度"尤其不可靠。
3. 模型差异难以完全归一。工具调用、结构化输出等行为仍有细微差别，换模型可能改变行为。
4. 多模型增加运维复杂度。每个模型的限流、配额、故障模式都不同。
5. 成本节约可能被延迟抵消。级联省了钱但慢了，对体验敏感场景不划算。
6. 路由决策本身有成本。如果用一个模型来判断该用哪个模型，又是一笔开销。
7. 模型迭代快。今天的最优路由策略，可能因为模型更新而失效，需要持续校准。

十、小结

• 模型选择有四个维度：能力、价格、延迟、稳定性，无法同时最优，必须按场景权衡。
• 四种路由策略：静态规则、分类器、级联、并行择优；推荐组合是分类器粗筛 + 级联兜底。
• 路由应偏保守：误判代价不对称——把难任务分给小模型是质量事故，把简单任务分给大模型只是浪费。
• 级联的核心是质量检查机制；其代价是延迟累加，可用并行预取、流式检查、限制升级次数缓解。
• 降级有五个触发条件，注意方向：限流/故障/超时是降级，质量不达标是升级。
• 熔断要有三态（CLOSED/OPEN/HALF_OPEN），"半开"让系统能自动恢复。
• 路由的前提是模型抽象层：上层不依赖具体模型，统一统计与差异归一。
• 核心度量是升级率：过高说明分类器激进，并要关注升级成本占比（可能少量升级消耗大量成本）。
  

十一、动手练习

1. 运行 static_route.py，为"总结一段 3000 字文档"和"解释一个算法原理"分别设计规则，验证路由结果。
2. 运行 classifier_route.py，把权重调整成"更保守"，观察路由分布变化，说明代价。
3. 运行 cascade.py，把 min_quality 提到 0.9，观察 pro 使用次数与总成本的变化。
4. 为 FallbackChain 实现 HALF_OPEN 状态：熔断后冷却 30 秒，放行一个探测请求，成功则恢复。
5. 用 route_quality.py 分析你手头的路由日志，计算升级成本占比，给出优化建议。