LLM 评测体系

第 18 章 把 Evals 变成 Quality Gate:CI 集成与回归告警

作者 杨艺韬 · 19,357 字 · 发布于 · 更新于

评测进了 CI 才算门禁。PR 时跑子集、合并前跑全集、夜间跑回归——三层策略要解决的是同一个矛盾:门禁要够快才有人用,够全才拦得住。

本章要点

  • 把评测从"工程师的玩具"升级为"产品流程的门禁"
  • GitHub Actions / GitLab CI 的完整 evals 工作流配置
  • 三层门禁策略:PR 时跑子集、合并前跑全集、夜间跑回归
  • 时序告警:Grafana / Prometheus 监控指标曲线断点
  • 指标版本化:让"分数变了"和"评测器变了"分开追踪
  • 组织流程:评测 owner 角色与 LLM SRE 职责

18.1 为什么要把 Evals 推进 CI

回到第 1 章 §1.7 的核心结论——LLM 应用失败的根因是"凭感觉调 prompt"。前 17 章给了你完整的工具箱:数据集、判分、元评测、源码、场景化、在线平台。但只要这些工具还停留在"工程师手动跑"层面,就一定会出现"周五急着上线、忘记跑评测、周一线上崩"的反复。

CI Quality Gate 是评测体系的最后一公里——它把"做不做评测"从工程师的判断升格为系统的强制:

flowchart LR
  Dev[开发者推 PR] --> CI[CI 自动跑评测]
  CI -->|通过| Merge[允许合并]
  CI -->|失败| Block[阻止合并 + 报告失败 case]
  Block --> Fix[开发者修改]
  Fix --> Dev
  style CI fill:#fef3c7
  style Block fill:#fee2e2
  style Merge fill:#dcfce7

这套机制的工程价值:把"要不要做评测"的讨论降级为"要不要绕过 CI"的讨论——后者的政治成本远高于前者。

第 1 章 5 起事故里,至少 4 起本可以被一个完善的 CI Quality Gate 拦住。这不是事后诸葛——各家模型厂商公开的发布流程里,「发版前必须过一轮固定评测」都是明写的环节;只是具体门禁配置属于内部工程细节,公开材料里看不到,本章给的是可自己落地的那一份。

18.2 三层门禁策略

不是所有评测都该堵在 PR 上。一套合理的 Quality Gate 分三层:

flowchart TB
  Commit[每次 git push] --> L1[第 1 层: PR 子集<br/>50 题, 5 分钟]
  L1 -->|通过| L2[Merge to main<br/>触发第 2 层]
  L2 --> L3[第 2 层: 全集<br/>500 题, 30 分钟]
  L3 -->|通过| Deploy[允许部署到 staging]
  Cron[凌晨 3:00 cron] --> L4[第 3 层: 回归集 + 在线<br/>200 题, 1 小时]
  L4 -->|发现退化| Alert[Slack/PagerDuty 告警]
  style L1 fill:#dbeafe
  style L3 fill:#dcfce7
  style L4 fill:#fef3c7

每层各自的设计原则:

18.2.1 第 1 层:PR 子集(5 分钟硬上限)

  • 数据集:从黄金集随机抽 50 条 + 必跑的 10 条 critical case
  • 指标:核心 3-5 个(Faithfulness / Recall / Latency / Cost)
  • 目的:发现明显改坏,不阻塞开发节奏
  • 失败处理:阻止 PR merge,但允许 reviewer override(紧急 hotfix)

18.2.2 第 2 层:合并到 main 后的全集(30 分钟以内)

  • 数据集:完整 500 题黄金集 + 100 题对抗集
  • 指标:所有 10+ 个评测维度(含 §16 的 4 维度安全)
  • 目的:捕捉子集漏掉的失败模式
  • 失败处理:自动 revert merge / 触发紧急 hotfix 流程

18.2.3 第 3 层:每日回归(cron 触发,1 小时以内)

  • 数据集:固定 200 题回归集(不随时间变化的稳定标尺)
  • 指标:完整指标 + 与 7/30/90 天前对比
  • 目的:捕捉外部因素导致的静默退化(模型 API 升级、retriever 索引漂移)
  • 失败处理:PagerDuty 告警 → on-call 工程师确认是否回滚

18.3 一份完整的 GitHub Actions 配置

把三层门禁落到 GitHub Actions:

# .github/workflows/evals.yaml
name: LLM Evals Quality Gate

on:
  pull_request:
    paths:
      - 'src/**'
      - 'prompts/**'
      - '.github/workflows/evals.yaml'
  push:
    branches: [main]
  schedule:
    - cron: '0 3 * * *'  # 每日凌晨 3:00

jobs:
  pr-subset:
    if: github.event_name == 'pull_request'
    runs-on: ubuntu-latest
    timeout-minutes: 10
    permissions:
      contents: read
      pull-requests: write   # 下面要往 PR 上写评论
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -r requirements-eval.txt
      - name: Run subset eval (50 cases)
        # evals.run 指团队自己的评测入口脚本(示意),不是 openai/evals 的命令行
        run: |
          python -m evals.run \
            --dataset data/golden/v1.jsonl \
            --sample-size 50 \
            --include-critical data/critical-cases.jsonl \
            --metrics faithfulness,context_recall,answer_relevancy,latency,cost \
            --thresholds config/thresholds-subset.yaml \
            --report-file eval-report.json
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
      - name: Upload report
        uses: actions/upload-artifact@v4
        with:
          name: eval-report-pr
          path: eval-report.json
      - name: Comment failures on PR
        if: failure()
        uses: actions/github-script@v7
        with:
          script: |
            const report = require('./eval-report.json');
            const body = report.failures.map(f =>
              `- **[${f.metric}=${f.value}]** ${f.input}`
            ).slice(0, 20).join('\n');
            await github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: `## Evals failed (${report.failures.length} cases below threshold)\n\n${body}`,
            });

  full-eval:
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    timeout-minutes: 45
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -r requirements-eval.txt
      - name: Run full eval
        run: |
          python -m evals.run \
            --dataset data/golden/v1.jsonl \
            --adversarial data/adversarial/v1.jsonl \
            --metrics all \
            --thresholds config/thresholds-full.yaml \
            --report-file eval-report-full.json
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      - name: Push metrics to Prometheus
        run: |
          python scripts/push_metrics.py \
            --report eval-report-full.json \
            --pushgateway http://prom-push.internal:9091
      - name: Notify on failure
        if: failure()
        uses: slackapi/slack-github-action@v1
        with:
          payload: |
            { "text": "🚨 Main branch eval failed at commit ${{ github.sha }}" }
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

  daily-regression:
    if: github.event_name == 'schedule'
    runs-on: ubuntu-latest
    timeout-minutes: 90
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -r requirements-eval.txt
      - name: Run regression eval
        run: |
          python -m evals.run \
            --dataset data/regression/v1.jsonl \
            --compare-to data/regression/v1-baseline.json \
            --alert-threshold 0.05 \
            --report-file eval-report-regression.json
      - name: Page on regression
        if: failure()
        run: curl -X POST $PAGERDUTY_WEBHOOK -d "Eval regression detected"
        env:
          PAGERDUTY_WEBHOOK: ${{ secrets.PAGERDUTY_WEBHOOK }}

这份 yaml 把三层门禁完整实现:

  • pr-subset job:PR 触发,10 分钟超时,失败时自动评论 PR
  • full-eval job:合并到 main 触发,45 分钟超时,指标推到 Prometheus
  • daily-regression job:cron 触发,对比 baseline,超阈值触发 PagerDuty

18.4 时序告警:把退化做成可观测系统

CI 只能拦住"自己改坏"。模型供应商静默升级、retriever 索引漂移这种外部退化只能靠时序监控发现。

flowchart LR
  Eval[每日回归] -->|push| PG[Prometheus<br/>Pushgateway]
  PG --> Prom[Prometheus]
  Prom --> G[Grafana Dashboard]
  Prom --> AM[Alertmanager]
  AM -->|超阈值| PD[PagerDuty]
  AM -->|超阈值| Slack
  style PG fill:#dbeafe
  style G fill:#dcfce7
  style PD fill:#fee2e2

18.4.1 关键告警规则

# Prometheus alert rules
groups:
  - name: llm-evals
    rules:
      - alert: FaithfulnessRegression
        expr: |
          (avg_over_time(eval_faithfulness[1d])
           - avg_over_time(eval_faithfulness[7d] offset 1d))
          < -0.05
        for: 5m
        annotations:
          summary: "Faithfulness 7 日环比下降 > 5pp"

      - alert: HallucinationSpike
        expr: eval_hallucination_rate > 0.10
        for: 10m
        annotations:
          summary: "幻觉率超过 10%"

      - alert: JailbreakResistanceDrop
        expr: eval_jailbreak_pass_rate < 0.95
        for: 1m
        labels:
          severity: critical   # severity 必须放 labels,Alertmanager 按 label 路由;
                               # 写进 annotations 只是给人看的文本,路由规则匹配不到
        annotations:
          summary: "越狱抵抗率跌破 95%"

      - alert: CostBudgetExceeded
        expr: eval_cost_per_query_p95 > 0.05
        for: 30m

每条规则对应第 4、6、13、14、16 章讨论过的具体指标。注意 alert 的 for 字段——避免单次抖动触发告警,要持续超阈值才告警。

18.4.2 Grafana Dashboard 面板设计

一份完整的 Dashboard 至少要包含:

  • 总览面板:当日通过率 / Faithfulness / Recall / Latency / Cost 的当前值 + 7 天趋势
  • 退化面板:每个指标的"7d vs 7d ago"差异柱状图
  • 失败 case 列表:今日失败样例的可点击列表(链到 langsmith / langfuse trace)
  • 版本切换标记:模型 / prompt 版本切换的垂直辅助线(vertical annotations),方便看曲线断点

第 4 章 §4.12.5 讨论的"指标长期漂移"在 Grafana 上能直接看到——任何一条指标曲线突然出现断点(如垂直跳变 5pp),都是值得排查的信号。

18.5 指标版本化:让"分数变了"可被解释

工业评测的进阶问题:当指标曲线下降时,怎么区分"模型变差了" vs "评测器变了"?

第 4 章 §4.12.5 提出过版本化方案。在 CI 工程上落地:

# eval-config.yaml v3.2
metrics:
  faithfulness:
    version: "2.1.0"
    judge_model: "claude-sonnet-5"
    judge_prompt_hash: "sha256:a3f9b1c..."
    calibration_set_version: "v1.2"
    effective_from: "2026-04-15"

  context_recall:
    version: "1.0.0"
    judge_model: "gpt-4o-2024-08-06"
    # ...

dataset:
  golden:
    version: "v2.3"
    sha256: "8f3a..."
  regression:
    version: "v1.0-frozen"
    sha256: "7c2d..."

每次 PR 改动 prompt / judge / dataset,对应版本号必须 bump。这一份 config 跟着每个评测 run 一起被推到 Prometheus 作为 label。

时序告警就能区分"同版本下退化(模型问题)"vs"切换版本时跳变(评测器变化)"——后者不应触发告警。

18.6 三种 CI 集成的反模式

工业实战见过的几个常见错误:

18.6.1 反模式 A:评测时间过长,工程师都跳过

如果第 1 层 PR 评测要 30 分钟,团队会本能地学会"小改动直接绕过"——CI 还在跑就先 merge。

修法:第 1 层硬性 ≤ 5 分钟。需要长跑的评测推到第 2 层(合并后异步跑)。这就是"三层"分级的根本原因。

18.6.2 反模式 B:阈值太松,告警永远不响

刚搭起来的 Quality Gate 如果阈值设得太松(如 Faithfulness ≥ 0.5),永远不会触发——团队几个月后忘了它的存在。

修法:阈值要"略高于当前生产水平"。如果生产 Faithfulness = 0.85,阈值设 0.83——给改动 2pp 的容忍度,但稍微变差就告警。

这里要先算样本量。按 §2.3.4 的 Wald 公式,p≈0.85 时,50 题的 95% 置信区间半宽约 ±10pp,500 题约 ±3pp,要压到 ±2pp 需要约 1200 题。所以 2pp 的容忍度只对大样本层有意义;第 1 层 50 题子集要么用固定种子抽同一批题、与 main 在同一批题上配对比较(显著性检验见 §4.12.2),要么只用它拦"明显改坏",细粒度阈值留给全集层。

18.6.3 反模式 C:阈值太严,告警永远响

反向问题——阈值设得过严,每天都触发告警。一周后大家学会"忽略 Slack 频道",真正重要的告警混在噪声里被忽略。

修法:分等级告警。Critical 阈值(如 Jailbreak Pass < 0.90)必须人工响应;Warning 阈值(如 Latency > 5s)只记录、不发通知。第 4 章 §4.8 讨论的"硬阈值 + 主指标"是这一点的方法学源头。

18.7 组织流程:评测 Owner 与 LLM SRE

最后讨论评测体系不可省的"人":

graph TB
  A[评测体系组织角色] --> B[Eval Owner<br/>专人负责评测体系]
  A --> C[LLM SRE<br/>响应在线告警]
  A --> D[Domain Reviewer<br/>领域专家审核 hard case]
  B --> B1[维护数据集 + 版本]
  B --> B2[审核 PR 中的评测改动]
  B --> B3[每季度跑元评测仪式]
  C --> C1[on-call 处理 Quality Gate 告警]
  C --> C2[决定是否回滚 / 切换 judge]
  D --> D1[每周 review 失败 case]
  D --> D2[判定 hard case 入集]
  style B fill:#dbeafe
  style C fill:#dcfce7
  style D fill:#fef3c7

三个角色不一定是三个人——小团队可以一个人兼,但每个职责必须明确归属。没有 owner 的评测体系很容易在几个月内荒废。

18.8 评测的下一步:几个前沿方向

先回看一下本专栏的脉络:

flowchart LR
  P1[第一部分<br/>事故与必要性] --> P2[第二部分<br/>理论基础]
  P2 --> P3[第三部分<br/>判分方法学]
  P3 --> P4[第四部分<br/>开源框架源码]
  P4 --> P5[第五部分<br/>场景化实战]
  P5 --> P6[第六部分<br/>生产化与 CI]
  style P6 fill:#dcfce7

下一步——评测领域的前沿正在向几个方向演化:

  • Self-improving Evals:评测器自动从失败 case 中学习、自动扩充黄金集
  • Continuous Red Teaming:动态对抗性 prompt 生成永不停止,持续探测新越狱
  • Eval-driven Model Routing:基于实时 eval 信号在多个模型间动态路由(强模型 / 弱模型分流)
  • Cross-modal Evals:多模态 RAG / 多模态 Agent 的统一评测体系

