Gemini 4 Argon 与旧模型并存——请求路由该按任务分还是按成本分?

文章导读
按任务分还是按成本分,取决于任务对返回格式的容忍度。旧模型还在跑线上,通常说明它的输出结构已经被下游解析器、评测脚本或人工流程认下来了;把这类请求整体挪到 Gemini 4 Argon,账单可能好看,但失败模式会变多,出问题时只能逐个请求翻日志。更可控的顺序是:先按任务类型分,同一任务类型内部再按预算或延迟做二级选择,并且每次选择都留下可追溯的判定依据。这个判断本身可以从现有日志里验证,不必先拍板
📋 目录
  1. 一 先列出当前请求的类型清单,标出哪些任务值得换模型
  2. 二 写一个路由函数骨架:按任务标签选模型名,默认走旧模型
  3. 三 在路由层记录决策日志:选中的模型、判定依据、耗时
  4. 四 用同一批输入分别走两条路由,比对返回格式与失败表现
  5. 五 给路由加降级分支:新模型报错时自动改走旧模型并打点
A A

按任务分还是按成本分,取决于任务对返回格式的容忍度。旧模型还在跑线上,通常说明它的输出结构已经被下游解析器、评测脚本或人工流程认下来了;把这类请求整体挪到 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 直接取路由函数的返回值,这样日志和代码是同一份依据,不会各说各话。

Gemini 4 Argon 与旧模型并存——请求路由该按任务分还是按成本分?
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;降级后返回结构仍不符合下游解析;降级次数随时间持续上升。出现任一种情况,先把该标签从放行表里摘掉,再回到对照记录表定位原因,而不是靠扩大降级范围硬扛。