跳到主内容

多模型灰度上线:从直接切流到安全切换的避坑实录 ​

问题背景:模型升级不是改一个 API 地址 ​

在 LLM 应用进入持续迭代阶段后,最危险的操作往往不是首次上线,而是“把旧模型换成新模型”。传统服务发布主要关注延迟、错误率和资源占用,但大模型切换还会引入回答风格漂移、工具调用格式变化、拒答率上升、长上下文遗忘、成本突增等问题。

典型场景是:线上稳定运行模型 A,现在希望将 10% 流量切到模型 B。如果只在网关层按随机比例转发,一旦 B 在某类意图上表现异常,可能要等到用户投诉才发现;如果直接全量切换,回滚又会丢失现场,难以判断是提示词、模型能力、检索结果还是工具协议造成的问题。

因此,多模型灰度不能只做“流量比例”,还要同时建设三件事:可控请求分流、线上可观测、离线与在线结合的效果回归。安全目标不是证明新模型绝对更好,而是在可观测、可比较、可回滚的条件下逐步放大流量。

方案对比:蓝绿、金丝雀与影子流量怎么选 ​

常见方案有三种。

第一,蓝绿发布。准备两套完整模型链路,通过网关一次性把流量从旧版本切到新版本。优点是回滚快,缺点是成本高,而且无法观察真实用户分布下的细粒度差异,适合风险较低、变更范围明确的场景。

第二,金丝雀发布。按用户、租户、地域或请求特征逐步放量,例如 1%、5%、20%、100%。它适合多模型灰度,因为可以把高风险用户与内部测试用户隔离,也便于按业务维度统计效果。

第三,影子流量。线上请求仍然由稳定模型回答,新模型只异步复制一份请求并记录结果,不返回给用户。它不会影响用户体验,适合早期验证提示词、输出格式、工具调用和成本,但影子结果不能完全代表真实用户满意度。

实践中推荐组合使用:先影子调用,再小比例金丝雀,最后按分层指标放量。任何阶段触发红线,都自动回滚到稳定模型。

线上请求用户/租户灰度分流器规则与哈希稳定模型 A候选模型 B指标与追踪延迟/成本/评分回归评测集黄金用例自动判决放量或回滚

分流设计:不要只随机,要按风险分层 ​

一个可落地的分流器至少需要四个输入:稳定版本、候选版本、灰度比例、分流键。分流键优先选择 user_id 或 tenant_id,而不是 request_id。因为同一用户在短时间内连续提问时,如果一会儿走 A、一会儿走 B,会感受到答案风格跳动,也难以统计会话级指标。

同时,要保留强制路由能力。例如内部员工、评测账号、高付费租户可以指定模型;包含敏感意图或高风险工具调用的请求应进入白名单逻辑,不参与早期灰度。

下面示例实现了一个最小可运行的内存版灰度路由器,支持稳定哈希、强制路由和灰度比例:

python
import hashlib
from dataclasses import dataclass
from typing import Dict, Optional

@dataclass(frozen=True)
class RouteRule:
    stable_model: str = "model-a"
    candidate_model: str = "model-b"
    canary_percent: int = 10

class ModelRouter:
    def __init__(self, rule: RouteRule):
        self.rule = rule

    def _bucket(self, route_key: str) -> int:
        # 稳定哈希:同一个用户在比例不变时始终落到同一模型
        digest = hashlib.sha256(route_key.encode("utf-8")).hexdigest()
        return int(digest[:8], 16) % 100

    def route(
        self,
        user_id: str,
        force_model: Optional[str] = None,
        risk_level: str = "normal",
    ) -> str:
        # 高风险流量先留在稳定模型,避免新模型早期直接影响业务
        if risk_level in {"high", "critical"}:
            return self.rule.stable_model

        # 评测账号或排障请求可强制指定模型
        if force_model:
            return force_model

        if self._bucket(user_id) < self.rule.canary_percent:
            return self.rule.candidate_model
        return self.rule.stable_model