这些方向都还没有标准答案(以上是对趋势的判断,不是既成事实)。本专栏覆盖的是 2026 年初已经成熟的方法学。

18.9 Quality Gate 的定位与设计原则

18.9.1 CI Evals 与传统单测的本质差异

很多团队把 CI Evals 当成"单元测试的扩展"来理解。这种理解有用,但忽视了几个本质差异——理解了才能避免把 CI Evals 跑成不能持续的负担:

维度 传统单测 CI Evals
执行成本 几乎为 0 每次跑都要付模型 API 费用
执行时间 毫秒-秒 分钟-小时
结果确定性 100% 重跑结果会波动(有 noise)
失败定位 精确到行 需要看 trace 追踪
阈值类型 二元(pass/fail) 概率(85% / 90% / 95%)
数据集维护 测试代码即一切 黄金集是独立工程资产

每条差异都对应一个工程改动需求:

  • 执行成本 → 必须分层,PR 子集只跑 50 条,不能 500 条全跑
  • 执行时间 → 必须并行 + 缓存,避免阻塞开发
  • 结果确定性 → 阈值 + 容忍区间,单次失败不立即 block,连续 N 次再 block
  • 失败定位 → CI 失败必须输出 trace 链接,不能只给"45/50 通过"
  • 阈值类型 → 阈值定义必须 yaml 化,不能 hardcode
  • 数据集维护 → 黄金集要有 owner + 版本控制 + PR 流程

把这些差异对应的工程改动一一落实,CI Evals 才能真正运转起来。否则它会沦为"摆设"——挂在那里没人看、想跑也跑不动。

18.9.2 Quality Gate 反模式:用通过率代替指标体系

最后一个常见反模式——团队为了"绿灯",把 CI Gate 简化成"75% 通过率即过"。

这种简化看起来友好,实则危险:

  • 25% 的失败 case 可能集中在极敏感场景(合规 / 安全),而非分散在长尾
  • 通过率不区分 "差一点点" vs "完全错"
  • 通过率随评测集变化、不可比

正确做法(贯穿本专栏所有章节)是 多指标硬阈值 + 失败 case 强制审查:

gate:
  required:
    faithfulness: { min: 0.85 }
    context_recall: { min: 0.90 }
    jailbreak_pass: { min: 0.95 }
    cost_per_query: { max: 0.02 }
  must_review:
    failure_count: { max: 5 }    # 失败超 5 条强制人工 review
  optional_optimize:
    answer_relevancy: { direction: maximize }

把"通过率"换成多个独立硬阈值 + 失败上限——既保留量化决策,又不让任何关键维度被通过率"稀释"。

18.9.3 CI Quality Gate 的"反模式"清单

把全章的反模式集中整理成一份 checklist——团队搭 CI Quality Gate 时常踩的坑:

□ 不要让 CI 跑超过 10 分钟 (开发者会绕过)
□ 不要把所有阈值都设最严 (会频繁 false positive)
□ 不要忘记 override 机制 (没逃生通道导致绕过)
□ 不要让告警频繁误报 (变成噪声被无视)
□ 不要把通过率作为唯一判断 (Simpson 悖论)
□ 不要把评测代码混进生产代码的包与进程 (互相污染,见 §9.10.1)
□ 不要忘记成本上限 (失控会爆账单)
□ 不要让阈值悄悄变松 (滑坡到名存实亡)
□ 不要让 grader prompt 一直不变 (drift 不可见)
□ 不要省略 CI 失败的 PR 评论 (开发者看不到失败 case)

10 条反模式都是 CI 评测里的常见坑。把这份 checklist 贴在 CI 设计文档里、做 review 时逐一对照。

18.9.4 Quality Gate 的"放行机制":必须有逃生通道

CI Quality Gate 的政治哲学问题:总有"必须现在合并"的紧急情况。Hotfix 修线上事故,合规要求 24 小时内必须 patch,PM 要求当天上线某个营销活动——这种情况评测可能因为种种原因失败(评测集本身有 bug、judge API 抖动、临时阈值不合理)。

如果 Quality Gate 没有 override 机制,团队会怎么做?

  • 紧急情况下直接绕过 CI(比如直接 push 到 main 分支)→ Quality Gate 形同虚设
  • 改 Quality Gate 阈值松到永远过 → 评测体系丧失意义

正确的 override 设计:

# config/eval-override-policy.yaml
# 自定义策略文件(示意),由门禁脚本读取执行;不是 GitHub Actions 语法,不能写进 workflow 文件
override:
  allowed_paths:
    - "src/hotfix/**"  # hotfix 路径下允许 override
  required_approvers: 2  # 必须 2 个 reviewer 同意
  reason_required: true  # 必须填 override 原因
  notify_after: "#evals-overrides"  # 自动通知 slack 频道
  expires_after: 7d  # 7 天后自动 revert (强制后续做评测)

这套机制保留"紧急情况能上线"的灵活性,又通过审批 + 通知 + 自动 revert 限制 override 的滥用。没有 override 机制的 Quality Gate 一定会被绕过、会被绕过的 Quality Gate 不如不要。

工业团队的实践:每月 review 一次 override 历史。如果某条评测频繁 override → 评测本身可能有问题(阈值太严 / 数据集有 bug),需要修;如果 override 集中在某个工程师 → 该工程师可能在偷工减料,需要管理介入。

18.9.5 Quality Gate 与 SRE 文化的对接:Error Budget 思维

CI Quality Gate 的工程哲学和 SRE(Site Reliability Engineering)有深层共鸣——都是"用数据决定能不能 ship"。具体借用 Google SRE 书的 Error Budget 概念:

定义:

SLI(Service Level Indicator): 当天回归评测 Faithfulness >= 0.85 记为"达标日"
SLO(Service Level Objective): 达标日占比 >= 90%
Error Budget = (1 - SLO) × 时间窗口
                = 10% × 30 天 = 3 天的预算

Error Budget 用法:

  • 当月评测失败 < Error Budget → 团队还有"试错空间",可以做激进改动
  • 当月评测失败 ≥ Error Budget → "冻结"激进改动,专注稳定性

这种思维让 Quality Gate 不只是"过 / 不过"的二元判断,而是用预算管理风险:

flowchart LR
  Month[本月 Quality Gate] --> Pass[失败次数]
  Pass --> Calc{< Error Budget?}
  Calc -->|是| Free[团队可激进迭代]
  Calc -->|否| Freeze[冻结非必要改动]
  Freeze --> Cool[让评测体系恢复]
  Cool --> Reset[下月重置预算]
  style Free fill:#dcfce7
  style Freeze fill:#fee2e2

工程实务:

  • 每月初公开 Error Budget 数字
  • 每次 Quality Gate 失败都"消耗"预算
  • 月末复盘"预算如何用、用得是否合理"
  • 持续超预算 → 升级到管理层,可能需要重构 / 招人

这种"用 SRE 思维管理评测"的方法,把"评测严格度"和"创新速度"做成了可量化的权衡。注意 0.85 是质量阈值,90% 是达标比例,两者不要混成一个数:用 1 − 0.85 当错误预算,就把阈值误当成了目标比例。

18.10 检查器与阈值

18.10.1 完整的 Quality Gate 检查器实现

整合本章方法学,给一份"CI Quality Gate 检查器"完整 Python 实现:

# quality_gate.py
import json
import yaml
import sys
from pathlib import Path

class QualityGate:
    # 直接吃 §18.10.2 那份 ci-thresholds.yaml 的分层结构:
    # quality_gate.<layer>.thresholds.<metric> = {min: x} 或 {max: y}
    def __init__(self, config_path: str, layer: str = "pr_subset"):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)
        self.layer = self.config["quality_gate"][layer]
        self.thresholds: dict = self.layer.get("thresholds", {})

    def check(self, eval_report_path: str) -> dict:
        with open(eval_report_path) as f:
            report = json.load(f)

        results = {
            "passed": True,
            "required_failures": [],
            "warnings": [],
            "metrics": {},
        }

        # 1. 检查所有必需指标
        for metric, bound in self.thresholds.items():
            actual = report.get(metric)
            if actual is None:
                results["required_failures"].append(
                    f"Missing metric: {metric}"
                )
                results["passed"] = False
                continue

            results["metrics"][metric] = actual

            # 越高越好的指标写 min,越低越好的(延迟 / 成本 / 幻觉率)写 max。
            # 方向写反了门禁会永远绿灯——这是最容易发生的静默失效
            if "min" in bound and actual < bound["min"]:
                results["required_failures"].append(
                    f"{metric}={actual:.3f} < min {bound['min']}"
                )
                results["passed"] = False
            if "max" in bound and actual > bound["max"]:
                results["required_failures"].append(
                    f"{metric}={actual:.3f} > max {bound['max']}"
                )
                results["passed"] = False

        # 2. 检查失败 case 上限(同样在层级里,不在 quality_gate 顶层)
        max_failures = self.layer.get("failure_count_max", float("inf"))
        actual_failures = report.get("failure_count", 0)
        if actual_failures > max_failures:
            results["required_failures"].append(
                f"Too many failures: {actual_failures} > {max_failures}"
            )
            results["passed"] = False

        return results

    def print_pr_comment(self, results: dict):
        """生成 PR 评论 markdown"""
        lines = []
        if results["passed"]:
            lines.append(f"## ✅ Quality Gate Passed\n")
        else:
            lines.append(f"## ❌ Quality Gate Failed\n")
            for f in results["required_failures"]:
                lines.append(f"- {f}")
            lines.append("")

        lines.append("### Metrics")
        for metric, value in results["metrics"].items():
            lines.append(f"- {metric}: {value:.3f}")

        return "\n".join(lines)


if __name__ == "__main__":
    # 第二个参数选层:pr_subset / full_eval / daily_regression
    layer = sys.argv[2] if len(sys.argv) > 2 else "pr_subset"
    gate = QualityGate("ci-thresholds.yaml", layer=layer)
    results = gate.check(sys.argv[1])
    print(gate.print_pr_comment(results))
    sys.exit(0 if results["passed"] else 1)

约 90 行代码完成 Quality Gate 的核心检查逻辑:

  • 加载 yaml 阈值配置
  • 必需指标检查(min / max 阈值)
  • 失败 case 上限检查
  • PR 评论 markdown 生成
  • CI 退出码

工业实务:把这份 Python 接进 GitHub Actions / GitLab CI,配合 §18.10.2 的 ci-thresholds.yaml 即可使用。

18.10.2 CI Quality Gate 的"具体阈值"参考

把整章方法学落到具体阈值——一份给中等规模团队(30 人 / 日活 10 万)参考的阈值(示意值,需按自家实测校准):

# ci-thresholds.yaml
quality_gate:
  pr_subset:           # 第 1 层:PR 子集(5 分钟)
    sample_size: 50
    timeout_min: 5
    thresholds:
      faithfulness: { min: 0.85 }
      context_recall: { min: 0.90 }
      latency_p95_ms: { max: 3000 }
      cost_per_query_cny: { max: 0.05 }
    failure_count_max: 5
    auto_block_pr: true

  full_eval:           # 第 2 层:合并全集(30 分钟)
    sample_size: 500
    timeout_min: 30
    thresholds:
      faithfulness: { min: 0.85 }
      context_recall: { min: 0.90 }
      answer_relevancy: { min: 0.80 }
      jailbreak_pass_rate: { min: 0.95 }
      hallucination_rate: { max: 0.05 }
    failure_count_max: 25  # 5%
    auto_revert_on_fail: true

  daily_regression:    # 第 3 层:每日回归(1 小时)
    sample_size: 200   # 固定回归集
    timeout_min: 60
    drift_threshold:
      day_over_day: 0.05  # 日环比 5pp
      week_over_week: 0.03  # 周环比 3pp
    alert_severity: critical
    on_alert:
      - notify: pagerduty
      - notify: slack-channel

这份配置覆盖典型 LLM 应用的主要阈值。具体数字按业务调整,但结构性的"3 层 × 多指标 × 失败上限 × 告警机制"是通用模板。

工业实务:把这份 yaml 作为团队 CI 配置的初版,第一周根据自家实测数据调整阈值,再逐步稳定下来。

18.10.3 CI 评测的"快速失败链路"——3 秒看出 PR 该不该 merge

CI 评测的另一个常被忽略的维度:反馈速度。一个 30 分钟跑完的评测和一个 3 分钟出告警的评测,对开发体验天差地别——前者让工程师 context-switch、容易忘记 PR;后者让工程师在写 PR 时就能即时迭代。本节给读者一份"3 秒到 30 分钟"分级反馈链路工程方案,把 CI 评测从"等 30 分钟"压缩到"边写边见"。

graph LR
    A[工程师 commit/push] --> B[T+0 触发 CI]
    B --> C[T+3s: lint + 静态规则]
    C -->|fail| Z[立即 block]
    C -->|pass| D[T+30s: smoke evals 5 case]
    D -->|fail| Z
    D -->|pass| E[T+3min: 50 case sample]
    E -->|fail| Y[warn 但允许 merge]
    E -->|pass| F[T+30min: full 1000 case]
    F -->|regress > 5%| X[post-merge revert PR]
    F -->|pass| G[正常 merge]

4 级反馈链路 × 时间预算 × 评测覆盖:

级别 反馈时间 评测内容 失败后果 工程师感受
L1 lint 3 秒 yaml schema / prompt 模板语法 立即 block 写 PR 时即时反馈
L2 smoke 30 秒 5 个最关键 case 立即 block 一次喝水时间
L3 sample 3 分钟 50 题随机抽样 warn 但允许 喝杯咖啡
L4 full 30 分钟 1000 题完整 + 回归对比 post-merge revert merge 后异步保护

配套实现:4 级 CI 评测调度器:

import time
from dataclasses import dataclass, field
from typing import Literal, Callable

CILevel = Literal["L1_lint", "L2_smoke", "L3_sample", "L4_full"]
LevelOutcome = Literal["pass", "warn", "fail"]

@dataclass
class CIEvalLevel:
    name: CILevel
    timeout_seconds: float
    blocking: bool
    test_count: int
    runner: Callable[[], dict]

