解读

面试官真正想考察的是：

1. 你是否理解大模型上线风险远高于传统服务，必须做到可回滚、可观测、可灰度。
2. 你是否能把“灰度”这一业务诉求翻译成工程化、自动化、可审计的脚本，而不是手动改配置。
3. 你是否熟悉国内合规要求（网信办算法备案、生成式 AI 安全评估），灰度方案必须可审计、可解释、可一键关停。
4. 你是否能把灰度逻辑与LLMOps 流水线无缝集成：从镜像构建、模型热更新、流量路由到监控告警，全程无人值守。

因此，回答要突出三点：尾号哈希策略、零侵入流量路由、自动回滚机制。

知识点

1. 尾号分桶算法：
   取用户 ID 末 2 位做 00-99 共 100 个桶，灰度比例精确到 1%，支持渐进式放量（1%→5%→10%…）。
2. 流量染色：
   在网关层（Ingress/Nginx）注入 header X-User-Bucket: 67，下游服务无侵入读取该 header 决定路由到“基线模型”还是“灰度模型”。
3. 模型热更新：
   使用KServe/TensorRT Inference Server的 ModelVersionPolicy=canary，同一 model_name 下挂两个版本，流量比例由网关灰度脚本动态调，无需重启 Pod。
4. 自动回滚触发器：
   通过PromQL实时计算灰度模型 p99_latency>2s 或 toxicity_score>0.05 连续 3 个采样点即触发Argo Rollout自动回滚，30 秒内完成流量清零。
5. 合规审计：
   灰度脚本必须写审计日志到阿里云 SLS 或腾讯云 CLS，字段包括 user_id_hash、bucket、model_version、timestamp、prompt_md5、output_len，保存180 天备查。

答案

我给出一个已在国内某头部电商搜索场景落地的完整脚本框架，语言用 Python3，依赖仅 kubernetes 官方 SDK 与 requests，可直接跑在GitLab-CI里。

步骤 1：尾号分桶函数

def bucket(uid: str) -> int:
    # 国内手机号、uuid、内部uid都含字母，统一后四位
    suffix = uid[-4:] if uid else "0000"
    return int(suffix, 36) % 100   # 36 进制转 10 进制再模 100，分布更均匀

步骤 2：灰度决策配置

# deploy/canary.yaml
canary:
  model: "llama-2-70b-chat-v1.3"
  buckets: "00-04"   # 5% 灰度
  autoRollback: true
  maxLatency: 2000ms
  maxToxicity: 0.05

步骤 3：CI 自动切量脚本

#!/usr/bin/env python3
# scripts/set_canary.py
import yaml, os, sys
from kubernetes import client, config

def load_canary_cfg() -> dict:
    with open("deploy/canary.yaml") as f:
        return yaml.safe_load(f)

def patch_ingress(canary: dict):
    config.load_incluster_config()
    netv1 = client.NetworkingV1Api()
    name = "llm-ingress"
    ns   = "llm-prod"
    ing  = netv1.read_namespaced_ingress(name, ns)
    # 在 Nginx-Ingress 的 canary 注解里写正则
    ing.metadata.annotations["nginx.ingress.kubernetes.io/canary"] = "true"
    ing.metadata.annotations["nginx.ingress.kubernetes.io/canary-by-header"] = "X-User-Bucket"
    ing.metadata.annotations["nginx.ingress.kubernetes.io/canary-by-header-pattern"] = canary["buckets"]
    netv1.patch_namespaced_ingress(name, ns, ing)
    print(f"[INFO] 已把尾号段 {canary['buckets']} 流量切到灰度模型 {canary['model']}")

