模型路由看似简单,真正困难的是把决策变成可控系统

2026-07-16 25 预计阅读时间: 1 分钟
来源: huggingface.co AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

模型路由的初版通常只需要一个条件判断:简单请求交给便宜模型,复杂请求交给能力更强的模型。但系统一旦进入生产环境,路由器就必须同时面对质量、延迟、成本、上下文长度、工具支持、配额和故障恢复。此时,问题已经不再是“选哪个模型”,而是“如何在不确定条件下执行可解释、可回退、可观测的决策”。

由于来源只提供了标题而没有具体实现细节,下面的策略和代码属于一种可实践的工程方案,不代表原文采用了相同设计。

一条 if 语句为什么很快失效

最直观的路由规则是按提示词长度分流:

model = "small-model" if len(prompt) < 500 else "large-model"

它可以运行,却把几个不同维度混在了一起。短提示词可能要求证明数学定理,长提示词也可能只是摘要一份格式固定的日志。字符数只能近似反映输入规模,不能可靠表示推理难度。

生产中的模型选择通常至少受到以下约束:

  • 任务能力:模型是否支持工具调用、视觉输入、结构化输出或长上下文。
  • 服务目标:交互请求可能要求两秒内返回,离线任务则更关注正确率。
  • 预算约束:单次成本、租户额度和整条工作流的累计成本都可能影响选择。
  • 运行状态:限流、区域故障和延迟抖动会让原本合适的模型暂时不可用。
  • 业务风险:代码修改、财务分析等高风险任务不能只按最低价格路由。

因此,路由器不应只返回模型名称。它还应返回选择原因、候选模型、预算估算和回退路径,让一次决策能够被记录和复盘。

把硬约束与偏好排序分开

一个更稳定的设计可以分成两步。

第一步执行硬过滤,删除不满足条件的模型。例如,请求包含图片时,只保留支持视觉输入的模型;上下文超过限制时,直接排除窗口不足的模型。这些条件不能通过“价格更低”来补偿。

第二步对剩余候选项评分。延迟、价格和质量属于偏好,可以按业务场景赋予不同权重:

score = quality_weight × quality
      - cost_weight × estimated_cost
      - latency_weight × observed_latency

评分不是客观真理,而是业务策略的显式表达。在线客服可能提高延迟权重,批量代码审查可能提高质量权重。关键在于保留评分分解结果,否则线上出现质量下降时,很难判断是模型能力变化、价格配置错误,还是延迟指标把流量推向了另一个候选项。

还要避免把所有条件写进一个不断膨胀的路由函数。模型能力适合放在配置中,实时健康状态来自监控系统,业务风险由请求分类器提供,最终决策层只负责组合这些信号。

一个可运行的最小路由器

下面的示例只使用 Python 标准库,可以直接保存为 router.py 并运行。这里假设模型的质量分、单位价格和延迟数据由团队自行维护;接入真实系统时,应把静态延迟替换为滚动窗口中的实测指标。

from dataclasses import dataclass
from typing import Iterable


@dataclass(frozen=True)
class Model:
    name: str
    max_tokens: int
    supports_tools: bool
    quality: float          # Internal benchmark score: 0.0 to 1.0
    cost_per_1k: float      # Assumed cost unit
    latency_ms: int         # Replace with observed p95 latency
    healthy: bool = True


@dataclass(frozen=True)
class Request:
    estimated_tokens: int
    needs_tools: bool
    risk: str               # low, medium, high
    latency_target_ms: int


MODELS = [
    Model("fast-small", 8_000, False, 0.68, 0.20, 350),
    Model("tool-medium", 32_000, True, 0.82, 1.20, 900),
    Model("reasoning-large", 128_000, True, 0.94, 4.50, 2_400),
]


def eligible(models: Iterable[Model], request: Request) -> list[Model]:
    return [
        model
        for model in models
        if model.healthy
        and model.max_tokens >= request.estimated_tokens
        and (not request.needs_tools or model.supports_tools)
    ]