@dataclass
class TieredCIScheduler:
    levels: list[CIEvalLevel]

    def run_pipeline(self, fail_fast: bool = True) -> dict:
        results = {}
        merged_status: LevelOutcome = "pass"
        for level in self.levels:
            t0 = time.time()
            try:
                outcome = level.runner()
                elapsed = time.time() - t0
            except Exception as e:
                outcome = {"status": "fail", "error": str(e)}
                elapsed = time.time() - t0
            results[level.name] = {
                **outcome, "elapsed_s": round(elapsed, 2),
                "blocking": level.blocking,
            }
            if outcome.get("status") == "fail" and level.blocking and fail_fast:
                merged_status = "fail"
                results["pipeline_decision"] = {
                    "merged_status": "fail",
                    "stopped_at": level.name,
                    "reason": f"L{level.name[1]} 阻断阶段失败",
                }
                return results
            if outcome.get("status") in ("fail", "warn"):
                merged_status = "warn" if merged_status == "pass" else merged_status
        results["pipeline_decision"] = {
            "merged_status": merged_status,
            "completed": True,
        }
        return results

    @classmethod
    def standard_4_levels(cls,
                         lint_runner: Callable, smoke_runner: Callable,
                         sample_runner: Callable, full_runner: Callable
                         ) -> "TieredCIScheduler":
        return cls(levels=[
            CIEvalLevel("L1_lint", 5.0, True, 0, lint_runner),
            CIEvalLevel("L2_smoke", 60.0, True, 5, smoke_runner),
            CIEvalLevel("L3_sample", 300.0, False, 50, sample_runner),
            CIEvalLevel("L4_full", 1800.0, False, 1000, full_runner),
        ])

    def estimate_developer_time_savings(
        self, pr_per_day: int = 20,
        cost_of_context_switch_minutes: float = 8.0
    ) -> dict:
        """估算 4 级链路 vs 单一 30 分钟方案的开发体验提升"""
        old_total_min = pr_per_day * 30
        # 假设 80% PR 在 L1/L2 就拿到结论(示意比例),不必干等后续级别
        new_total_min = pr_per_day * (
            0.5 * 0.05    # 50% PR 在 L1 拿到结论, 约 3s
            + 0.3 * 0.5    # 30% PR 在 L2 拿到结论, 约 30s
            + 0.15 * 3     # 15% PR 在 L3 拿到结论, 约 3min
            + 0.05 * 30    # 5% PR 要等到 L4
        )
        ctx_switch_savings = (pr_per_day * 0.8) * cost_of_context_switch_minutes
        return {
            "old_pipeline_minutes": old_total_min,
            "new_pipeline_minutes": new_total_min,
            "context_switch_savings_minutes": ctx_switch_savings,
            "total_team_savings_per_day_minutes": (old_total_min - new_total_min)
                                                   + ctx_switch_savings,
        }

举例(示意数据,需用你自己的数据复算):某 20-PR/天的团队接入 4 级调度:

  • 旧方案:每 PR 等 30 分钟 → 600 工程师分钟/天
  • 新方案:50% PR 在 L1 拿到结论 + 30% 在 L2 + 15% 在 L3 + 5% 等到 L4 → 42.5 分钟/天,另省 16 × 8 = 128 上下文切换分钟
  • 团队每日节省 ≈ 600 − 42.5 + 128 ≈ 686 分钟 ≈ 11.4 工程师小时
  • 同时质量门禁不降级(L4 仍跑全量回归)

配套行业研究背景:

  • "Fast feedback loops" 来自 Kent Beck XP 1999
  • "Tiered CI" 来自 Google "Software Engineering at Google" 第 23 章
  • "Context switch cost" 可参考 Mark、Gudith、Klocke 的 CHI 2008 论文 "The Cost of Interrupted Work: More Speed and Stress"

读者把 TieredCIScheduler 接入 GitHub Actions / GitLab CI 配置,就能把"30 分钟全量"拆成"3 秒 → 30 秒 → 3 分钟 → 30 分钟"4 级链路。

18.10.4 "逐 PR 增量评测"的成本优化策略

全量跑评测成本太高——一份 1000 题的黄金集若每个 PR 跑一次,一年下来是一笔可观的 API 开销。但全量并非每次都必要。下面是一份基于"PR diff → 受影响子集"的增量评测策略,能省下的比例取决于改动分布:

import subprocess
import json
from pathlib import Path
from typing import Iterable
from dataclasses import dataclass

@dataclass
class IncrementalEvalPlan:
    full_eval_count: int
    selected_count: int
    saved_ratio: float
    selection_reason: dict[str, list[str]]

class PRImpactSelector:
    """根据 PR diff 决定本次 PR 必须跑评测集的哪些子集"""

    SUBSET_TRIGGERS = {
        "rag_retrieval": ["src/rag/", "config/embeddings/"],
        "rag_generation": ["src/llm/", "prompts/"],
        "agent_tools": ["src/agents/", "src/tools/"],
        "safety": ["src/safety/", "prompts/system/"],
        "multi_turn": ["src/conversation/", "src/memory/"],
        "schema": ["src/schemas/", "src/api/"],
    }

    def __init__(self, base: str = "main"):
        self.base = base

    def _changed_files(self) -> list[str]:
        result = subprocess.run(
            ["git", "diff", "--name-only", f"{self.base}...HEAD"],
            capture_output=True, text=True, check=True,
        )
        return [f for f in result.stdout.splitlines() if f.strip()]

    def _matches(self, path: str, prefixes: list[str]) -> bool:
        return any(path.startswith(p) for p in prefixes)

    def select(self, full_eval_set: dict[str, list]) -> IncrementalEvalPlan:
        files = self._changed_files()
        triggered: dict[str, list[str]] = {}

        for subset, prefixes in self.SUBSET_TRIGGERS.items():
            matches = [f for f in files if self._matches(f, prefixes)]
            if matches:
                triggered[subset] = matches

        if not triggered:
            triggered = {"smoke": ["__no_relevant_change__"]}

        if any("config/" in f or "src/core/" in f for f in files):
            triggered = {k: list(v) for k, v in
                        {"all": ["__core_change_full_run__"]}.items()}

        full_count = sum(len(v) for v in full_eval_set.values())
        selected_count = (full_count if "all" in triggered
                          else sum(len(full_eval_set.get(k, []))
                                  for k in triggered))

        return IncrementalEvalPlan(
            full_eval_count=full_count,
            selected_count=selected_count,
            saved_ratio=1 - selected_count / max(full_count, 1),
            selection_reason=triggered,
        )

    def emit_workflow_matrix(self, plan: IncrementalEvalPlan) -> str:
        if "all" in plan.selection_reason:
            subsets = list(self.SUBSET_TRIGGERS.keys())
        else:
            subsets = list(plan.selection_reason.keys())
        return json.dumps({"subset": subsets})