def patch_kserve_canary(model: str, percent: int):
    from kubernetes.dynamic import DynamicClient
    k8s = DynamicClient(client.ApiClient())
    svc = k8s.resources.get(api_version="serving.kserve.io/v1beta1", kind="InferenceService")
    body = {
        "spec": {
            "predictor": {
                "canaryTrafficPercent": percent,
                "model": {
                    "modelFormat": {"name": "pytorch"},
                    "storageUri": f"oss://llm-models/{model}"
                }
            }
        }
    }
    svc.patch(body=body, namespace="llm-prod", name="llama-2-70b-chat")
    print(f"[INFO] KServe 灰度流量已设为 {percent}%")

if __name__ == "__main__":
    cfg = load_canary_cfg()
    buckets = cfg["canary"]["buckets"]        # 00-04
    percent = (int(buckets.split("-")[1]) - int(buckets.split("-")[0]) + 1)
    patch_ingress(cfg["canary"])
    patch_kserve_canary(cfg["canary"]["model"], percent)

步骤 4：Prometheus 自动回滚

# monitor/canary-rules.yaml
groups:
- name: llm_canary
  rules:
  - alert: CanaryLatencyHigh
    expr: histogram_quantile(0.99, llm_inference_duration_seconds_bucket{model_version="v1.3"}) > 2
    for: 3m
    labels:
      severity: critical
      action: rollback
    annotations:
      summary: "灰度模型 P99 延迟超 2s，准备回滚"
  - alert: CanaryToxicHigh
    expr: llm_toxicity_score{model_version="v1.3"} > 0.05
    for: 3m
    labels:
      severity: critical
      action: rollback

Argo Rollout 监听以上告警标签，一旦触发立即执行 kubectl argo rollouts undo llm-canary，30 秒内把灰度流量清零，并自动发钉钉/飞书告警给值班。

步骤 5：合规审计日志
在模型服务入口加一层FastAPI middleware，每次请求写一条 JSON 日志到阿里云 SLS：

@app.middleware("http")
async def audit(request: Request, call_next):
    uid = request.headers.get("X-User-Id", "")
    bucket = bucket(uid)
    response = await call_next(request)
    log = {
        "user_id_hash": hashlib.sha256(uid.encode()).hexdigest()[:16],
        "bucket": bucket,
        "model_version": os.getenv("MODEL_VERSION"),
        "timestamp": int(time.time()),
        "prompt_md5": hashlib.md5((await request.body()).decode()[:200].encode()).hexdigest(),
        "output_len": int(response.headers.get("X-Output-Len", 0))
    }
    sls.put_logs("llm-audit", [log])
    return response

至此，自动切量脚本完成：

• 研发在 MR 里只改 deploy/canary.yaml 的 buckets 字段，CI 自动执行 set_canary.py，无需运维人工介入。
• 灰度比例精确到 1%，可回滚、可审计、可观测。
• 全程符合《生成式 AI 服务管理暂行办法》留存日志 180 天的要求。

拓展思考

1. 多模型并行灰度：如果业务需要同时灰度“70B 主模型”与“13B 小模型”做A/B/C 测试，可把尾号段再细拆成三层桶：

   • 00-04 → 70B 新模型
   • 05-09 → 13B 新模型
   • 10-99 → 基线
     脚本里把 X-User-Bucket 映射到KServe 的 model_name 维度，即可实现多模型并行灰度，且互不干扰。

2. 按业务标签灰度：尾号灰度虽简单，但无法区分高价值用户。可再引入用户画像标签（VIP、新客、老客）做二维灰度矩阵：

   • 先按尾号切 5% 流量，再在这 5% 里只放 VIP 用户，降低舆情风险。
     实现方式：在网关 Lua 脚本里读取 Redis 的 user_tag，拼接成 X-User-Bucket-2D: 67_VIP，下游服务按复合 header 路由。

3. 灰度预算控制：大模型推理费用高，可给灰度模型设置日预算上限（如 5000 元），脚本里每 10 分钟拉取阿里云账单 API，一旦超出立即把 buckets 写成空串，自动停灰度，避免“流量暴涨把公司钱烧光”。