router = ModelRouter(RouteRule(canary_percent=20))
for uid in ["u-001", "u-002", "u-003"]:
    print(uid, router.route(uid))

生产环境中,这段逻辑通常放在 API Gateway、BFF 或模型网关中。比例配置应支持动态下发,但每次变更要记录版本号、操作人、时间和规则内容,否则出现效果波动时无法复现当时的分流策略。

可观测埋点:把一次模型调用变成可分析事件 ​

LLM 可观测不能只记录 HTTP 状态码。一次模型调用至少应记录以下字段:request_id、session_id、user_id 的脱敏值、路由模型、提示词版本、检索文档版本、工具定义版本、输入 token、输出 token、首 token 延迟、总耗时、错误类型、缓存命中率、成本、业务意图、风险标签。

如果是智能体或工具调用场景,还要记录每一步 thought、action、observation 的结构化摘要。注意不要把用户隐私、密钥、身份证号等原文直接写入追踪系统,应在埋点前做脱敏或最小化存储。

下面示例用装饰器模拟一次模型调用的指标采集:

python
import time
import random
from functools import wraps
from typing import Callable, Dict, Any

traces: list[Dict[str, Any]] = []

def observe(model: str, prompt_version: str) -> Callable:
    def decorator(fn: Callable) -> Callable:
        @wraps(fn)
        def wrapper(request: Dict[str, Any], **kwargs):
            start = time.perf_counter()
            status = "success"
            error_type = None
            output = None
            try:
                output = fn(request, **kwargs)
                return output
            except Exception as exc:
                status = "error"
                error_type = type(exc).__name__
                raise
            finally:
                latency_ms = (time.perf_counter() - start) * 1000
                traces.append({
                    "request_id": request["request_id"],
                    "route_key_hash": hash(request["user_id"]),
                    "model": model,
                    "prompt_version": prompt_version,
                    "intent": request.get("intent", "unknown"),
                    "latency_ms": round(latency_ms, 2),
                    "status": status,
                    "error_type": error_type,
                    # 真实系统应从模型 usage 字段读取 token 与成本
                    "input_tokens": random.randint(120, 600),
                    "output_tokens": random.randint(80, 500),
                    "answer_preview": (output or "")[:80],
                })
        return wrapper
    return decorator

@observe(model="model-b", prompt_version="prompt-2025-01")
def fake_chat(request: Dict[str, Any]) -> str:
    time.sleep(random.uniform(0.05, 0.2))
    if random.random() < 0.05:
        raise RuntimeError("upstream timeout")
    return f"{request['intent']} 的处理结果"

for i in range(5):
    try:
        fake_chat({
            "request_id": f"req-{i}",
            "user_id": f"user-{i % 3}",
            "intent": "refund_policy",
        })
    except RuntimeError:
        pass

print(traces[:2])

有了这些事件后,看板不能只看全局平均值。至少要按意图、会话长度、是否命中 RAG、是否调用工具、新老模型分组对比。平均值很容易掩盖问题:例如整体满意度只下降 1%,但“退款政策”意图可能下降 15%。

回归评测:线上分流之外必须有黄金用例 ​

灰度发布解决的是“真实流量下观察”,回归评测解决的是“关键能力不能退化”。评测集不应只堆大量随机问题,而应分层建设:核心业务黄金用例、历史故障回放用例、对抗与安全用例、长上下文用例、工具调用结构化用例、边界与拒答用例。

每条样本至少包含输入、上下文、期望行为、评分方式和所属标签。对于事实性问题,可以用关键词、引用来源或结构化字段判断;对于开放生成,建议使用规则评分加 LLM-as-a-judge,并抽样人工复核。

下面示例展示一个轻量回归流程:

python
from dataclasses import dataclass

@dataclass
class EvalCase:
    case_id: str
    intent: str
    question: str
    must_include: list[str]
    forbidden: list[str]