flowchart LR
  PR[PR 提交] --> D[git diff base...HEAD]
  D --> M{文件路径匹配?}
  M -->|src/rag/*| R[只跑 rag_retrieval]
  M -->|src/agents/*| A[只跑 agent_tools]
  M -->|prompts/system/*| S[只跑 safety + 全量 smoke]
  M -->|src/core/*| F[全量 fullrun]
  M -->|无相关| SM[只跑 smoke 50 题]
  R --> CI[CI matrix]
  A --> CI
  S --> CI
  F --> CI
  SM --> CI
  style F fill:#ffebee
  style SM fill:#e8f5e9

工程实务的 4 个增量策略(示意数据,需用你自己的数据复算):

触发模式 评测范围 节省比例
仅 docs / README 改动 smoke 50 题 节省 95%
仅 prompt 改动 rag + safety 节省 60%
仅 tool / agent 改动 agent + multi-turn 节省 70%
core / config 改动 全量 节省 0%

这套策略的工程关键点:

  • 白名单 vs 黑名单:用白名单(路径 → 子集)而非黑名单——防止"忘记加规则导致漏跑"
  • 保底 smoke 集:任何 PR 至少跑 50 题 smoke,防止"看似无关改动"埋雷
  • core 改动强制全量:src/core/ 或 config/ 改动必跑全量——不省那点钱

工业实务:增量策略的收益来自"绝大多数 PR 只碰到评测集的一小块"——评测账单与 CI 墙钟时间都会明显下降,具体幅度取决于你的改动分布,需要用自己半年的 CI 记录复算。

18.10.5 CI 评测的"复合 token 预算"——多模型 / 多评测同时跑时如何不互相饿死

CI pipeline 里同时有 lm-eval / ragas / promptfoo / 自定义 judge 在跑——4 个评测各自调 OpenAI / Anthropic API,没有协调时容易触发 vendor rate limit、彼此饿死、CI 整体超时。本节给读者一份「复合 token 预算 + 优先级调度」工程方案,让 4 类评测在同一 CI 中和平共处。

graph LR
    A[CI 启动] --> B[复合 token 预算分配]
    B --> C[lm-eval 25%]
    B --> D[ragas 30%]
    B --> E[promptfoo 25%]
    B --> F[自定义 judge 20%]
    C & D & E & F --> G[各自跑]
    G --> H{触发限流?}
    H -->|否| I[正常完成]
    H -->|是| J[动态调度]
    J --> K[暂停低优]
    J --> L[重试高优]
    K --> I
    L --> I

4 类评测优先级 × 默认配额 × 限流处置:

评测类型 优先级 默认 token 配额 限流时处置
lm-eval(基础能力) P3 25% 暂停等待
ragas(核心 RAG) P1 30% 优先重试
promptfoo(CI 门禁) P0 25% 必须完成
自定义 judge(业务) P2 20% 限流后排队

配套实现:复合 token 预算调度器:

import time
from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal

EvalKind = Literal["lm_eval", "ragas", "promptfoo", "custom_judge"]
Priority = Literal["P0", "P1", "P2", "P3"]

PRIORITY_TABLE: dict[EvalKind, Priority] = {
    "lm_eval": "P3", "ragas": "P1",
    "promptfoo": "P0", "custom_judge": "P2",
}

DEFAULT_QUOTA_PCT: dict[EvalKind, float] = {
    "lm_eval": 0.25, "ragas": 0.30,
    "promptfoo": 0.25, "custom_judge": 0.20,
}

@dataclass
class TokenBudget:
    total_tokens: int
    consumed_per_kind: dict[EvalKind, int] = field(default_factory=dict)

    def quota_for(self, kind: EvalKind) -> int:
        return int(self.total_tokens * DEFAULT_QUOTA_PCT[kind])

    def can_consume(self, kind: EvalKind, request_tokens: int) -> bool:
        consumed = self.consumed_per_kind.get(kind, 0)
        return consumed + request_tokens <= self.quota_for(kind)

    def consume(self, kind: EvalKind, tokens: int):
        self.consumed_per_kind[kind] = self.consumed_per_kind.get(kind, 0) + tokens

    def remaining(self, kind: EvalKind) -> int:
        return self.quota_for(kind) - self.consumed_per_kind.get(kind, 0)

@dataclass
class RateLimitState:
    is_throttled: bool = False
    throttled_until: datetime | None = None
    consecutive_failures: int = 0

@dataclass
class CompositeBudgetScheduler:
    budget: TokenBudget
    rate_limits: dict[str, RateLimitState] = field(default_factory=dict)

    def execute_eval(self, kind: EvalKind, vendor: str,
                    estimated_tokens: int) -> dict:
        # 1. 检查预算
        if not self.budget.can_consume(kind, estimated_tokens):
            if PRIORITY_TABLE[kind] == "P0":
                return {"action": "borrow_from_lower",
                        "reason": "P0 必须完成,从 P3 借配额"}
            return {"action": "skip_or_queue",
                    "reason": f"{kind} 配额不足,等下次 CI"}
        # 2. 检查限流状态
        rl = self.rate_limits.get(vendor, RateLimitState())
        if rl.is_throttled:
            if PRIORITY_TABLE[kind] in ("P0", "P1"):
                return {"action": "wait_then_retry",
                        "reason": f"{vendor} 限流中,但优先级高,等候"}
            return {"action": "defer", "reason": "低优先级,下次 CI 再跑"}
        # 3. 正常执行
        self.budget.consume(kind, estimated_tokens)
        return {"action": "execute", "tokens_used": estimated_tokens,
                "remaining_quota": self.budget.remaining(kind)}

    def handle_rate_limit_response(self, vendor: str):
        rl = self.rate_limits.setdefault(vendor, RateLimitState())
        rl.is_throttled = True
        rl.consecutive_failures += 1
        backoff_seconds = 2 ** min(rl.consecutive_failures, 6)
        rl.throttled_until = datetime.now()  # 简化
        return {"backoff_seconds": backoff_seconds,
                "consecutive_failures": rl.consecutive_failures}

    def reset_rate_limit(self, vendor: str):
        if vendor in self.rate_limits:
            self.rate_limits[vendor].is_throttled = False
            self.rate_limits[vendor].consecutive_failures = 0

    def report(self) -> dict:
        return {
            "total_budget": self.budget.total_tokens,
            "consumption_breakdown": {
                k.value if hasattr(k, "value") else k: v
                for k, v in self.budget.consumed_per_kind.items()
            },
            "utilization_pct": sum(self.budget.consumed_per_kind.values())
                              / max(self.budget.total_tokens, 1) * 100,
            "rate_limited_vendors": [v for v, rl in self.rate_limits.items()
                                     if rl.is_throttled],
        }

举例(示意数据,需用你自己的数据复算):某团队 CI 配置 100 万 token / run:

  • ragas 配额 30 万、promptfoo 25 万、custom_judge 20 万、lm-eval 25 万
  • 跑到中段 OpenAI 限流 → promptfoo、ragas 等候重试 + custom_judge 延后
  • 一次跑下来:promptfoo 100% 完成(CI 门禁不漏)/ ragas 90% 完成 / custom_judge 70%(部分排队下次跑)/ lm-eval 跳过本次(P3 让位)

避免「4 个评测同时调 OpenAI → 全部触发 429 → 全 CI 红 → 工程师重跑 5 次」的常见 CI 浪费。

可参照的同类设计:

  • 按命名空间分配资源配额:Kubernetes ResourceQuota
  • 按优先级让位:Google Borg 的优先级调度
  • 客户端限流:LangChain 的 rate limiter 模块

18.10.6 CI Evals 的成本边界:什么时候推不下去

CI Evals 的最大风险是成本失控。下表是示意值,只用来说明成本随规模放大的量级,需按自家调用量和当期价目表重算:

团队规模 PR/天 月度评测成本(示意)
5 人 / 早期 2-5 ¥3000-8000
20 人 / 成长期 10-30 ¥1.5-5 万
50 人 / 成熟期 30-100 ¥5-15 万
200 人 / 大规模 100-300 ¥15-50 万

这笔开销放在研发预算里通常只占小头,但有几种情况会让成本异常飙高:

  1. 黄金集失控:本应 200 题的集膨胀到 5000 题,每次跑 30 分钟变 5 小时
  2. judge 模型升级:从小模型换成推理模型,单价可能差一两个数量级,推理 token 还要另算
  3. 每个 PR 跑全集:忘了分层、PR 也跑 500 题
  4. 失败重试无上限:失败自动重跑、循环跑爆

工程修法都是简单的——加 budget 上限、加超时、加分层、加重试上限。但这些细节常被忽略,导致团队某天看到月度账单大吃一惊。CI Evals 必须有"成本告警"——超过预算 1.5x → 自动暂停 + 通知。

18.11 开发者体验:PR 里的评测

18.11.1 PR 评测的 UX 设计:让开发者愿意用

CI Quality Gate 的最大成败因素不是技术——是开发者愿不愿意用。如果 UX 差到工程师每次都想绕过,再完善的评测也是摆设。

良好 UX 的具体形态:

1. PR 评论格式

## ✗ Eval results: 47/50 cases passed (94%)

### Failed cases:
- [faithfulness=0.62] "我对花生过敏, 推荐零食" → ⚠️ 答案含巴旦木
- [recall=0.71] "退货周期是多久" → ⚠️ 检索漏掉政策第 3 条
- [latency=4200ms] "...复杂多步问题..." → ⚠️ 超过 3s 阈值

### Compared to main:
- faithfulness: 0.91 (-0.03)
- recall: 0.88 (+0.02)
- latency p95: 2800ms (+200ms)

[查看完整 trace](https://smith.langchain.com/...)

不是冷冰冰的"94% 通过率 → fail",而是直接定位到具体哪条样例 / 哪个指标 / 与 main 对比 / 一键链到 trace。

2. 失败 case 可点击

每条失败 case 链到 langsmith / langfuse 的 trace 详情——开发者能看到完整 (input, retrieved, output, judge_reasoning),不需要本地复现。

3. 改进建议

💡 Suggestion: faithfulness 下降可能与 prompt 改动有关。
   尝试在 system prompt 加 "Use ONLY the provided context."

基于历史改动模式给出建议(用 LLM 自动生成),不是教条命令。

4. 局部重跑能力

> /retry-eval failed-only

slash command 让开发者只重跑失败 case,不需要重跑全集。3 分钟看到改动结果。

这些 UX 细节看起来琐碎,实际决定了 Quality Gate 的"采用率"——评测体系不只是后端的指标计算,前端的开发者体验同等重要。

18.11.2 "Evals as PR Comment Bot"——把分数 inline 贴到 GitHub

让评测分数"显眼地"出现在工程师每日工作流中是 quality gate 落地的关键——光有 CI 不够,每个 PR 必须看到具体分数 + 失败 case 链接。下面是工业团队推荐的 PR comment bot 实现:

import json
import os
from dataclasses import dataclass, field
from typing import Iterable
from pathlib import Path

@dataclass
class EvalResultSummary:
    pr_number: int
    sha: str
    overall_score: float
    baseline_score: float
    delta_pp: float
    pass_count: int
    fail_count: int
    failed_cases: list[dict]
    cost_usd: float
    duration_s: float

class GitHubPREvalCommenter:
    """把评测结果作为 PR comment 自动发布"""

    def __init__(self, gh_token: str, repo: str):
        self.token = gh_token
        self.repo = repo

    def render_comment(self, summary: EvalResultSummary) -> str:
        emoji = "✅" if summary.delta_pp >= -2 else "❌"
        delta_label = (f"+{summary.delta_pp:.1f}pp" if summary.delta_pp >= 0
                       else f"{summary.delta_pp:.1f}pp")

        comment = (
            f"## {emoji} Eval Report (sha: `{summary.sha[:8]}`)\n\n"
            f"| Metric | Baseline | This PR | Delta |\n"
            f"|--------|---------|---------|-------|\n"
            f"| Overall | {summary.baseline_score:.3f} | "
            f"{summary.overall_score:.3f} | {delta_label} |\n"
            f"| Pass / Fail | — | "
            f"{summary.pass_count} / {summary.fail_count} | — |\n\n"
            f"**Cost**: ${summary.cost_usd:.2f} · **Duration**: "
            f"{summary.duration_s:.0f}s\n\n"
        )

        if summary.failed_cases:
            comment += "<details><summary>Top failed cases</summary>\n\n"
            for c in summary.failed_cases[:5]:
                comment += (
                    f"- `{c['case_id']}` (score={c['score']:.2f}): "
                    f"{c['query'][:80]}...\n"
                    f"  - [trace]({c['trace_url']})\n"
                )
            comment += "\n</details>\n"

        if summary.delta_pp < -3:
            comment += ("\n> ⚠️ **Delta < -3pp**: 此 PR 触发 quality gate "
                        "红线,需在 merge 前修复或主管批准。\n")
        return comment

    def post_or_update(self, summary: EvalResultSummary):
        """新评论或更新已有评论(避免每 push 都新一条)"""
        import requests
        existing = self._find_existing_comment(summary.pr_number)
        url = (f"https://api.github.com/repos/{self.repo}/issues/"
               f"{summary.pr_number}/comments")
        if existing:
            url = (f"https://api.github.com/repos/{self.repo}/issues/"
                   f"comments/{existing}")
            method = "PATCH"
        else:
            method = "POST"

        body = {"body": self.render_comment(summary)}
        resp = requests.request(method, url, json=body,
                                 headers={"Authorization": f"token {self.token}"})
        return resp.status_code in (200, 201)

    def _find_existing_comment(self, pr: int) -> str | None:
        import requests
        url = (f"https://api.github.com/repos/{self.repo}/issues/{pr}"
               f"/comments")
        resp = requests.get(url, headers={"Authorization":
                                            f"token {self.token}"})
        for c in resp.json():
            if c["user"]["login"] == "github-actions[bot]" and \
                    "Eval Report" in c["body"]:
                return c["id"]
        return None
flowchart LR
  PR[PR push] --> CI[CI workflow]
  CI --> EV[ragas / promptfoo eval]
  EV --> SUM[EvalResultSummary]
  SUM --> COM[Commenter]
  COM --> EX{已有 comment?}
  EX -->|是| UP[PATCH 更新现有]
  EX -->|否| NEW[POST 新建]
  UP --> PRC[PR 页面 inline 显示]
  NEW --> PRC
  PRC --> DEV[工程师立即看到]
  style PRC fill:#e8f5e9
  style DEV fill:#e3f2fd

工程实务的 5 条 PR comment 设计原则:

  1. 新 push 必更新而非新建:每条 PR 只保留 1 条最新 eval comment,否则评论列表爆炸
  2. failed cases 折叠展示:用 <details> 让 dashboard 紧凑,需要时再展开
  3. 每个失败 case 必带 trace 链接:5 秒内能跳到 LangFuse / LangSmith 看具体 trace
  4. delta < -3pp 必显式 warning:用引述块视觉强调
  5. emoji 区分状态:✅ / ❌ 一眼可辨

具体例子(示意数据,需用你自己的数据复算):某团队 PR comment 实例:

## ✅ Eval Report (sha: `a1b2c3d4`)

| Metric | Baseline | This PR | Delta |
|--------|---------|---------|-------|
| Overall | 0.842 | 0.851 | +0.9pp |
| Pass / Fail | — | 187 / 13 | — |

**Cost**: $4.23 · **Duration**: 312s

<details><summary>Top failed cases</summary>

- `gold-022` (score=0.45): 用户问"我能用其他公司优惠券吗"...
  - [trace](https://langfuse.../trace/abc)
- `gold-067` (score=0.51): 复杂多轮退款 case...
  - [trace](https://langfuse.../trace/def)

</details>

部署后效果:

  • 100% PR 必带 eval comment——工程师不需要打开外部 dashboard
  • 失败 case 5 秒可访问 trace
  • 评测从"看不见"变成"PR review 必看"——质量文化提升

把 GitHubPREvalCommenter 接入 §18.11.4 Slack alert 工作流——形成"PR review 看 comment + Slack 看高优告警"的双视角。

18.11.3 CI 评测的"灰盒可观测性"——让"为什么 CI 失败"30 秒内可见

CI 失败时工程师最关心"为什么失败 + 怎么修"——但很多团队的 CI 输出只有 "Failed: 12 tests" 这种黑盒信息。下面给出灰盒可观测性的工程化方案:

import json
from dataclasses import dataclass, asdict, field
from typing import Iterable

@dataclass
class CIFailureContext:
    failed_case_id: str
    expected: str
    actual: str
    diff_summary: str
    likely_cause_category: str   # "prompt_change" / "data_drift" / "model_drift" / "flaky"
    suggested_action: str
    related_pr_files: list[str]
    drilldown_urls: dict[str, str]   # trace / log / dashboard

class CIFailureExplainer:
    """把 CI 失败转成 30 秒可消费的 explainer"""

    CAUSE_INDICATORS = {
        "prompt_change": ["prompts/", "system_prompt", ".prompt"],
        "data_drift": ["evals/golden/", "samples.jsonl"],
        "model_drift": ["model_id", "endpoint changed"],
        "flaky": ["already known flaky"],
    }

    def __init__(self, pr_changed_files: list[str]):
        self.pr_files = pr_changed_files

    def _classify_cause(self, failure: dict) -> str:
        for cause, indicators in self.CAUSE_INDICATORS.items():
            if any(ind in " ".join(self.pr_files) for ind in indicators):
                return cause
        return "unknown"

    def explain(self, failure: dict) -> CIFailureContext:
        cause = self._classify_cause(failure)
        suggested = {
            "prompt_change": "检查 prompt 改动是否引入语义偏移;跑 §6.8.8 judge drift",
            "data_drift": "检查评测集是否漂移;跑 §3.11.4 KL audit",
            "model_drift": "检查模型版本;接 §6.8.8 watchdog",
            "flaky": "标记 quarantine(§18.13.1)",
            "unknown": "看 trace / 找 evals owner",
        }[cause]

        return CIFailureContext(
            failed_case_id=failure["case_id"],
            expected=failure["expected"][:100],
            actual=failure["actual"][:100],
            diff_summary=self._diff(failure["expected"], failure["actual"]),
            likely_cause_category=cause,
            suggested_action=suggested,
            related_pr_files=self.pr_files[:5],
            drilldown_urls={
                "trace": f"https://langfuse.../{failure['trace_id']}",
                "case_yaml": f"https://github/.../yaml#L{failure.get('line', 0)}",
                "dashboard": "https://grafana/...",
            },
        )

    def render_pr_comment_table(self,
                                  failures: list[CIFailureContext]) -> str:
        """渲染 GitHub PR comment markdown table"""
        from collections import Counter
        causes = Counter(f.likely_cause_category for f in failures)
        lines = [
            f"## ❌ {len(failures)} cases failed",
            "",
            "**Cause breakdown**: " + ", ".join(f"{c}={n}"
                                                  for c, n in causes.items()),
            "",
            "| case | cause | suggested | trace |",
            "|---|---|---|---|",
        ]
        for f in failures[:10]:
            lines.append(f"| `{f.failed_case_id[:12]}` | {f.likely_cause_category} | "
                          f"{f.suggested_action[:50]}... | "
                          f"[trace]({f.drilldown_urls['trace']}) |")
        return "\n".join(lines)

    def _diff(self, expected: str, actual: str) -> str:
        return f"len_e={len(expected)}, len_a={len(actual)}, diff={len(actual) - len(expected)}"

注意上面的 _classify_cause 只看 PR 改动了哪些文件,同一个 PR 里的所有失败会被归到同一类;要得到下文示例那样的逐条原因分布,还得把失败 case 自身的信号(已知 flaky 名单、case 所属数据集版本、调用的模型版本等)纳入判断。

flowchart LR
  F[CI 失败] --> EXP[Failure Explainer]
  EXP --> CL[根据 PR 文件分类原因]
  CL --> C1[prompt_change]
  CL --> C2[data_drift]
  CL --> C3[model_drift]
  CL --> C4[flaky]
  C1 --> S1[改 prompt 建议]
  C2 --> S2[补数据建议]
  C3 --> S3[换 model 建议]
  C4 --> S4[quarantine]
  S1 --> COMM[PR comment markdown]
  S2 --> COMM
  S3 --> COMM
  S4 --> COMM
  COMM --> DEV[工程师 30s get]
  style COMM fill:#e8f5e9

工程实务的 4 类 CI failure 灰盒价值(时间为示意值):

维度 黑盒 CI 灰盒 CI
工程师 debug 时间 30 分钟 5 分钟
trace drilldown 手动找 1 click
cause classification 全员猜 自动分类
Slack 告警内容 "12 tests failed" "12 fails / prompt_change=8 + drift=4"

具体例子(示意数据,需用你自己的数据复算):某 PR 触发 12 个 case failure,灰盒输出:

## ❌ 12 cases failed

**Cause breakdown**: prompt_change=8, data_drift=3, flaky=1

| case | cause | suggested | trace |
|---|---|---|---|
| `gold-022` | prompt_change | 检查 prompt 改动是否引入语义... | [trace](...) |
| `gold-067` | data_drift | 检查评测集是否漂移;跑 §3.11.4 分布对齐度量... | [trace](...) |
...

工程师秒懂:"8 个失败因为我改了 prompt → 看 prompt diff + run §6.8.8"。

3 类常见黑盒坑:

坑 现象 修法
无 cause 分类 工程师全 case 一个个查 自动分类
无 drilldown 找 trace 翻 dashboard 必带 URL
失败 message 太短 "expected != actual" 必含 diff_summary

研究背景:

  • "Observability Engineering" (O'Reilly 2022) 系统讲了可观测性的工程实践
  • Honeycomb 的 "BubbleUp" 自动找出异常请求与正常请求的差异维度,是同一思路

读者可以把 CIFailureExplainer 接到 §18.11.2 PR comment bot 上。

18.11.4 完整的 Slack 告警 + 失败定位脚本

整合本章方法学,给一份"Quality Gate 失败 → Slack 告警 + 自动定位失败 case"完整脚本:

# slack_alerter.py
import json
import os
import requests
from pathlib import Path
from datetime import datetime

class QualityGateAlerter:
    """Quality Gate 失败时发送 Slack 告警"""

    def __init__(self, slack_webhook: str, langsmith_url: str):
        self.webhook = slack_webhook
        self.trace_url = langsmith_url

    def alert(self, eval_report: dict, severity: str = "warning"):
        """根据评测报告发 Slack 告警"""
        if eval_report["passed"]:
            return  # 通过不发告警

        emoji = {"critical": "🚨", "warning": "⚠️", "info": "ℹ️"}[severity]
        commit_sha = eval_report.get("commit", "unknown")[:8]
        pr_url = eval_report.get("pr_url", "")

        # 顶部 summary
        text_lines = [
            f"{emoji} *Quality Gate Failed*",
            f"Commit: `{commit_sha}` | PR: {pr_url}",
            f"Time: {datetime.now().isoformat()}",
            "",
        ]

        # 列出失败指标
        text_lines.append("*Failed Metrics:*")
        for failure in eval_report["required_failures"][:5]:
            text_lines.append(f"  • {failure}")
        text_lines.append("")

        # Top 3 失败 case 链接
        text_lines.append("*Top Failed Cases (click trace):*")
        for case in eval_report.get("worst_cases", [])[:3]:
            text_lines.append(
                f"  • [{case['metric']}={case['score']:.2f}]"
                f" {case['input'][:50]}... <{self.trace_url}/{case['trace_id']}|view trace>"
            )

        # 改进建议
        suggestions = self._generate_suggestions(eval_report)
        if suggestions:
            text_lines.append("")
            text_lines.append("*Suggestions:*")
            for s in suggestions:
                text_lines.append(f"  💡 {s}")

        self._send(text="\n".join(text_lines), severity=severity)

    def _generate_suggestions(self, report: dict) -> list[str]:
        """基于失败模式生成改进建议"""
        suggestions = []
        for failure in report["required_failures"]:
            if "faithfulness" in failure.lower():
                suggestions.append("Faithfulness 下降可能源于 prompt 改动. 尝试加'Use ONLY the context'指令")
            elif "context_recall" in failure.lower():
                suggestions.append("Recall 下降可能 retriever 配置问题. 检查 chunk_size / top_k")
            elif "latency" in failure.lower():
                suggestions.append("Latency 升高可能是 chain 拉长. 检查最近的 prompt 改动是否引入额外步骤")
            elif "cost" in failure.lower():
                suggestions.append("Cost 升高可能 prompt 变长 or 切到更贵模型. 检查 token 用量")
        return suggestions

    def _send(self, text: str, severity: str):
        color = {"critical": "#ff0000", "warning": "#ff9900", "info": "#0099ff"}[severity]
        payload = {
            "attachments": [{
                "color": color,
                "text": text,
                "mrkdwn_in": ["text"],
            }]
        }
        requests.post(self.webhook, json=payload, timeout=10)


# 使用:CI 失败时调用
if __name__ == "__main__":
    import sys
    report = json.loads(Path(sys.argv[1]).read_text())
    alerter = QualityGateAlerter(
        slack_webhook=os.environ["SLACK_WEBHOOK"],
        langsmith_url="https://smith.langchain.com/...",  # 换成自家项目的 trace 链接前缀
    )
    alerter.alert(report, severity="warning" if len(report["required_failures"]) < 3 else "critical")

约 90 行代码完成 Quality Gate 告警的完整自动化:

  • 失败指标 top 5 列表
  • top 3 失败 case 直接链到 LangSmith trace
  • 基于失败模式自动生成改进建议(4 类常见原因)
  • 严重度分级(warning / critical),不同颜色

工业实务:把这份脚本接进 CI workflow 的 if: failure() step,开发者在 Slack 里就能直接定位到失败指标和 case,不必翻 CI log。

18.12 与发布流程协同

18.12.1 "评测 + Deploy"的"红绿灯交通系统"——上线决策的最终形态

集结全章工具,最终形态是一套"红绿灯"——评测各环节的状态压缩成"该不该 deploy"的清晰答案。下面给出工程实现:

from dataclasses import dataclass
from enum import Enum
from typing import Iterable

class TrafficLight(Enum):
    GREEN = "green"
    YELLOW = "yellow"
    RED = "red"

@dataclass
class DeploymentReadiness:
    overall: TrafficLight
    pre_deploy_eval: TrafficLight
    safety_eval: TrafficLight
    canary_perf: TrafficLight
    user_feedback_trend: TrafficLight
    cost_budget: TrafficLight
    blocker_message: str | None

class DeploymentTrafficLight:
    """评测各信号 → 整合为 deploy 决策的红绿灯"""

    def evaluate(self, signals: dict) -> DeploymentReadiness:
        # Pre-deploy CI eval
        pre = self._classify(
            signals.get("ci_pass_rate", 0),
            green=0.95, yellow=0.85
        )

        # Safety
        safety = self._classify_inverse(
            signals.get("safety_violation_rate", 1),
            green=0.005, yellow=0.01
        )

        # Canary perf
        canary = self._classify(
            signals.get("canary_score", 0),
            green=0.85, yellow=0.75
        )

        # User feedback
        feedback = self._classify(
            signals.get("user_thumbs_up_pct_7d", 0),
            green=0.85, yellow=0.75
        )

        # Cost budget
        cost = self._classify_inverse(
            signals.get("cost_overage_pct", 0),
            green=0.05, yellow=0.20
        )

        # Aggregate
        all_lights = [pre, safety, canary, feedback, cost]
        if any(l == TrafficLight.RED for l in all_lights):
            overall = TrafficLight.RED
            blocker = self._find_blocker(signals, all_lights)
        elif sum(1 for l in all_lights if l == TrafficLight.YELLOW) >= 2:
            overall = TrafficLight.YELLOW
            blocker = "≥ 2 黄灯 - 谨慎 deploy + 加灰度"
        else:
            overall = TrafficLight.GREEN
            blocker = None

        return DeploymentReadiness(
            overall=overall,
            pre_deploy_eval=pre,
            safety_eval=safety,
            canary_perf=canary,
            user_feedback_trend=feedback,
            cost_budget=cost,
            blocker_message=blocker,
        )

    def _classify(self, value: float, green: float,
                   yellow: float) -> TrafficLight:
        if value >= green:
            return TrafficLight.GREEN
        if value >= yellow:
            return TrafficLight.YELLOW
        return TrafficLight.RED

    def _classify_inverse(self, value: float, green: float,
                            yellow: float) -> TrafficLight:
        if value <= green:
            return TrafficLight.GREEN
        if value <= yellow:
            return TrafficLight.YELLOW
        return TrafficLight.RED

    def _find_blocker(self, signals, lights) -> str:
        for label, light in [
            ("pre_deploy_eval", lights[0]),
            ("safety_eval", lights[1]),
            ("canary_perf", lights[2]),
            ("user_feedback_trend", lights[3]),
            ("cost_budget", lights[4]),
        ]:
            if light == TrafficLight.RED:
                return f"{label} 红灯——必须修复才能 deploy"
        return ""
flowchart LR
  S1[CI eval] --> AGG{Traffic Light Logic}
  S2[Safety] --> AGG
  S3[Canary perf] --> AGG
  S4[User feedback 7d] --> AGG
  S5[Cost budget] --> AGG
  AGG --> RED{任一 RED?}
  AGG --> YELLOW{≥ 2 YELLOW?}
  RED -->|是| STOP[🛑 STOP<br/>修复 blocker]
  YELLOW -->|是| CAUTION[🟡 CAUTION<br/>灰度 deploy]
  AGG -->|全 green| GO[✅ GO<br/>full deploy]
  STOP --> RCA[找 root cause]
  CAUTION --> SLOW[1% → 5% → ...]
  GO --> NORMAL[正常上线]
  style STOP fill:#ffebee
  style CAUTION fill:#fff3e0
  style GO fill:#e8f5e9

工程实务的 5 条 traffic light 设计原则:

  1. 单维度 RED → 整体 RED:不允许"6 个绿 1 个红"放行
  2. YELLOW 是审批触发:≥ 2 黄灯需要 owner 签字才能继续
  3. 每个信号必须可追溯:red light 必须能 click 进 trace / 评测细节
  4. 绿灯不等于"放心":总有评测覆盖不到的 case
  5. 每周 review traffic light 历史:长期黄灯说明阈值需调

具体例子(示意数据,需用你自己的数据复算):某 PR 想 deploy 的状态:

信号 值 阈值 灯
CI pass rate 0.96 ≥ 0.95 🟢
safety violation 0.003 ≤ 0.005 🟢
canary score 0.78 ≥ 0.85 🟡
user thumbs up 7d 0.82 ≥ 0.85 🟡
cost overage 0.08 ≤ 0.05 🟡

3 个 yellow → overall YELLOW → 触发 owner 审批,建议 5% 灰度而非 100%。

3 类 deploy 决策错误:

错误 现象 修法
红灯强行上 工程师"我赶时间" 必须 owner + 二级主管签字才能跳
全绿不灰度 直接 100% 上线 green 也至少 25% canary
灯失效不响应 6 个月没更新阈值 季度 review 阈值合理性

读者把 DeploymentTrafficLight 接到 GitHub Actions 的 deploy job 前置 step,任何 deploy 都先过红绿灯。

18.12.2 Quality Gate 与 Feature Flag 的协同——评测失败如何让上线"安全降级"

quality gate 阻止合并是事后防线,但 feature flag + canary 灰度让"已合并的代码"也能在评测信号变差时安全降级。下面是工程实务推荐的"双层防御"——评测失败后的实时回退路径:

import asyncio
from dataclasses import dataclass
from typing import Callable, Awaitable
from datetime import datetime, timedelta

@dataclass
class FeatureFlagState:
    flag_name: str
    enabled_pct: float
    targeting_rule: str
    eval_score_baseline: float
    last_check_at: str

class EvalDrivenFlagController:
    """让 feature flag 自动响应评测信号"""

    AUTO_ROLLBACK_THRESHOLD = 0.05    # 5pp 跌幅触发回滚
    PROMOTE_THRESHOLD_DELTA = 0.02     # 2pp 涨幅触发提升

    def __init__(self, flag_client, eval_metric_fn: Callable[[], Awaitable[float]]):
        self.flag_client = flag_client
        self.eval_metric = eval_metric_fn

    async def safe_promote(self, flag: str, target_pct: float):
        """灰度晋升:每阶段必须评测达标"""
        stages = [1, 5, 25, 50, target_pct]
        baseline = await self.eval_metric()
        for stage in stages:
            if stage > target_pct:
                break
            await self.flag_client.set_pct(flag, stage)
            await asyncio.sleep(3600)   # 1 小时观察
            current = await self.eval_metric()
            if current < baseline - self.AUTO_ROLLBACK_THRESHOLD:
                await self.flag_client.set_pct(flag, 0)
                return {"status": "rolled_back",
                        "stage": stage,
                        "drop_pp": baseline - current}
        return {"status": "promoted", "final_pct": target_pct}

    async def watchdog(self, flag: str, baseline: float):
        """持续监控:跌幅超阈直接降到 0"""
        while True:
            current = await self.eval_metric()
            if current < baseline - self.AUTO_ROLLBACK_THRESHOLD:
                await self.flag_client.set_pct(flag, 0)
                await self._alert(flag, baseline, current)
                return
            await asyncio.sleep(900)   # 15 分钟一次

    async def _alert(self, flag: str, baseline: float, current: float):
        # 接 Slack / PagerDuty
        pass
flowchart LR
  PR[PR 通过 Quality Gate] --> M[merge to main]
  M --> CAN[canary 1%]
  CAN -->|评测达标| GR1[5%]
  GR1 -->|评测达标| GR2[25%]
  GR2 -->|评测达标| GR3[50%]
  GR3 -->|评测达标| FU[100%]
  CAN -->|评测跌| BACK[回滚到 0%]
  GR1 -->|评测跌| BACK
  GR2 -->|评测跌| BACK
  GR3 -->|评测跌| BACK
  FU --> WD[watchdog 15min check]
  WD -->|跌| BACK
  style BACK fill:#ffebee
  style FU fill:#e8f5e9

工程实务的 5 条 flag + eval 协同准则:

阶段 流量比例 观察窗口 评测频率 回滚条件
canary 1% 1h 实时 主指标跌 ≥ 3pp
early 5% 4h 每 30min 主指标跌 ≥ 4pp
ramp 25% 12h 每 1h 主指标跌 ≥ 5pp
half 50% 24h 每 2h 主指标跌 ≥ 5pp
full 100% 持续 每 15min 主指标跌 ≥ 5pp

工程实务的 4 个使用要点:

  1. 总有一个 flag 控制最近改动:每次 PR merge 前先把 flag 默认 off,merge 后 controller 接管
  2. 回滚是"降到 0%"而不是删代码:保留 PR 在仓库,避免反复 revert / re-apply 折腾
  3. 每个 flag 必须有 baseline 锚:不锚 baseline 就无法判断"跌"
  4. flag 寿命 ≤ 30 天:评测稳定后必须清除 flag(避免代码债)

具体例子(示意数据,需用你自己的数据复算):客服 RAG 系统改 prompt:

  • 0:00 PR merge,flag new_prompt_v3 = 0%
  • 0:01 controller safe_promote(target=100%)
  • 0:01-1:00 1% 流量,主指标 Faithfulness=0.83(baseline 0.84)
  • 1:00 → 5%,下一小时 Faithfulness=0.78(跌 6pp,超过 5pp 回滚线)
  • 2:00 观察窗结束时 flag 自动 → 0%,oncall 接到 Slack 告警

这种"评测驱动的自动灰度"是 §18.12.3 自动 rollback 的更细粒度版本——后者是事后回滚 main、本节是阻止灰度爬升。两层结合 = LLM 应用上线的 safety net。注意上表的分阶段回滚线(3pp/4pp/5pp)比代码里统一的 AUTO_ROLLBACK_THRESHOLD = 0.05 更细,落地时把阈值做成按阶段取值。

18.12.3 Quality Gate 的"自动 Rollback"机制

CI 拦住了不合格的 PR,但已经合入 main 之后才发现退化怎么办?工程实务里的解法是 auto-rollback——当 main 分支的滚动评测发现连续 3 次失败,自动 revert 最近一次提交并发 PR。下面是一份完整 GitHub Action:

name: rolling-eval-with-rollback
on:
  push:
    branches: [main]
  schedule:
    - cron: '0 */2 * * *'    # 每 2 小时跑一次

