按任务分还是按成本分,取决于任务对返回格式的容忍度。旧模型还在跑线上,通常说明它的输出结构已经被下游解析器、评测脚本或人工流程认下来了;把这类请求整体挪到 Gemini 4 Argon,账单可能好看,但失败模式会变多,出问题时只能逐个请求翻日志。更可控的顺序是:先按任务类型分,同一任务类型内部再按预算或延迟做二级选择,并且每次选择都留下可追溯的判定依据。这个判断本身可以从现有日志里验证,不必先拍板。
路由维度建议先按任务类型分,成本只在同类型任务内部做二级选择。做法是:给现有请求打任务标签,默认全部走旧模型,只对返回格式已对齐、失败可回退的任务放行 Gemini 4 Argon;每次路由写入任务标签、判定依据、选中模型和耗时。边界是:格式差异未做完对照验证前不放量,降级分支和打点未接好前不切主链路;成本指标只有在任务标签稳定之后才具备可比性。
先列出当前请求的类型清单,标出哪些任务值得换模型
路由维度要来自真实流量,而不是来自对新模型能力的印象。先别改代码,从现有应用日志、网关访问记录或调用方统计里把请求按任务归纳成一张清单。清单至少三列:任务标签、调用频次级别、对返回格式的依赖程度。任务标签用稳定短名,例如 summarize、classify、extract_json,不要用自然语言描述,否则后面日志里没法聚合。频次级别用高/中/低或近似量级即可,不需要精确数字。返回格式依赖度指下游是否按固定 JSON 字段、固定枚举值或固定分隔符解析。
| 任务标签 | 频次级别 | 返回格式依赖 | 是否候选 |
|---|---|---|---|
| summarize | 高 | 低,纯文本给人看 | 是 |
| classify | 高 | 中,输出固定枚举 | 是 |
| extract_json | 中 | 高,下游按字段解析 | 先否 |
| untagged | 未知 | 未知 | 否 |
统计方式可以先用命令行粗筛,字段名替换成你日志里真实的键:
# 从应用日志按任务标签计数(字段名按实际替换)
grep -o 'task_tag=[a-zA-Z_]*' app.log | sort | uniq -c | sort -rn
# 若日志是 JSON 行,用 python 分组计数
python -c 'import json,collections;c=collections.Counter(json.loads(l).get("task_tag") for l in open("app.log"));print(c.most_common())'
验证方式:清单里的每个标签都能在日志里找到对应的原始记录,抽三条人工核对标签是否打得一致。找不到记录的标签说明只是设计稿上存在,不应该进入候选。
写一个路由函数骨架:按任务标签选模型名,默认走旧模型
放行表只写明确放行的任务标签,任何未命中的标签都回落旧模型。这样新增任务不会因为配置漏写而误走新模型,出问题时的搜索范围也有限。
DEFAULT_MODEL = "legacy-model-v1" # 兜底:现在线上的旧模型
ALLOWLIST = {
"summarize": "gemini-4-argon",
"classify": "gemini-4-argon",
}
def pick_model(payload):
# 返回 (模型名, 判定依据),依据用于写日志
task = payload.get("task_tag") or "untagged"
model = ALLOWLIST.get(task)
if model is None:
return DEFAULT_MODEL, "not_allowlisted"
return model, "allowlist_hit:" + task
替换项:DEFAULT_MODEL 换成当前的旧模型名,ALLOWLIST 的值换成新模型名,标签名与上一步统计到的保持一致。这个函数只做选择、不做调用,便于单独写测试用例。注意这里不对任何模型的能力下判断,是否放行完全由清单和对照实验的结果决定。
在路由层记录决策日志:选中的模型、判定依据、耗时
决策日志至少包含这些字段:request_id、task_tag、chosen_model、reason、prompt 或配置版本、调用方、开始时间与耗时。reason 直接取路由函数的返回值,这样日志和代码是同一份依据,不会各说各话。
import json, logging, time
log = logging.getLogger("router")
def route(payload):
model, reason = pick_model(payload)
started = time.monotonic()
log.info(json.dumps({
"event": "route_decision",
"request_id": payload.get("request_id"),
"task_tag": payload.get("task_tag"),
"chosen_model": model,
"reason": reason,
"prompt_version": payload.get("prompt_version"),
}, ensure_ascii=False))
return model, reason, started
耗时建议在调用返回后再补一条完成日志,把 started 换算成毫秒,避免把排队时间误算成模型耗时。回溯一次请求走了哪条路,按任务标签或请求 ID 检索即可:
# 按任务标签回溯最近若干条路由决策
grep '"task_tag": "summarize"' router.log | tail -n 20
# 按 request_id 查单次请求
grep 'req-8f3a' router.log
验证方式:挑一条已完成请求的 request_id,路由日志里应当只有一条对应的 route_decision,reason 与放行表配置对得上。出现多条或找不到,先修日志与调用链,再谈放量。
用同一批输入分别走两条路由,比对返回格式与失败表现
放量前用同一批输入各跑一次旧模型和新模型。样本从日志里抽样并脱敏,固定成一份文件,保证两次调用输入完全一致,差别只在被强制指定的模型名。执行方式可以是一个脚本,读样本、按 force_model 参数分别调用,把两次结果写进同一张记录表。
- sample_id:样本编号,便于回查原文
- task_tag:任务标签
- structure_ok:返回能否按约定解析,是或否
- missing_fields:约定字段中缺失的部分
- fail_type:超时、限流、服务端错误、结构不符、内容截断等分类
- retry_count:该样本的调用重试次数
- latency_ms:单次调用耗时
- notes:人工判断是否可接受
比对重点不是笼统判断哪个模型更好,而是同一任务在两个模型上结构是否一致、失败类型是否落在你的重试和降级逻辑能覆盖的范围里。structure_ok 出现差异时,先改解析器还是先改提示词,取决于下游能否容忍两种结构并存。这一步建议用本地样本或影子流量执行,不要直接对生产用户请求双跑,避免重复计费和重复副作用。
给路由加降级分支:新模型报错时自动改走旧模型并打点
降级的作用是把新模型的不确定性挡住,不是提升效果。触发条件收敛到明确的几类异常,不要用裸 Exception 全兜;结构不符对下游来说和调用失败没区别,也应算触发条件。
FALLBACK_ERRORS = ("Timeout", "RateLimit", "Server5xx", "ShapeMismatch")
def call_with_fallback(payload, call_fn):
model, reason = pick_model(payload)
try:
return call_fn(model, payload)
except Exception as exc:
if model == DEFAULT_MODEL or type(exc).__name__ not in FALLBACK_ERRORS:
raise
emit_metric("route.fallback", {
"request_id": payload.get("request_id"),
"task_tag": payload.get("task_tag"),
"from_model": model,
"to_model": DEFAULT_MODEL,
"error_class": type(exc).__name__,
})
return call_fn(DEFAULT_MODEL, payload)
打点字段里 request_id 和 task_tag 必须带上,否则降级只能看到总量,无法定位到具体任务。降级后需要人工复核的判据:同一 task_tag 在短时间内反复触发降级;出现此前没见过的 error_class;降级后返回结构仍不符合下游解析;降级次数随时间持续上升。出现任一种情况,先把该标签从放行表里摘掉,再回到对照记录表定位原因,而不是靠扩大降级范围硬扛。