eval_cases = [
    EvalCase(
        case_id="refund-001",
        intent="refund_policy",
        question="订单签收 7 天内如何退款?",
        must_include=["7天", "申请退款"],
        forbidden=["一定能退", "无需审核"],
    )
]

def rule_based_score(answer: str, case: EvalCase) -> dict:
    missing = [x for x in case.must_include if x not in answer]
    violated = [x for x in case.forbidden if x in answer]
    passed = not missing and not violated
    return {
        "case_id": case.case_id,
        "passed": passed,
        "missing": missing,
        "violated": violated,
    }

def run_regression(generate_fn) -> list[dict]:
    results = []
    for case in eval_cases:
        answer = generate_fn(case.question)
        results.append(rule_based_score(answer, case))
    return results

# 模拟候选模型输出,真实项目中替换为模型网关调用
results = run_regression(lambda q: "您可以在签收7天内申请退款,系统会进行审核。")
pass_rate = sum(r["passed"] for r in results) / len(results)
print({"pass_rate": pass_rate, "results": results})

放量门槛建议同时包含系统指标、效果指标和业务指标。例如:P95 延迟不超过旧模型 1.2 倍,错误率不高于旧模型 0.5 个百分点,核心回归通过率不低于 98%,高风险意图不得出现新增违规,单请求成本增幅不超过预算。满足门槛才从 5% 提升到 20%,否则停止放量并分析差异样本。

避坑总结:多模型切换最容易踩的 6 个坑 ​

  1. 只按 request_id 随机分流,导致同一用户体验抖动。 现象是用户连续追问时,新老模型回答口径不一致,客服侧看起来像“系统失忆”。修正方法是使用 user_id、tenant_id 或 session_id 做稳定哈希,并在会话生命周期内固定模型。

  2. 只监控错误码,不监控语义退化。 现象是新模型 HTTP 200、延迟正常,但答案变短、拒答增多、工具参数 JSON 字段丢失。修正方法是埋点记录输出长度、工具调用成功率、结构化解析失败率、意图级评分和回归通过率。

  3. 评测集长期不更新,模型“刷榜”但线上变差。 现象是离线分数很高,真实新问题却频繁失败。修正方法是把线上低分会话、人工修正记录、用户负反馈和故障案例持续沉淀为 eval-dataset,并按版本冻结一份用于横向对比。

  4. 灰度比例直接写死在代码里。 现象是发现异常后需要重新发版才能回滚,恢复时间被拉长。修正方法是将模型路由、提示词版本和灰度比例配置化,并提供一键回滚、变更审计和最小放量策略。

  5. 候选模型与旧模型的参数没有对齐。 现象是新模型看起来“更发散”或“更保守”,实际只是 temperature、top_p、max_tokens、系统提示词不同。修正方法是回归时固定解码参数、提示词版本和工具定义,并把这些版本写入 trace。

  6. LLM-as-a-judge 既当运动员又当裁判。 现象是同一厂商新模型评分显著提高,但人工复核不认可。修正方法是裁判模型与候选模型隔离,给出明确评分 rubric,保留理由与置信度,并对争议样本进行人工抽检。

结语 ​

多模型灰度上线的核心,不是把流量按比例转发出去,而是建立“分流—观测—评测—判决—回滚”的工程闭环。影子流量帮助我们在不影响用户的情况下发现格式与成本问题,金丝雀发布帮助我们观察真实分布下的业务差异,回归评测则保证关键能力不会在一次模型升级中悄悄退化。

当请求路由、提示词、检索、工具和模型版本都能被追踪,当每一次放量都有明确门槛和回滚路径,模型切换才从一次高风险操作,变成可持续、可审计、可复盘的 LLM Ops 常规流程。

免责声明

本工具与文章内容仅用于学习测试场景,优化建议仅供参考,请勿直接用于生产环境,由此产生的风险自行承担。