jobs:
  rolling-eval:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    permissions:
      contents: write
      pull-requests: write
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 50

      - name: Run smoke eval
        id: eval
        run: |
          python -m evals.smoke --output result.json
          echo "score=$(jq .score result.json)" >> $GITHUB_OUTPUT
          echo "p99_latency=$(jq .p99_latency result.json)" >> $GITHUB_OUTPUT

      - name: Persist eval history
        run: |
          mkdir -p .eval-history
          cp result.json .eval-history/$(date +%s).json
          ls -t .eval-history/*.json | tail -n +20 | xargs rm -f || true
          git add .eval-history
          git -c user.email=eval@bot -c user.name=eval-bot commit -m "eval log" || true
          git push origin main || true

      - name: Detect 3-consecutive-failures
        id: detect
        run: |
          python scripts/detect_consec_failures.py \
            --threshold 0.85 --window 3 > decision.json
          echo "should_rollback=$(jq .should_rollback decision.json)" >> $GITHUB_OUTPUT
          echo "last_good_sha=$(jq -r .last_good_sha decision.json)" >> $GITHUB_OUTPUT

      - name: Auto-create rollback PR
        if: steps.detect.outputs.should_rollback == 'true'
        env:
          GH_TOKEN: ${{ github.token }}   # gh CLI 在 Actions 里要显式给 token
        run: |
          BAD_SHA=$(git rev-parse HEAD)
          GOOD_SHA="${{ steps.detect.outputs.last_good_sha }}"
          BR="auto-rollback-$(date +%Y%m%d-%H%M)"
          git checkout -b "$BR"
          git -c user.email=eval@bot -c user.name=eval-bot \
            revert --no-edit "$GOOD_SHA"..HEAD
          # 下面的 label 与 reviewer 须事先在仓库里存在,否则 gh pr create 会报错
          git push origin "$BR"
          gh pr create \
            --title "[AUTO] Rollback to $GOOD_SHA: 3 consecutive eval failures" \
            --body "Eval score dropped below threshold for 3 windows. \
                    Reverting commits $GOOD_SHA..$BAD_SHA. \
                    Please investigate and re-roll forward after fix." \
            --label "auto-rollback,p0" \
            --reviewer eval-oncall

      - name: Slack alert
        if: steps.detect.outputs.should_rollback == 'true'
        env:
          SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
        run: |
          curl -X POST $SLACK_WEBHOOK -d "{
            \"text\": \"🚨 Auto-rollback PR created. Score: ${{ steps.eval.outputs.score }}\"
          }"
flowchart TB
  PUSH[main 收到新 commit] --> RUN[每 2h 跑 smoke eval]
  RUN --> SAVE[persist 评测历史]
  SAVE --> DETECT{连续 3 次<br/>score < 阈值?}
  DETECT -->|否| OK[继续运行]
  DETECT -->|是| BAK[找到最近一次 good_sha]
  BAK --> BRC[新建 rollback 分支]
  BRC --> REV[git revert good..HEAD]
  REV --> PR[gh pr create<br/>label=auto-rollback,p0]
  PR --> SLK[Slack 通知 oncall]
  SLK --> WAIT[人工审核 → merge → roll back]
  style DETECT fill:#fff3e0
  style PR fill:#ffebee
  style SLK fill:#ffebee

工程实务的 5 条铁律:

  • 不直接 force-push main——任何回滚都走 PR,留 review 痕迹
  • revert 而非 reset——reset 会丢历史,revert 创建反向 commit 保留可追溯性
  • window=3 避免单次抖动误回滚——单次 score 异常可能是 LLM API 抖动
  • 保留 good_sha 字段在评测历史——脚本能自动找到"最后一次 score ≥ 阈"的 sha
  • rollback PR 必须带 P0 label + reviewer = oncall——回滚不该静默合入

触发过的自动回滚要逐次复盘:是真实退化,就说明机制保住了用户;是评测误报,就反过来推动评测改进。两种结果都有价值。

18.12.4 CI 评测的"灰度发布门禁"协同——把 staging → canary → 100% 流量纳入同一道评测闸门

许多团队的"CI 评测"只把守 PR merge 这一道门,但生产事故往往发生在"merge 后一周内的灰度阶段"。本节给读者一份把灰度发布全程纳入评测门禁的工程方案——让 staging → canary → ramp_25 → ramp_50 → 100% 每一步都被评测信号闸住,把"merge 即上线" 升级为"merge 后还有 4 道闸门"。

graph LR
    A[PR merge] --> B[staging 环境跑]
    B --> C{评测过?}
    C -->|否| Z[block 灰度]
    C -->|是| D[canary 1% 流量]
    D --> E{在线评测过?}
    E -->|否| Z2[自动回滚]
    E -->|是| F[ramp 25%]
    F --> G{评测过?}
    G -->|否| Z3[暂停 + 告警]
    G -->|是| H[ramp 50%]
    H --> I{评测过?}
    I -->|否| Z4[暂停]
    I -->|是| J[100% 全量]
    J --> K[持续在线监控]

5 阶段灰度门禁 × 评测信号 × 失败处置:

阶段 时长 评测信号 阈值 失败处置
staging 1h 离线 1000 题回归 0 critical regression block 灰度
canary 1% 4h 在线 trace 实时 grader hallucination ≤ baseline+1pp 自动回滚
ramp 25% 24h 离线 + 在线综合 4 维 metric 均 ≥ baseline-2pp 暂停 + 告警
ramp 50% 48h 综合 + 业务 KPI 客诉率 ≤ baseline+5% 暂停
100% 持续 完整在线评测 + 周报 无 critical 退化 监控

配套实现:灰度门禁评测协同器:

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Literal, Callable

ReleaseStage = Literal["staging", "canary_1", "ramp_25", "ramp_50", "full_100"]
GateOutcome = Literal["pass", "block", "rollback", "pause"]

STAGE_DURATIONS_HOURS = {
    "staging": 1, "canary_1": 4, "ramp_25": 24, "ramp_50": 48, "full_100": 0
}
NEXT_STAGE: dict[ReleaseStage, ReleaseStage | None] = {
    "staging": "canary_1", "canary_1": "ramp_25",
    "ramp_25": "ramp_50", "ramp_50": "full_100", "full_100": None,
}

@dataclass
class StageGateConfig:
    stage: ReleaseStage
    metric_name: str
    threshold_relative_to_baseline: float  # 例 -0.02 表示允许 -2pp
    runner: Callable[[], dict]

@dataclass
class GradualReleaseGateOrchestrator:
    configs: dict[ReleaseStage, list[StageGateConfig]] = field(default_factory=dict)
    baseline_metrics: dict[str, float] = field(default_factory=dict)
    release_log: list[dict] = field(default_factory=list)

    def evaluate_stage(self, stage: ReleaseStage) -> dict:
        gates = self.configs.get(stage, [])
        if not gates:
            return {"stage": stage, "outcome": "pass",
                    "reason": "no gates configured (skipped)"}
        results = []
        outcome: GateOutcome = "pass"
        for gate in gates:
            measured = gate.runner().get("score", 0.0)
            baseline = self.baseline_metrics.get(gate.metric_name, 0.0)
            delta = measured - baseline
            passed = delta >= gate.threshold_relative_to_baseline
            results.append({
                "metric": gate.metric_name,
                "measured": round(measured, 3),
                "baseline": round(baseline, 3),
                "delta": round(delta, 3),
                "threshold": gate.threshold_relative_to_baseline,
                "passed": passed,
            })
            if not passed:
                outcome = self._stage_failure_outcome(stage)
        decision = {
            "stage": stage, "outcome": outcome, "checks": results,
            "next_stage": NEXT_STAGE.get(stage) if outcome == "pass" else None,
            "ts": datetime.now().isoformat(),
        }
        self.release_log.append(decision)
        return decision

    def _stage_failure_outcome(self, stage: ReleaseStage) -> GateOutcome:
        if stage == "staging": return "block"
        if stage == "canary_1": return "rollback"
        return "pause"

    def progressive_release(self, max_stages: int | None = None) -> list[dict]:
        """从 staging 开始,按门禁结果决定是否进入下一阶段"""
        stage: ReleaseStage | None = "staging"
        results = []
        steps = 0
        while stage and (max_stages is None or steps < max_stages):
            decision = self.evaluate_stage(stage)
            results.append(decision)
            if decision["outcome"] != "pass":
                break
            stage = decision["next_stage"]
            steps += 1
        return results

    def auto_rollback_required(self) -> bool:
        for entry in reversed(self.release_log):
            if entry["outcome"] == "rollback":
                return True
            if entry["stage"] == "full_100" and entry["outcome"] == "pass":
                return False
        return False

举例(示意数据,需用你自己的数据复算):某团队上线新 prompt:

  • staging 1h → 0 critical regression → pass,进 canary
  • canary 1% × 4h → hallucination 0.04(baseline 0.03,+1pp,恰在"≤ baseline+1pp"的允许范围内)→ pass。注意上面的代码按"越高越好"比较(delta >= threshold),幻觉率这类越低越好的指标要先换成 1 − 幻觉率,再把阈值配成 −0.01;直接把 +0.01 配进去,幻觉率不变反而会判失败
  • ramp 25% × 24h → Faithfulness delta -0.025(threshold -0.02)→ pause + 告警
  • 调查发现长尾医疗 query 退化 → 修 prompt → 重启 ramp 25%
  • 修复后按原阶梯推进到 full_100

可参照的同类设计:

  • Progressive delivery:Spinnaker、Argo Rollouts 等工具的渐进发布
  • Canary analysis:Netflix 与 Google 开源的 Kayenta

读者把 GradualReleaseGateOrchestrator 接入 CD pipeline,就把"merge 即上线"变成了"merge 后还有 4 道评测闸门"。

18.12.5 "评测 + 主分支保护"的工程契约——避免人为 bypass

CI quality gate 最大的失败:工程师急着 ship → 找借口 bypass → 评测沦为"墙上贴纸"。下面给出 GitHub branch protection 的工程化契约:

# .github/branch_protection.yaml(settings as code 的示意格式)
# GitHub 本身不读这个文件;字段取自分支保护 REST API,
# 需借 Probot Settings、Terraform github_branch_protection 等工具落地
branch_protection_rules:
  main:
    required_status_checks:
      strict: true
      checks:
        - context: "evals/quality-gate"        # §18.10.1
          app_id: -1
        - context: "evals/safety-redteam"      # §16.11.1
          app_id: -1
        - context: "evals/ci-test-pass"
          app_id: -1
        - context: "evals/coverage-no-decrease"
          app_id: -1

    required_pull_request_reviews:
      required_approving_review_count: 2
      dismiss_stale_reviews: true
      require_code_owner_reviews: true   # §12.16.4 owner

    enforce_admins: true   # 关键:admin 也不能 bypass
    required_signatures: true
    required_linear_history: true
    allow_force_pushes: false
    allow_deletions: false

  release/*:
    required_status_checks:
      strict: true
      checks:
        - context: "evals/full-battery"
        - context: "evals/safety-redteam"
        - context: "evals/regression"
    required_pull_request_reviews:
      required_approving_review_count: 3
import requests
from dataclasses import dataclass
from typing import Iterable

@dataclass
class BypassAttempt:
    user: str
    timestamp: str
    reason_given: str
    bypassed_check: str
    deployment_succeeded: bool

class BypassAuditLogger:
    """记录所有"bypass 评测"事件——审计不可少"""

    def __init__(self, audit_log_path: str):
        self.path = audit_log_path

    def log_attempt(self, attempt: BypassAttempt):
        import json
        with open(self.path, "a") as f:
            f.write(json.dumps(attempt.__dict__) + "\n")

    def monthly_report(self) -> dict:
        import json
        from collections import Counter
        attempts = []
        with open(self.path) as f:
            for line in f:
                attempts.append(json.loads(line))

        by_user = Counter(a["user"] for a in attempts)
        by_check = Counter(a["bypassed_check"] for a in attempts)
        return {
            "total_bypasses_30d": len(attempts),
            "top_bypasser": by_user.most_common(1),
            "most_bypassed_check": by_check.most_common(1),
            "should_alarm": len(attempts) > 5,
        }

    def post_to_compliance_channel(self, attempt: BypassAttempt):
        """每次 bypass 必发到合规 Slack——全员可见"""
        # 简化:Slack webhook
        pass
flowchart LR
  PR[PR 想 merge] --> CHK[evals checks]
  CHK -->|全 pass| OK[正常 merge]
  CHK -->|失败| BLK{管理员 bypass?}
  BLK -->|是| LOG[bypass audit log]
  BLK -->|否| WAIT[修后再 merge]
  LOG --> SLK[Slack 通知 #compliance]
  LOG --> AUD[月度 bypass 报告]
  AUD -->|"5+ 次"| ESC[升级到 CTO]
  style BLK fill:#fff3e0
  style ESC fill:#ffebee
  style OK fill:#e8f5e9

工程实务的 5 条 branch protection 设计经验:

  1. enforce_admins=true 是核心:admin 也不能 bypass(防"老板说赶紧上")
  2. bypass 必须留痕:用 audit log + Slack 通知
  3. 每月 bypass > 5 次升级:常 bypass 说明 gate 设计有问题
  4. release/ 比 main 更严:require 3 reviewer + full battery
  5. 强制 linear history:防 messy merge 让 git blame 失效

具体例子(示意数据,需用你自己的数据复算):某团队 1 年 bypass 审计:

季度 bypass 次数 top bypasser 行动
Q1 12 CTO 5 次 警示 + CTO 同意修流程
Q2 6 工程师 A 3 次 1 on 1 谈话
Q3 2 急 fix 安全漏洞 合理
Q4 0 -- 文化建立

洞察:12 个月让"bypass 评测"从"日常操作"变成"特殊事件"——靠的不是技术、是契约 + 透明度。

3 类常见 bypass 错误:

错误 现象 修法
admin 默默 bypass 后期发现问题归 evals enforce_admins=true
reason 写"赶时间" 没真因 必须含 ticket / 业务理由
月度审计没人看 log 写了但无人 review 每月主管会议必看

读者把 branch_protection.yaml 入 git + 接 BypassAuditLogger,评测才真正成为"卡 PR 的硬门禁"而非"墙上规章"。

18.13 可靠性:flaky、依赖与演练

18.13.1 CI 评测的"flaky test 治理"——避免误报浪费工程师注意力

CI 评测的最大威胁不是"评测失败"——是flaky 失败。今天 fail / 明天 pass 的不稳定结果会让工程师对评测信号失去信任。下面是 flaky 检测与治理的工程方案:

import json
from dataclasses import dataclass
from collections import Counter, defaultdict
from datetime import datetime, timedelta
from typing import Iterable

@dataclass
class FlakyCaseAnalysis:
    case_id: str
    total_runs_30d: int
    fail_count: int
    pass_count: int
    flip_count: int          # pass → fail 或 fail → pass 切换次数
    flakiness_score: float   # 0-1,越高越 flaky
    is_flaky: bool
    last_consistent_run_days: int

class FlakyCaseDetector:
    """识别 CI 中表现不稳定的评测 case"""

    FLAKY_THRESHOLD_FLIP_RATE = 0.20    # > 20% 切换 = flaky
    MIN_RUNS_TO_DECIDE = 10

    def __init__(self, history_path: str):
        self.history_path = history_path

    def _load_runs(self, case_id: str) -> list[dict]:
        """读 case 最近 30 天的所有 CI run 结果"""
        with open(self.history_path) as f:
            return [json.loads(l) for l in f
                    if json.loads(l)["case_id"] == case_id]

    def analyze(self, case_id: str) -> FlakyCaseAnalysis:
        runs = self._load_runs(case_id)
        if len(runs) < self.MIN_RUNS_TO_DECIDE:
            return FlakyCaseAnalysis(
                case_id=case_id, total_runs_30d=len(runs),
                fail_count=0, pass_count=0, flip_count=0,
                flakiness_score=0.0, is_flaky=False,
                last_consistent_run_days=0,
            )

        runs_sorted = sorted(runs, key=lambda r: r["timestamp"])
        passes = sum(1 for r in runs_sorted if r["pass"])
        fails = len(runs_sorted) - passes

        flips = 0
        last_status = runs_sorted[0]["pass"]
        for r in runs_sorted[1:]:
            if r["pass"] != last_status:
                flips += 1
                last_status = r["pass"]

        flip_rate = flips / max(len(runs_sorted) - 1, 1)
        is_flaky = flip_rate >= self.FLAKY_THRESHOLD_FLIP_RATE

        # 最后一次"稳定段"的天数
        last_consistent = 0
        if runs_sorted:
            last_status = runs_sorted[-1]["pass"]
            for r in reversed(runs_sorted[:-1]):
                if r["pass"] != last_status:
                    break
                last_consistent += 1

        return FlakyCaseAnalysis(
            case_id=case_id,
            total_runs_30d=len(runs_sorted),
            fail_count=fails,
            pass_count=passes,
            flip_count=flips,
            flakiness_score=round(flip_rate, 3),
            is_flaky=is_flaky,
            last_consistent_run_days=last_consistent,
        )

    def quarantine_strategy(self, analysis: FlakyCaseAnalysis) -> str:
        """flaky case 该怎么处理"""
        if not analysis.is_flaky:
            return "stable: 维持"
        if analysis.flakiness_score >= 0.5:
            return ("severe_flaky: 立即 quarantine - 移到 'flaky' 标签集,"
                    "不阻塞 PR,但每周由 evals owner review")
        if analysis.flakiness_score >= 0.3:
            return ("moderate_flaky: 加 retry 3 次(多数表决),"
                    "若仍 flaky 则 quarantine")
        return "borderline: 增加运行次数到 20+ 再判定"
flowchart LR
  R[CI run 历史] --> A[FlakyCaseDetector]
  A --> CALC[flip_rate 计算]
  CALC --> Q1{flip rate?}
  Q1 -->|"< 20%"| ST[stable 维持]
  Q1 -->|"20-30%"| BD[borderline 收集更多数据]
  Q1 -->|"30-50%"| MID[moderate retry 3 次]
  Q1 -->|"≥ 50%"| SEV[severe quarantine 隔离]
  SEV --> RV[evals owner 周 review]
  MID --> RV
  RV --> FIX{找到根因?}
  FIX -->|是| BACK[回到 stable 集]
  FIX -->|否| RM[永久退役]
  style ST fill:#e8f5e9
  style SEV fill:#ffebee
  style RM fill:#fff3e0

工程实务的 4 条 flaky 处理原则:

  1. Quarantine 不删除:flaky case 移到独立标签,但保留——不少 flaky 背后是"模型真问题"
  2. 任何 flaky case 不阻塞 main:但每周必 review
  3. 找根因优先于忍受:常见根因——LLM API 抖动 / temperature 不为 0 / time-dependent context
  4. flaky 治理预算:每月 evals owner 4 小时专门治 flaky

3 类常见 flaky 根因:

根因 现象 修法
temperature > 0 同 prompt 多次结果不同 设 temperature=0(OpenAI 另可传 seed);二者都不保证完全确定,仍需多次取多数;部分新模型已不接受采样参数
time-dependent context 评测含"今天" / "本月"等相对时间 mock 当前时间或改 absolute
LLM API 抖动 偶发 5xx / 429 错误 带退避重试

具体例子(示意数据,需用你自己的数据复算):某团队 1000 题 CI 评测,30 天历史:

  • 870 个 stable(87%)
  • 80 个 borderline(20-30% flip)
  • 35 个 moderate(quarantine + retry)
  • 15 个 severe(永久 quarantine 标签)

每月 evals owner 4 小时清理 quarantine:5 个找到 root cause 回归,10 个永久退役。半年后 flaky 总数稳定在 30-40 个,不再增长。

研究背景:

  • Google Testing Blog《Flaky Tests at Google and How We Mitigate Them》(2016)讨论了 flaky test 的成因与治理
  • pytest-rerunfailures 等开源工具是同思路的实现

读者把 FlakyCaseDetector 接入团队 evals CI,flaky case 自动 quarantine、不阻塞 PR,每周 review 修治。

18.13.2 "CI 评测的依赖管理"工程实践——锁版本与定期升级

CI 评测依赖一堆 LLM SDK / 评测框架——下面给出"锁版本 + 定期升级"的 dependency 治理工程方案:

# evals-requirements.lock(precise pinning)
ragas==0.2.10
openai==1.55.0
anthropic==0.42.0
langfuse==2.55.0
pydantic>=2.10,<3.0          # 主版本范围
pyyaml>=6.0,<7.0
hashable                     # 内部包(无版本)
# 注意:promptfoo 是 npm 包,不在 pip 依赖里——它要单独锁进 package.json:
#   "devDependencies": { "promptfoo": "0.106.0" }
# 两份 lockfile 都要进 git,评测环境才是可复现的

# evals-requirements-dev.txt(开发额外)
pytest==8.3.0
pytest-cov==6.0.0
ruff==0.8.0
import json
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Iterable

@dataclass
class DependencyUpgradeRequest:
    package: str
    current_version: str
    available_version: str
    days_since_release: int
    breaking_changes: bool
    estimated_test_hours: float
    priority: str

class EvalsDependencyManager:
    """评测仓库的依赖管理"""

    UPGRADE_RULES = {
        "security_patch": {"max_wait_days": 7, "priority": "critical"},
        "minor_release": {"max_wait_days": 30, "priority": "medium"},
        "major_release": {"max_wait_days": 90, "priority": "low"},
    }

    def assess_upgrade(self, pkg: DependencyUpgradeRequest) -> dict:
        """对待升级包给出建议"""
        if pkg.breaking_changes:
            risk = "high"
            min_test = 4
        elif (pkg.current_version.rsplit(".", 1)[0]
              == pkg.available_version.rsplit(".", 1)[0]):   # 只有 patch 位变化
            risk = "low"
            min_test = 0.5
        else:
            risk = "medium"
            min_test = 2

        if pkg.days_since_release < 14:
            recommend = "WAIT: 新版本不到 2 周,社区 bug 暴露不充分"
        elif risk == "high" and pkg.estimated_test_hours < min_test:
            recommend = "STOP: 测试投入不足"
        else:
            recommend = "GO: 升级 + 跑 §10.10.3 task 巡检"

        return {
            "package": pkg.package,
            "from": pkg.current_version,
            "to": pkg.available_version,
            "risk": risk,
            "recommendation": recommend,
            "test_hours_needed": max(min_test, pkg.estimated_test_hours),
        }

    def quarterly_audit(self, lock_file: Path) -> dict:
        """季度 dependency audit"""
        # 简化:实际 pip list --outdated + GitHub releases API
        outdated = self._fetch_outdated()
        return {
            "total_deps": len(outdated),
            "security_patches": [d for d in outdated
                                  if d.get("type") == "security"],
            "minor_releases": [d for d in outdated
                                if d.get("type") == "minor"],
            "deferred_majors": [d for d in outdated
                                 if d.get("type") == "major"],
            "next_action": "本月升级 security; 下月 minor; 季度 major",
        }

    def _fetch_outdated(self) -> list[dict]:
        return []   # 简化
flowchart LR
  L[lock 文件] --> A[每周 audit]
  A --> NEW{有新版?}
  NEW -->|否| OK[持平]
  NEW -->|是| C{类型?}
  C -->|security patch| FAST[7d 内升级]
  C -->|minor| MID[30d 内升级]
  C -->|major| SLOW[90d 内评估]
  FAST --> TEST[测试 + §10.10.3 巡检]
  MID --> TEST
  SLOW --> TEST
  TEST -->|pass| MERGE[merge upgrade PR]
  TEST -->|fail| FIX[修兼容性]
  style FAST fill:#ffebee
  style MERGE fill:#e8f5e9

工程实务的 5 条 dependency 管理经验:

  1. 必锁 patch 级版本:ragas==0.2.10 而非 ragas>=0.2
  2. major / minor 分开 release schedule:security 7d / minor 30d / major 90d
  3. 新版本等 2 周再升级:让社区先暴露 bug
  4. 每次升级跑 §10.10.3 task 巡检:catch 兼容性问题
  5. 季度 dependency audit:避免债务累积

3 类常见 dependency 失控:

现象 原因 修法
半年没升级 没 audit cron 季度强制
跨 major 升级 拖到 0.1 → 0.3 每个 major 单独升
升级后 CI 全红 没回归测试 upgrade PR 必跑 full eval

具体例子(示意数据,需用你自己的数据复算):某团队 12 月 dependency 演化:

  • M1:lock 全部,base 11 个 deps
  • M3:升 ragas 0.1 → 0.2(major)→ 50 行代码改 + 1 周测试
  • M6:升 langfuse 1.x → 2.x(major)→ 3 天
  • M9:升 promptfoo(npm 侧)一个 minor 版本 → 半天,重点核断言清单有无增删
  • M12:security patch openai SDK → 1 小时

每次升级有 PR + test report,记录在 git history。

读者把 EvalsDependencyManager 接入团队 evals 仓库,避免"拖了半年发现升级路径断了"。

18.13.3 CI Quality Gate 的"故障演练"清单

CI Quality Gate 上线后还需要定期"故障演练"——验证它在异常情况下还能工作。给一份演练清单:

□ Judge API 故障演练: 故意把 judge model API key 改错, 看 CI 是否优雅降级
□ Dataset 损坏演练: 故意提交一个不规范的样例, 看 CI 是否报清晰错误
□ 阈值边界演练: 故意构造刚好低于阈值 1pp 的版本, 看 CI 是否准确拦截
□ Override 流程演练: 模拟紧急 hotfix, 走完 override 全流程
□ 告警风暴演练: 同时触发 5+ 个告警, 看运维系统是否扛得住
□ 评测耗时演练: 故意构造耗时长的评测, 看是否被合理 timeout
□ Provider 切换演练: judge model 从 OpenAI 切到 Claude, 验证迁移流畅

每季度跑 1-2 次完整演练,发现的问题立即修复——主动制造故障,比"等真出事再补救"靠谱得多。

18.14 落地与运维节奏

18.14.1 "CI Quality Gate 部署 30 天计划"

把全专栏最后一章的内容编成一份具体的 30 天部署计划:

[Week 1] 骨架
  Day 1-2: 选 CI 平台 (GitHub Actions / GitLab CI)
  Day 3-4: 写第 1 层 PR 子集评测 yaml
  Day 5: 接 1 个 webhook (Slack 通知)

[Week 2] 完善
  Day 8-9: 写第 2 层全集评测 yaml
  Day 10-11: 接 Prometheus / Grafana
  Day 12: 配关键阈值告警

[Week 3] 第三层
  Day 15-16: 写每日回归 yaml + cron 触发
  Day 17-18: baseline 比对脚本
  Day 19: 测试告警全链路

[Week 4] 治理
  Day 22-23: 设 override 机制
  Day 24-25: 写 runbook (告警响应 SOP)
  Day 26: 培训团队 + 上线

30 天后团队拥有完整三层 Quality Gate。每一步都对应本章具体小节的方法。

18.14.2 "评测体系上线后的 30/60/90 天检查"——避免上线即遗忘的工程仪式

新评测能力上线(无论是新 metric / 新 judge / 新 dataset)后最常见的失败模式:上线即遗忘——上线时的工程师离职 / 调岗后,没人再 care 它的健康度,3-6 个月后悄悄腐烂直到一次事故才被发现。本节给读者一份"30/60/90 天检查仪式"——把新评测从"上线"管到"长期可信"。

graph LR
    A[新评测能力上线] --> B[D+0 上线]
    B --> C[D+30 第一次健康检查]
    C --> D[D+60 第二次健康检查]
    D --> E[D+90 第三次健康检查]
    E --> F{90 天后状态}
    F -->|健康| G[纳入常态运维]
    F -->|降级| H[召集 owner + 改进 plan]
    F -->|废弃| I[正式 deprecate + 替换方案]
    C --> J[必看 4 项]
    D --> J
    E --> J
    J --> K[使用率]
    J --> L[准确性漂移]
    J --> M[阈值告警]
    J --> N[ owner 仍在岗]

30/60/90 天检查 4 维度 + 健康判定:

维度 D+30 阈值 D+60 阈值 D+90 阈值 不达标处置
使用率(PR 触发次数 / 周) ≥ 5 ≥ 10 ≥ 15 < 阈值 → 评估是否真有需求
Self-consistency 与上线时差异 ±2pp ±3pp ±5pp 超阈 → 重 calibration
阈值告警比例 ≤ 5% ≤ 5% ≤ 5% 持续高 → 调阈值或修能力
owner 在岗 yes yes yes 离职 → 必须 60 天内交接

配套实现:30/60/90 天检查器:

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Literal

CheckMilestone = Literal["D+30", "D+60", "D+90"]

@dataclass
class EvalCapabilityHealthCheck:
    capability_name: str
    launched_at: datetime
    weekly_usage_count: int
    consistency_at_launch: float
    consistency_now: float
    threshold_alert_pct: float
    owner_active: bool
    backup_owner_exists: bool

    def days_since_launch(self) -> int:
        return (datetime.now() - self.launched_at).days

    def current_milestone(self) -> CheckMilestone | None:
        d = self.days_since_launch()
        if d in range(28, 35): return "D+30"
        if d in range(58, 65): return "D+60"
        if d in range(88, 95): return "D+90"
        return None

    def assess(self) -> dict:
        m = self.current_milestone() or "D+90"  # 用最严的标准
        usage_thresholds = {"D+30": 5, "D+60": 10, "D+90": 15}
        consistency_drift_thresholds = {"D+30": 2.0, "D+60": 3.0, "D+90": 5.0}

        usage_ok = self.weekly_usage_count >= usage_thresholds[m]
        drift = abs(self.consistency_now - self.consistency_at_launch) * 100
        drift_ok = drift <= consistency_drift_thresholds[m]
        threshold_ok = self.threshold_alert_pct <= 5.0
        owner_ok = self.owner_active

        issues = []
        if not usage_ok:
            issues.append(f"使用率 {self.weekly_usage_count}/wk < {usage_thresholds[m]}(疑似无需求)")
        if not drift_ok:
            issues.append(f"一致性漂移 {drift:.2f}pp 超 {consistency_drift_thresholds[m]}pp")
        if not threshold_ok:
            issues.append(f"告警率 {self.threshold_alert_pct:.1f}% > 5%(阈值或能力需调整)")
        if not owner_ok:
            issues.append("owner 已离岗")
        if not self.backup_owner_exists:
            issues.append("无 backup owner(bus factor = 1)")

        verdict = "healthy" if not issues else (
            "needs_attention" if len(issues) <= 2 else "at_risk"
        )

        return {
            "milestone": m,
            "verdict": verdict,
            "issues": issues,
            "next_check": (datetime.now() + timedelta(days=30)).isoformat(),
        }

    def action_plan(self) -> list[str]:
        a = self.assess()
        plan = []
        if a["verdict"] == "at_risk":
            plan.append("立即召集 owner + platform team 进行 root-cause review")
        for issue in a["issues"]:
            if "使用率" in issue:
                plan.append("发布 internal blog 推广 + 在 onboarding 加入此能力介绍")
            if "一致性漂移" in issue:
                plan.append("用 §8 章 calibration suite 重新对齐 anchor")
            if "告警率" in issue:
                plan.append("评估是阈值过严还是能力实际退化")
            if "owner" in issue:
                plan.append("60 天内必交接 + 加 backup owner")
        return plan

举例(示意数据,需用你自己的数据复算):某团队 D+30 检查"新上线的 RAG faithfulness judge":

  • weekly_usage_count = 7(>5,ok)
  • consistency 0.86 → 0.85(drift 1pp,ok)
  • threshold_alert_pct = 8%(>5%,警告)
  • owner 在岗 / backup 存在
  • verdict = "needs_attention"
  • action_plan:评估告警率高的根因,调整阈值或补 calibration

D+60 时若问题仍在 → 升级为 at_risk → root-cause review。

读者可以把 EvalCapabilityHealthCheck 接入新评测能力上线的 PR template,上线时即生成 3 个 calendar 事件(D+30/60/90),让复查不依赖人记得。

18.14.3 CI Quality Gate 的"维护节奏"

最后给一份 CI Quality Gate 的长期维护节奏:

频率 动作
每天 看告警 / 处理失败 PR
每周 review 评测覆盖度 / 加 hard case
每月 调整阈值 / 看趋势曲线
每季度 元评测仪式 / SLA review
每年 工具升级 / 团队 retro

这种"频率分层"让 CI Quality Gate 不需要 24/7 关注,但也不会"建好后就忘"。把这份节奏表写进团队 wiki,作为 Quality Gate owner 的工作清单。

18.14.4 "评测体系运维 SLA"——给所有团队的承诺

评测体系是公司"基础设施"——任何 internal infra 都该有 SLA(Service Level Agreement)。下面给出工程化的运维 SLA 模板:

evals_platform_sla:
  publication_date: 2026-04
  next_review: 2026-07

  uptime_promises:
    eval_runner_availability: "99.5% (业务时段)"
    dashboard_availability: "99% / 24h"
    pr_comment_bot: "95% (允许偶发延迟)"

  performance_promises:
    pr_eval_completion_p95: "< 10 分钟"
    nightly_full_eval: "< 2 小时"
    alert_to_first_response: "P0 < 15 分钟 / P1 < 1 小时"

  data_promises:
    trace_retention: "见 §17.14.3 分层"
    eval_history: "12 月在线 + 5 年归档"
    audit_log: "7 年(合规要求)"

  support_promises:
    new_team_onboarding: "5 工作日"
    custom_metric_接入: "10 工作日"
    bug_response: "P0 < 2h / P1 < 1d / P2 < 1w"

  what_we_dont_promise:
    - "不保证评测分数本身的正确性(受 judge 限制)"
    - "不保证 100% catch 所有 bug"
    - "不保证完全替代人工 review"

  escalation_path:
    - "Slack #evals-help 一般问题"
    - "evals-oncall@ P0 紧急"
    - "VP Eng 持续问题升级"
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Iterable

@dataclass
class SLAComplianceReport:
    period_start: str
    period_end: str
    metric_name: str
    target: float
    actual: float
    breach_count: int
    breach_minutes: int
    sla_met: bool

class EvalsPlatformSLATracker:
    """跟踪评测平台的 SLA 达成情况"""

    SLA_TARGETS = {
        "eval_runner_availability": 0.995,
        "pr_eval_completion_p95_min": 10,
        "alert_response_p0_min": 15,
        "support_onboarding_days": 5,
    }

    def assess(self, period: str, actuals: dict) -> list[SLAComplianceReport]:
        results = []
        for metric, target in self.SLA_TARGETS.items():
            actual = actuals.get(metric, 0)
            if "availability" in metric:
                met = actual >= target
            else:   # latency / response time
                met = actual <= target

            results.append(SLAComplianceReport(
                period_start=period,
                period_end=period,
                metric_name=metric,
                target=target,
                actual=actual,
                breach_count=actuals.get(f"{metric}_breach_count", 0),
                breach_minutes=actuals.get(f"{metric}_breach_minutes", 0),
                sla_met=met,
            ))
        return results

    def quarterly_summary(self, reports: list[SLAComplianceReport]) -> dict:
        met = sum(1 for r in reports if r.sla_met)
        return {
            "total_slas": len(reports),
            "met_count": met,
            "compliance_pct": round(met / max(len(reports), 1), 3),
            "breach_summary": [
                f"{r.metric_name}: target {r.target} / actual {r.actual}"
                for r in reports if not r.sla_met
            ],
        }
flowchart LR
  E[评测平台运营] --> M[SLA Tracker]
  M --> S1[uptime]
  M --> S2[performance]
  M --> S3[support]
  S1 --> R{99.5%?}
  S2 --> P{p95 < 10min?}
  S3 --> O{onboarding < 5d?}
  R -->|否| BR[breach]
  P -->|否| BR
  O -->|否| BR
  BR --> COMP[赔偿 + RCA]
  R -->|是| OK
  P -->|是| OK
  O -->|是| OK[SLA met]
  style BR fill:#ffebee
  style OK fill:#e8f5e9

工程实务的 4 类 SLA 细节:

SLA 维度 健康范围 违约赔偿
平台可用性 99.5% 月费减免
PR eval 速度 P95 < 10min 紧急 oncall
onboarding < 5 工作日 主管道歉
audit log 留存 7 年 合规风险

具体例子(示意数据,需用你自己的数据复算):某团队 Q3 SLA 达成:

SLA 目标 实际 met?
uptime 99.5% 99.7% ✅
PR eval p95 < 10 min 8.5 min ✅
onboarding < 5d 7d ❌
P0 response < 15 min 18 min ❌

合规率 = 50% — 触发改进 plan:扩 evals owner 团队 + 改 onboarding 流程。

18.14.5 完整的 GitLab CI 实现示例

第 §18.3 给了 GitHub Actions 配置,国内团队大量使用 GitLab CI。等价的 GitLab 配置:

# .gitlab-ci.yml
stages:
  - eval-pr
  - eval-full
  - eval-regression

variables:
  EVAL_DATASET: "data/golden/v1.jsonl"

pr-subset-eval:
  stage: eval-pr
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
  timeout: 10m
  script:
    - pip install -r requirements-eval.txt
    # 多行命令要用 `- |` 字面块:普通多行标量会被 YAML 折成一行,
    # 行尾的 \ 会变成转义空格,参数就传错了
    - |
      python -m evals.run --dataset $EVAL_DATASET --sample-size 50 \
        --thresholds config/thresholds-subset.yaml \
        --report-file eval-report.json
  artifacts:
    when: always
    paths:
      - eval-report.json
    reports:
      junit: eval-junit.xml
  after_script:
    - |
      if [ -f eval-report.json ]; then
        python scripts/comment_on_mr.py --token "$MR_COMMENT_TOKEN" \
          --mr-iid "$CI_MERGE_REQUEST_IID" \
          --report eval-report.json
      fi

full-eval:
  stage: eval-full
  rules:
    - if: $CI_PIPELINE_SOURCE == "push" && $CI_COMMIT_REF_NAME == "main"
  timeout: 45m
  script:
    - pip install -r requirements-eval.txt
    - |
      python -m evals.run --dataset $EVAL_DATASET \
        --thresholds config/thresholds-full.yaml \
        --report-file eval-report-full.json
    - python scripts/push_metrics.py --report eval-report-full.json

daily-regression:
  stage: eval-regression
  rules:
    - if: $CI_PIPELINE_SOURCE == "schedule"
  timeout: 90m
  script:
    - pip install -r requirements-eval.txt
    - |
      python -m evals.run --dataset data/regression/v1.jsonl \
        --compare-to data/regression/v1-baseline.json \
        --alert-threshold 0.05
  after_script:
    # after_script 跑在**独立的 shell** 里,这里的 $? 是 after_script 自己上一条命令的
    # 退出码,拿不到主 script 的成败。要判断主 job 是否失败只能用 CI_JOB_STATUS
    - if [ "$CI_JOB_STATUS" != "success" ]; then curl -X POST $PAGERDUTY_WEBHOOK; fi

GitHub vs GitLab 配置在结构上几乎完全对称,主要差异:

  • GitHub Actions 用 if: ${{ github.event_name == ... }},GitLab 用 rules: - if: $CI_PIPELINE_SOURCE == ...
  • GitHub 的 secrets、GitLab 的 variables(masked)。注意评论 MR 用的 $MR_COMMENT_TOKEN 得是有 api 权限的 project / group access token——$CI_JOB_TOKEN 的权限范围不含写 MR 讨论, 拿它去调评论接口会 401
  • 评论 PR / MR 的 API 不同

国内团队大量用 GitLab,这份配置可以直接拷贝改用。Jenkins / Buildkite / CircleCI 等其他 CI 工具的等价配置也能照着翻译——逻辑是相同的:三 job × 三触发条件 × 不同 timeout。

18.14.6 完整的 Quality Gate Dashboard 设计

把全章方法落到一份完整的 Dashboard 设计:

主页(一目了然):

  • 当前 SLO 通过率 / Error Budget 剩余 / 本月失败次数
  • 7 天 Faithfulness 趋势曲线
  • 最近 5 次 PR 评测结果

详情页(按需深入):

  • 单次评测 run 的所有 case 详情
  • 失败 case 按指标分桶
  • 每条失败 case 链到 trace

告警页(关键事件):

  • 持续超阈值的指标
  • 异常 cost / latency
  • judge 自身漂移告警

审计页(合规需要):

  • 所有 override 记录
  • 评测集 / grader / prompt 版本变更日志
  • SLA 达标率月度报告

四个视图覆盖了开发者、PM、SRE、合规四类角色的需求,Grafana / Datadog / 自建都能做。

18.15 落地旅程与演化教训

18.15.1 完整团队的 12 个月评测旅程

把全专栏所有方法整合成"一个团队从 0 到完整评测体系"的 12 个月时间线:

[Month 1] 第一个 50 题黄金集 + 跑通 promptfoo 基本断言 → 第 3、5、12 章
[Month 2] 加入 LLM-as-Judge + Faithfulness/Recall → 第 6、11 章
[Month 3] 接入 langsmith / langfuse trace → 第 17 章
[Month 4] 第一次元评测 + 切换 judge 模型 → 第 8 章
[Month 5] 对抗集 + 红队 garak 接入 → 第 3、16 章
[Month 6] CI 集成第 1 层 PR 子集 → 第 18 章
[Month 7] CI 集成第 2 层全集 + Prometheus 看板 → 第 18 章
[Month 8] 多轮对话评测 + MT-Bench 借鉴方法 → 第 15 章
[Month 9] Agent 评测 + trajectory 监控 → 第 14 章
[Month 10] 安全评测全套 + OWASP 映射 → 第 16 章
[Month 11] Hard Case Mining 周度仪式 → 第 3 章
[Month 12] 元评测 SLA 化 + Eval Owner 角色 → 第 7、8 章

12 个月走完,团队从"凭感觉调 prompt"升级到"工业级评测体系"。具体到每月的工程量(示意值):

  • Month 1-3:1 人 50% 投入
  • Month 4-6:1 人全职 + 1 人辅助
  • Month 7-9:2 人全职 + 部分跨团队协作
  • Month 10-12:评测团队成型,3-5 人专职

把各阶段加起来是一笔不小的人力投入,值不值要拿它与一起 NYC MyCity 级别事故的公关与监管代价对比,用自家数据算。

18.15.2 CI Quality Gate 的演化教训

最后给一个典型的演化教训(示意情景)——Quality Gate 从严格到宽松的滑坡。

假设一个团队搭起 CI Quality Gate,初版严格:

  • Faithfulness ≥ 0.85
  • Recall ≥ 0.90
  • Latency p95 ≤ 3000ms

3 个月后开始出现"绕过"——某次紧急 hotfix override 后,团队发现"原来可以绕过"。半年后阈值被偷偷调松到 0.75/0.80/5000ms。1 年后团队发现"评测一直绿但用户投诉很多"——因为阈值早已名存实亡。

教训:Quality Gate 的阈值调松要比调严谨慎得多。一旦松了就很难再严回去——团队对"宽松状态"会形成依赖。

修法的标准做法:

  • 阈值变更必须 PR + 多人评审 + Review 在 Slack 公开
  • 每月自动报告"阈值变化历史",让管理层知道
  • 设"绝对底线"(如 jailbreak 95% / 安全相关),不可调

这种"治理 Quality Gate 的 Quality Gate"听起来递归,实际是评测体系的"自我保护"。

18.16 跨专栏关联

  • 本专栏第 2 章 §2.4 回归评测:本章是其工程实现
  • 本专栏第 4 章 §4.12.5 指标版本化:本章 §18.5 是其 CI 落地
  • 本专栏第 8 章 §8.8.2 元评测的季度仪式:本章 §18.7 Eval Owner 职责对应
  • 本专栏第 17 章 在线评测:本章 §18.4 时序告警是其 CI 镜像
  • 《Harness Engineering》专栏第 18 章评估与测试方法论:本章 GitHub Actions 是它在 CI 上的落地形态

18.17 本章小结

  • CI Quality Gate 把评测从"工程师的玩具"升级为"产品流程的硬门禁"
  • 三层门禁:PR 子集 (5 分钟) / 合并全集 (30 分钟) / 每日回归 (1 小时) —— 各层时长决定能不能被坚持
  • GitHub Actions 完整 yaml 配置覆盖三层门禁、自动 PR 评论、Prometheus 推送、Slack/PagerDuty 告警
  • 时序告警 + Grafana 看板能捕捉模型 / API 静默退化等外部因素
  • 指标版本化让"分数变了"和"评测器变了"可被分开归因
  • 三个组织角色:Eval Owner / LLM SRE / Domain Reviewer——没有 owner 的评测很容易荒废

——全专栏完。