def score(model: Model, request: Request) -> tuple[float, dict[str, float]]:
    quality_weight = 8.0 if request.risk == "high" else 4.0
    cost_penalty = model.cost_per_1k * request.estimated_tokens / 1_000
    latency_penalty = max(0, model.latency_ms - request.latency_target_ms) / 1_000

    parts = {
        "quality": quality_weight * model.quality,
        "cost": -cost_penalty,
        "latency": -2.0 * latency_penalty,
    }
    return sum(parts.values()), parts


def route(request: Request) -> dict:
    candidates = eligible(MODELS, request)
    if not candidates:
        raise RuntimeError("No model satisfies the hard constraints")

    ranked = []
    for model in candidates:
        total, parts = score(model, request)
        ranked.append((total, model, parts))
    ranked.sort(key=lambda item: item[0], reverse=True)

    total, selected, parts = ranked[0]
    return {
        "selected_model": selected.name,
        "score": round(total, 3),
        "score_parts": {key: round(value, 3) for key, value in parts.items()},
        "fallbacks": [item[1].name for item in ranked[1:]],
    }


if __name__ == "__main__":
    request = Request(
        estimated_tokens=6_000,
        needs_tools=True,
        risk="high",
        latency_target_ms=1_500,
    )
    print(route(request))

运行命令:

python router.py

这个示例刻意把硬约束、评分和排序拆开。实际接入模型 API 时,可以根据 selected_model 调用对应供应商,并按 fallbacks 顺序处理可重试故障。认证失败、参数错误等确定性错误不应盲目回退,否则只会放大成本和流量。

回退机制不能等同于捕获所有异常

模型路由的复杂性常常在主模型失败后暴露出来。简单的 except Exception 加备用模型可能导致重复提交、预算失控,甚至在具有副作用的工具调用中重复执行操作。

可以这样划分错误:

  • 429、连接超时和部分 5xx 通常可以在退避后重试,或切换到兼容模型。
  • 上下文超限应触发截断、摘要或重新路由,而不是原样发送给下一个同样受限的模型。
  • 鉴权失败和无效参数需要停止请求并修复配置。
  • 工具已经执行但响应中断时,必须依赖幂等键或调用记录判断是否能够重试。

回退模型还需要满足原请求的硬约束。如果主模型支持工具调用,而备用模型不支持,切换成功并不意味着任务成功。路由器应保存一组经过能力过滤的有序候选项,而不是维护一个全局的“备用模型”。

观测路由结果,而不只是模型响应

只记录最终模型和 HTTP 状态码不足以调试路由系统。建议为每次决策记录以下字段:

{
  "request_id": "req-123",
  "route_policy_version": "2025-03-01",
  "task_class": "code_review",
  "risk": "high",
  "selected_model": "reasoning-large",
  "eligible_models": ["tool-medium", "reasoning-large"],
  "score_parts": {
    "quality": 7.52,
    "cost": -27.0,
    "latency": -1.8
  },
  "fallback_count": 0,
  "estimated_tokens": 6000,
  "actual_tokens": 5412,
  "latency_ms": 2180
}

其中的策略版本很重要。模型目录、权重或阈值变化后,团队需要区分“模型变差了”和“路由规则改变了”。日志还应避免写入原始敏感提示词;可记录任务标签、长度、哈希和脱敏后的特征。

评估时不要只看平均成本。至少应同时观察任务成功率、人工接受率、延迟分位数、回退率和单位成功请求成本。低价模型如果需要频繁重试,最终可能比一次调用高能力模型更贵。

上线前的工程检查

模型路由适合从少量可解释规则开始,再根据生产数据扩展。上线时可以检查这些问题:

  • 每个模型的能力、上下文窗口和健康状态是否由配置或服务目录统一维护?
  • 路由结果是否包含原因、候选项和策略版本?
  • 回退是否仅处理明确可重试的错误,并遵守原始能力约束?
  • 是否设置单次请求与整条工作流的成本上限?
  • 是否能用历史流量离线重放新策略,再进行小比例灰度?
  • 质量指标是否按任务类型分别统计,而不是只看全局平均值?

模型路由的条件表达式并不难,困难的是管理条件背后的不确定性。把硬约束、偏好评分、故障回退和观测数据拆成清晰组件,才能让路由策略在模型价格、能力和服务状态不断变化时仍然可控。


相关推荐