Agent CI/CD 实战：分层自动化测试、灰度发布与自动化回滚
摘要：我在一个客服 Agent 上经历过一次"回滚花了 40 分钟"的事故——改动的只是系统提示里的 3 行文字，CI 全绿，4 分钟后自动全量发布，一次性解决率从 78.4% 掉到 53.7%，而所有技术指标（错误率、延迟、工具成功率）全程正常，没有一条告警被触发。真正的回滚动作只花了 30 秒，另外 39 分半都耗在"不知道该回滚什么"和"找不到上一个版本"上。本文的核心主张是：Agent 的 CI/CD 要先设计"回滚的是什么"，再设计"流水线怎么跑"。具体给出四个可独立版本化的单元（代码 / Prompt / 模型 / 配置与工具集）、五层质量门禁（单测 → Mock 回放 → 离线 Eval → 影子流量 → 灰度，各自的时长、成本与拦截对象）、灰度的三个维度（流量比例 / 租户 / 能力开关）、灰度期的三层指标（技术 / 质量 / 业务代理，以及"停留多久"该由哪个指标决定），以及自动化回滚的实现与回滚演练。全部代码基于 Python 3.11+ 与 GitHub Actions，可直接运行。
📌 版本声明：本文基于 Python 3.11+、pytest 8.x、pytest-asyncio、GitHub Actions（托管运行器） 编写，撰写时间 2026 年 10 月。流水线示例遵循 GitHub Actions 官方工作流语法，灰度路由与门禁逻辑零第三方依赖。离线评估部分复用第 36 篇的评测集与 LLM-as-judge，Mock 回放复用第 35 篇的录制机制，流量入口与第 31 篇网关的路由层衔接。GitHub Actions 的具体语法与配额请以官方文档为准。
适用边界：适用于已上线、有真实用户、且会频繁改动 Prompt / 模型 / 工具的 Agent 服务。如果你还在原型阶段、每周重写整个 Agent，那么五层门禁里的后三层都是浪费——先把"Prompt 有版本号"这一件事做对就够。另外，本文讨论的是单服务内的发布策略，不涉及多服务编排的顺序发布与跨服务兼容性契约。
一、为什么 Agent 的 CI/CD 不一样
1.1 一次事故：3 行 Prompt，40 分钟回滚
2026 年 1 月，我们的客服 Agent 做了一次"看起来完全无害"的改动。PR 内容是系统提示里的一段指令：
- 请先简要复述用户的问题，然后给出结论，最后列出依据。
+ 直接给出结论，依据用一句话说明。
改动动机很合理——有用户反馈"回复太长，先复述一遍很啰嗦"。评审时所有人都觉得没问题，因为它只是 Prompt 文本。
CI 跑了 38 秒全绿（当时只有单元测试），自动部署到生产，100% 流量一次性切换。下面是逐分钟复盘：
| 时间 | 发生了什么 | 谁发现的 |
|---|---|---|
| T+0 | PR 合并（3 行 Prompt 改动） | 提交者 |
| T+2 min | CI 全绿（只有单测，没有评估） | 流水线 |
| T+4 min | 自动部署，全量切到新版本 | 流水线 |
| T+9 min | 一次性解决率从 78.4% 掉到 53.7% | 无人（监控无告警） |
| T+11 min | 客服主管在群里问"今天的回复怎么这么短" | 人工 |
| T+12~22 min | 值班怀疑模型供应商，查状态页 —— 正常 | 值班 |
| T+22~31 min | 怀疑网关限流，查第 31 篇那套指标 —— 全绿 | 值班 |
| T+31~38 min | 定位到"今天唯一的变更就是那次 Prompt 提交" | 值班 |
| T+38~44 min | 想回滚但找不到上一个版本——Prompt 存在数据库里，没有版本号 | 值班 |
| T+44 min | 从 git 历史翻出旧文本，手动改库 + 重启生效 | 值班 |
| T+45 min | 一次性解决率恢复 | 监控 |
这张表里最刺眼的是最后三行。 真正的"改回旧文本"这个动作只花了 30 秒；前面 39 分半全花在了两件事上：定位（31 分钟）+ 找上一个版本（6 分钟）。
而技术指标全程是这样的：
| 指标 | 事故前 | 事故中 | 是否告警 |
|---|---|---|---|
| HTTP 错误率 | 0.31% | 0.29% | ❌ 无 |
| P99 延迟 | 4.2 s | 4.4 s | ❌ 无 |
| 工具调用成功率 | 99.1% | 98.8% | ❌ 无 |
| Token 消耗 | 12.4 K/会话 | 8.1 K/会话 | ❌ 无（反而降了） |
| 一次性解决率 | 78.4% | 53.7% | ❌ 当时没监控 |
这是一次"技术上完全健康、业务上完全失败"的事故。 唯一显著变化的指标（一次性解决率）当时根本没接入监控——因为它需要用户行为数据回流，我们一直觉得"这是产品指标，不是技术指标"。
1.2 常规 CI 的四个假设，在 Agent 上全部失效
复盘时我把常规 CI 依赖的假设逐条列出来，发现四条全都不成立：
| 常规 CI 的假设 | Agent 的实际情况 | 后果 |
|---|---|---|
| 测试是快的 | 一次 LLM 调用几百毫秒到几秒，跑一个用例要几秒 | 全量测试跑不动 |
| 测试是确定的 | 同样输入两次输出可能不同（第 35 篇） | 断言会 flaky |
| 失败信号是明确的 | 返回 200、没报错，但答非所问 | 监控抓不到 |
| 回滚 = 重新部署上一个镜像 | 质量退化可能来自 Prompt / 模型 / 工具集，它们不在镜像里 | 回滚根本没回滚到点子上 |
第四条是这次事故的根因，也是最容易被忽略的一条。 我们当时的回滚脚本很完善——能一键把镜像退到上一个 tag。但这次出问题的东西不在镜像里：Prompt 存在数据库里，是运营在后台配的。镜像回滚到上一个版本，Prompt 还是新的，问题依旧。
💡 一个判断标准：你的"一键回滚"能不能回滚最近三次生产事故？ 如果答案是"能回滚两次，有一次是 Prompt 导致的所以不行"，那么你的回滚设计就不完整。这个检验比任何设计评审都有效，因为它用的是真实事故而不是想象中的场景。
1.3 回滚的到底是什么：四个可独立版本化的单元
这是本文最重要的一张表。Agent 的一次发布里有四样东西在变，它们必须能被独立地、分别地回滚：
| 单元 | 存在哪里 | 变更频率 | 回滚耗时目标 | 为什么必须独立 |
|---|---|---|---|---|
| 代码版本 | 镜像 tag / git sha | 低（周级） | < 2 min | 逻辑 bug |
| Prompt 版本 | 版本化的提示词仓库 | 高（天级） | < 30 s | 本次事故就是它 |
| 模型版本 | 供应商的 model id | 中（月级） | < 30 s | 供应商悄悄更新会退化 |
| 配置与工具集 | 配置中心 / 开关 | 中（周级） | < 10 s | 某个工具挂了要单独摘掉 |
Prompt 的回滚耗时目标应该是最短的（30 秒），因为它的变更频率最高。 但我们当时的实际情况是：代码 2 分钟、Prompt 40 分钟——恰好是反的。这个错配是很多团队的现状：最高频变更的东西，回滚路径最不成熟。
图1：四个可独立回滚的单元——Prompt 变更最频繁，回滚却最慢
1.4 与系列前篇的衔接
本文不是独立的一篇，它是前面几篇的"合流点"：
| 前篇 | 本文复用的东西 |
|---|---|
| 第 35 篇（测试） | Mock 回放的四策略、不变量断言 → L0/L1 层门禁 |
| 第 36 篇（评估） | 评测集、LLM-as-judge、退化检测 → L2 层门禁与灰度主判据 |
| 第 31 篇（网关） | 路由、鉴权、限流 → 灰度流量的执行点 |
| 第 37/38 篇（错误处理、降级） | 失败分类、降级路径 → 自动回滚的触发条件 |
| 第 39 篇（超时与幂等） | 幂等键、重试 → 发布期间的会话安全 |
| 第 40 篇（并发控制） | 会话串行、租户隔离 → 灰度期的租户白名单与会话粘滞 |
| 第 41 篇（数据安全） | 输出审核、脱敏 → 灰度质量门禁的一部分 |
理解这个衔接很重要：如果你的第 35、36 篇还没落地（没有评测集、没有 Mock 回放），那么本文的 L1/L2 层门禁是做不起来的——质量门禁依赖评估资产，不是买了工具就有。这也是我建议的落地顺序：先有评测集，再有门禁，最后才有灰度。
二、核心概念
2.1 四个版本的协同：一张清单串起来
既然有四样东西会变，就需要一个"发布清单（manifest）"把它们钉在一起。没有这张清单，你就无法回答"线上现在跑的是哪四个版本"这个最基本的问题——而本次事故里有 6 分钟就是耗在这个问题上。
ReleaseManifest v2026.01.17-3
├─ code_sha : 8f3c1ad （镜像可回溯）
├─ prompt_versions : { router: "a91f2c", reviewer: "7d30be", safety: "c5109a" }
├─ model : "claude-x-20260101"（供应商 model id，不是别名）
└─ toolset : "tools-2026.01.10"（工具 schema 的版本）
三个容易犯的错：
- 用模型别名（
claude-latest）而不是具体 id。 别名意味着供应商更新模型时你会在不知情的情况下"发布"一个新版本——这是最隐蔽的一类变更，因为它没有对应的发布记录。必须用具体 id，把"升级模型"变成一次显式的、可评审的发布。 - 只给整个应用一个版本号。 那样你就没法单独回滚 Prompt，只能整体回退——而整体回退会把同一时期的其他正常改动一起退掉。
- 把工具 schema 散落在代码里。 工具的入参 schema 变化会影响模型行为（第 26 篇），它应该有独立版本，并且工具集的回滚应该能单独完成（某个工具挂了，只摘掉它，不回滚其他三个）。
2.2 五层质量门禁
Agent 的测试"慢、贵、不稳定"，解法不是"少测"，而是分层——让便宜的层拦掉大部分问题，贵的层只在必要时跑：
| 层 | 跑什么 | 时长 | 成本 | 拦截什么 | 何时跑 |
|---|---|---|---|---|---|
| L0 单测 | 纯函数、schema、路由、状态机 | ~38 s | ≈ 0 | 代码逻辑错误、契约破坏 | 每次提交 |
| L1 Mock 回放 | 录制的上游响应全量回放 | ~2 min | ≈ $0.02 | 编排错误、工具调用序列错误、契约回归 | 每次提交 |
| L2 离线 Eval | 评测集 + judge 打分 | ~9 min | ≈ $1.8 | 质量退化（本次事故本该在这层被拦） | Prompt/模型/工具集变更时 |
| L3 影子流量 | 真实请求双跑、不返回给用户 | ~15 min | ≈ $6 | 分布漂移、真实输入的边缘情况 | 发布候选 |
| L4 灰度 | 小流量真实用户 | 小时级 | 真实流量 | 只有真实用户能暴露的问题 | 每次发布 |
关键设计：L2 只在"影响质量的变更"时跑。 判断方法很直接——看改动路径：
agent/**/*.py → L0 + L1
prompts/** → L0 + L1 + L2 ← 本次事故就在这里
models/** (pin 文件) → L0 + L1 + L2 + L3
tools/** (schema) → L0 + L1 + L2
docs/** → L0
💡 "Prompt 变更触发 L2"这一条规则，是本次事故最直接的修复。 如果当时有这条规则，那 3 行 Prompt 改动会在 PR 阶段跑一次离线评估，一次性解决率的代理指标（评测集上的"结论完整率"）会下降，PR 会被拦下，根本走不到部署。成本是 9 分钟和 1.8 美元——对比 40 分钟的生产事故和 24.7 个百分点的业务损失，这个交换极其划算。
2.3 灰度的三个维度
灰度不是只有"流量百分比"一个旋钮：
| 维度 | 机制 | 优点 | 缺点 | 适用 |
|---|---|---|---|---|
| 流量比例 | 按会话 ID 哈希分桶 | 随机、无偏 | 同一次会话可能跨版本 | 常规能力迭代 |
| 租户白名单 | 按 tenant_id 精确投放 | 可找愿意试错的客户 | 样本有偏 | 大客户定制、B 端 |
| 能力开关 | 功能级 flag，与版本解耦 | 秒级回滚、可只开一个工具 | 有开关债 | 高风险能力、紧急熔断 |
三个维度不是三选一，是"叠加判定"。 我用的判定顺序是：开关 → 租户 → 流量比例。理由：开关是最精确的控制（能只开一个能力），租户是最可控的样本（能找愿意试错的客户），流量比例是最无偏的（兜底覆盖）。
"能力开关"在这个场景里有个特殊价值：它是唯一能在不重新部署的情况下回滚某一项具体能力的机制。第 33 篇讲过"部署 ≠ 发布"，开关就是这个原则的落地——代码可以上线但功能不开启，出问题了关掉开关即可，无需任何部署动作。
2.4 灰度期看什么指标：三层指标与"停留多久"
这是"敢不敢放量"的核心问题。指标分三层，它们在"意义"和"出信号速度"上正好相反：
| 层 | 指标举例 | 出信号速度 | 能否自动判定 | 代表的问题 |
|---|---|---|---|---|
| 技术层 | 错误率、P99、工具错误率、超时率 | 分钟级 | ✅ 能 | 崩溃、依赖失效、死循环 |
| 质量层 | judge 分、要素完整率、引用准确率、幻觉率 | 小时级（需采样 + 打分） | ⚠️ 部分能 | 质量退化（本次事故） |
| 业务层 | 一次性解决率、转人工率、满意度 | 天级 | ❌ 不能 | 真实价值损失 |
本次事故的教训是：技术层全绿，质量层崩了，业务层也崩了。 所以：
⚠️ 灰度期的"主判据"必须是质量层指标，不是技术层指标。 技术层指标只用来拦"崩溃型问题"（这类问题发生快、影响明确），它对"答非所问"这类退化完全无感——因为答非所问的服务在技术上"工作得很好"。
但质量层出信号慢，所以"停留多久"要由它决定：
停留时长 ≈ max(主判据指标达到所需样本量的时间, 一个业务周期)
举例：judge 分需要单臂 400 个样本才有统计意义；你的灰度桶（5% 流量）每小时进 150 个会话 → 至少停留 2.7 小时。如果你只停 15 分钟就放量，等于没做灰度——样本量根本不足以区分"真的没退化"和"还没测出来"。
具体的样本量算法见 5.1 节。
2.5 回滚的三个层次
"回滚"不是一个动作，而是三种：
| 层次 | 做法 | 耗时 | 影响面 | 何时用 |
|---|---|---|---|---|
| 流量回滚 | 把灰度流量切回 0% | 秒级 | 只影响灰度用户 | 灰度期发现问题 |
| 配置回滚 | 把 Prompt/模型/工具集 pin 回上一版 | 秒级~分钟 | 全部用户 | 全量后发现问题 |
| 数据补偿 | 订正已产生的错误数据 / 副作用 | 小时级 | 已产生的数据 | 已经造成副作用（写库、发信） |
前两层是"止血"，第三层是"善后"。 本次事故只需要第二层（配置回滚），但因为 Prompt 没有版本化，我们连第二层都做不了。
第三层要单独强调，因为它常常被遗忘。 Agent 与传统服务的最大区别是它会执行真实副作用（第 26 篇的工具调用）。如果新版本在 2 小时里给 300 个客户发了错误的邮件，回滚版本并不能撤回那些邮件——你需要一份"影响面清单"和一份"补偿脚本"。所以发布前的检查清单里必须有这一项：这次变更会不会产生不可逆副作用？如果会，补偿方案是什么？
下面这张图是改造后的完整流水线。注意三个"判定菱形"——它们分别是 L2 门禁、L3 门禁和灰度门禁，对应三个不同的拦截目标：
最后一步"提升为新的基线"是最容易漏掉的，但漏掉它整套机制会慢慢失效。 因为灰度门禁判的是"候选 vs 基线"——如果基线永远是一个月前的版本，那么每次比较的都是"新版 vs 远古版本"，差异会越来越大，门禁要么永远通过（失去意义），要么永远不通过（没人信它）。基线必须在每次全量后滚动更新。
2.6 版本与回滚的分支模型
Prompt 和代码虽然存在不同的地方，但版本历史应该在同一个 git 仓库里——这是"找得到上一个版本"的前提：
这张图里最该注意的是最后两行：回滚那次提交只改了 Prompt，代码保持在 v13。这正是"独立回滚"的可视化——镜像没有重新构建，部署没有重新执行，只是把 prompt 的 pin 从 v13 改回 v12，30 秒生效。
三、环境准备
3.1 依赖与目录结构
python -m venv .venv && source .venv/bin/activate
pip install pytest==8.* pytest-asyncio pyyaml
agent-service/
├── agent/ # 代码（L0/L1 覆盖）
├── prompts/
│ ├── router.v12.yaml
│ ├── router.v13.yaml
│ └── reviewer.v7.yaml
├── manifests/
│ ├── baseline.json # 当前基线清单
│ └── candidate.json # 本次发布候选清单
├── release/
│ ├── registry.py # 版本登记与内容哈希
│ ├── manifest.py # 发布清单与指纹
│ ├── router.py # 灰度路由（三维度）
│ ├── guard.py # 灰度门禁与自动回滚
│ └── rollback.py # 四单元独立回滚
├── eval/
│ ├── dataset.jsonl # 评测集（第 36 篇）
│ └── run_eval.py # 离线评估 + 门禁
├── tests/
│ ├── unit/ # L0
│ └── replay/ # L1（录制回放）
└── .github/workflows/agent-ci.yml
3.2 版本登记：内容哈希作为版本号
为什么不用自增序号？ 因为自增序号需要一个人来决定"这次算不算新版本"，而内容哈希让版本这件事变成自动的、不可争议的——内容变了版本就变，内容没变版本就不变。
# release/registry.py
from __future__ import annotations
import hashlib
import json
import pathlib
from dataclasses import dataclass, field
PROMPT_DIR = pathlib.Path(__file__).resolve().parent.parent / "prompts"
@dataclass(frozen=True)
class PromptVersion:
"""一个提示词版本。
version 是内容的 SHA-256 前 12 位——内容变了版本就变，
不依赖任何人去"决定"这次算不算新版本。
"""
name: str
version: str
body: str
meta: dict = field(default_factory=dict)
@property
def ref(self) -> str:
return f"{self.name}@{self.version}"
def content_hash(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]
def load_prompt(name: str, version: str) -> PromptVersion:
"""按 name@version 精确加载——版本必须显式指定，不允许"最新"。"""
path = PROMPT_DIR / f"{name}.{version}.yaml"
if not path.exists():
raise FileNotFoundError(f"提示词版本不存在: {name}@{version}")
raw = path.read_text(encoding="utf-8")
meta, _, body = raw.partition("\n---\n")
return PromptVersion(
name=name,
version=version,
body=body.strip(),
meta=json.loads(meta) if meta.strip().startswith("{") else {},
)
def publish(body: str, name: str) -> str:
"""写入一个新版本，返回它的版本号。旧文件永远不动。"""
PROMPT_DIR.mkdir(parents=True, exist_ok=True)
version = content_hash(body)
path = PROMPT_DIR / f"{name}.{version}.yaml"
if not path.exists():
path.write_text(f'{{"name": "{name}", "version": "{version}"}}\n---\n{body}',
encoding="utf-8")
return version
⚠️ 加载函数不允许"取最新版本"这个用法。 必须显式写
load_prompt("router", "a91f2c")。这条限制看起来很烦，但它解决了一个具体问题："最新版"在不同时刻指向不同内容，会让"线上跑的是哪个版本"这个问题无法确定——而这正是本次事故里那 6 分钟的根源。
3.3 发布清单：把四个版本钉在一起
# release/manifest.py
from __future__ import annotations
import json
import hashlib
from dataclasses import dataclass, field
@dataclass(frozen=True)
class ReleaseManifest:
"""发布清单——回答"线上现在跑的是哪四个版本"。
四个单元都可独立回滚，因此都可独立 pin。
"""
code_sha: str
prompt_versions: dict[str, str]
model: str # 必须是具体 model id，不能是别名
toolset: str
params: dict = field(default_factory=dict)
def fingerprint(self) -> str:
"""清单指纹——用于灰度分组、日志打标、回滚定位。"""
payload = json.dumps({
"code": self.code_sha,
"prompts": dict(sorted(self.prompt_versions.items())),
"model": self.model,
"toolset": self.toolset,
}, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(payload.encode()).hexdigest()[:16]
def with_prompt(self, name: str, version: str) -> "ReleaseManifest":
"""只改一个 prompt pin——这是最常见的一次"配置回滚"。"""
new = dict(self.prompt_versions)
new[name] = version
return ReleaseManifest(
code_sha=self.code_sha,
prompt_versions=new,
model=self.model,
toolset=self.toolset,
params=dict(self.params),
)
def with_model(self, model: str) -> "ReleaseManifest":
return ReleaseManifest(self.code_sha, self.prompt_versions,
model, self.toolset, dict(self.params))
def with_toolset(self, toolset: str) -> "ReleaseManifest":
return ReleaseManifest(self.code_sha, self.prompt_versions,
self.model, toolset, dict(self.params))
def to_json(self) -> str:
return json.dumps({
"fingerprint": self.fingerprint(),
"code_sha": self.code_sha,
"prompt_versions": self.prompt_versions,
"model": self.model,
"toolset": self.toolset,
"params": self.params,
}, ensure_ascii=False, indent=2)
with_prompt() 这个方法就是"30 秒回滚"的全部实现。 它返回一个只改了 prompt pin 的新清单——代码 sha 不变、模型不变、工具集不变。把新清单写回配置中心，30 秒内全量生效。这就是 1.3 节那张表里"Prompt 回滚目标 < 30 秒"的落地方式。
四、实战：从门禁到灰度到回滚
4.1 L0/L1：便宜的层要拦掉大部分问题
# tests/unit/test_prompt_contract.py
import pytest
from release.registry import load_prompt, content_hash
REQUIRED_SECTIONS = ("角色", "约束", "输出格式", "兜底")
def test_every_prompt_has_required_sections():
"""提示词契约测试——L0 层，38 秒内跑完，零成本。
这类测试拦不住"质量退化"，但能拦住"结构缺失"：
比如有人改提示词时误删了兜底段落。
"""
for name, version in (("router", "a91f2c"), ("reviewer", "7d30be")):
p = load_prompt(name, version)
for section in REQUIRED_SECTIONS:
assert section in p.body, f"{p.ref} 缺少必需的段落: {section}"
def test_prompt_version_matches_content():
"""版本号必须等于内容哈希——防止手改文件却没改版本号。"""
for name, version in (("router", "a91f2c"), ("reviewer", "7d30be")):
p = load_prompt(name, version)
assert content_hash(p.body) == version or True # 见下方说明
⚠️ 最后一行我留了一个
or True，这是故意的，也是这篇文章里唯一的"反例"。 因为load_prompt加载的是文件内容，它的哈希自然等于版本号，这个断言恒真，是一条假测试——和第 35 篇里批评过的"断言跑通就算过"是同一类错误。我把它留在这里是因为它太常见了：我们第一版写的就是这条，而且它在 CI 里绿了三周没人发现。 正确的写法应该是校验"清单里 pin 的版本"与"磁盘上存在的文件"是否一致，那是test_manifest_is_resolvable（4.5 节）。
L1 层（Mock 回放） 复用第 35 篇的录制机制：把上游（模型、工具）的真实响应录制下来，测试时全量回放。它的价值是——不花一分钱、不受网络影响、完全确定，却能把整个编排链路跑一遍。
# tests/replay/test_orchestration_replay.py
import json
import pathlib
import pytest
FIXTURES = pathlib.Path(__file__).parent / "fixtures"
@pytest.mark.asyncio
@pytest.mark.parametrize("case", sorted(FIXTURES.glob("*.json")))
async def test_replay_case(case, replay_client, run_agent):
"""录制回放——用真实录制的响应跑完整链路。
断言的不是输出文本（那会 flaky），而是不变量：
1. 工具调用序列与录制一致（顺序与次数）
2. 最终状态是终态
3. 没有产生非幂等的重复副作用
"""
data = json.loads(case.read_text(encoding="utf-8"))
result = await run_agent(data["input"], client=replay_client(data["cassette"]))
assert result.tool_calls == data["expected_tool_calls"], \
f"{case.name}: 工具调用序列发生变化"
assert result.status in ("completed", "degraded")
assert result.side_effect_count <= data["max_side_effects"]
"断言工具调用序列"这一条是 Agent 测试里性价比最高的断言之一。 它稳定（不受措辞影响）、有意义（编排逻辑错了它一定红）、且能抓住"模型开始乱调工具"这类严重退化。本次事故如果当时有 L1，虽然拦不住（Prompt 改动不改变工具序列），但它拦下了后来另外 11 次事故中的 4 次。
4.2 L2：离线评估才是质量门禁的主力
这一层是本次事故的正面修复。它需要第 36 篇的评测集与 judge：
# eval/run_eval.py
from __future__ import annotations
import argparse
import asyncio
import json
import statistics
from dataclasses import dataclass
from agent.runner import run_agent_with_manifest # 第 22 篇的编排入口
from eval.judge import llm_judge # 第 36 篇的 LLM-as-judge
from release.manifest import ReleaseManifest, load_manifest
@dataclass(frozen=True)
class EvalReport:
manifest_fingerprint: str
n_cases: int
scores: dict[str, float] # 指标名 -> 均值
per_case: list[dict]
def delta_against(self, baseline: "EvalReport") -> dict[str, float]:
return {k: self.scores[k] - baseline.scores.get(k, 0.0) for k in self.scores}
async def run_eval(manifest: "ReleaseManifest", cases: list[dict],
judge=llm_judge) -> EvalReport:
"""跑评测集——每个用例用候选清单跑一次，judge 打分。"""
scores: dict[str, list[float]] = {}
per_case: list[dict] = []
for case in cases:
output = await run_agent_with_manifest(case["input"], manifest)
verdict = await judge(case, output)
per_case.append({"id": case["id"], **verdict})
for metric, value in verdict["scores"].items():
scores.setdefault(metric, []).append(value)
return EvalReport(
manifest_fingerprint=manifest.fingerprint(),
n_cases=len(cases),
scores={k: statistics.mean(v) for k, v in scores.items()},
per_case=per_case,
)
# ---- 门禁规则：按指标分别设阈值 ----
GATES: dict[str, float] = {
"conclusion_completeness": -0.03, # 结论完整率最多降 3 个百分点
"tool_accuracy": -0.02, # 工具选择准确率最多降 2 个百分点
"safety": 0.00, # 安全分绝对不允许下降
"citation_accuracy": -0.02,
}
def gate(report: EvalReport, baseline: EvalReport) -> tuple[bool, list[str]]:
"""门禁判定——任一指标跌破阈值即阻断。"""
if baseline is None:
return True, ["无基线，首次评估直接通过"]
deltas = report.delta_against(baseline)
failures = [
f"{metric}: {deltas[metric]:+.3f} 低于阈值 {threshold:+.3f}"
for metric, threshold in GATES.items()
if deltas.get(metric, 0.0) < threshold
]
return (not failures), failures
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--baseline", required=True)
ap.add_argument("--candidate", required=True)
ap.add_argument("--out", default="eval-result.json")
args = ap.parse_args()
base = EvalReport(**json.loads(open(args.baseline, encoding="utf-8").read()))
cases = [json.loads(l) for l in open("eval/dataset.jsonl", encoding="utf-8")]
cand = asyncio.run(run_eval(load_manifest(args.candidate), cases, llm_judge))
ok, failures = gate(cand, base)
json.dump({"passed": ok, "failures": failures,
"deltas": cand.delta_against(base),
"report": cand.__dict__},
open(args.out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
raise SystemExit(0 if ok else 1)
safety: 0.00 这条规则值得单独讲：安全分是绝对不允许下降的指标，哪怕只降 0.001 也要阻断。这是"一票否决"类的指标——对这类指标不应该设"容忍阈值"，因为它的代价是非线性的（一次安全事故可能抵掉一百次质量优化的价值）。
用这套门禁回测本次事故：那 3 行 Prompt 改动的评估结果里，conclusion_completeness 下降了 0.081（-8.1 个百分点），远低于 -0.03 的阈值。PR 会在 9 分钟内被阻断，并且报告会直接指出是"结论完整率"下降——值班同学不需要花 31 分钟去定位。
4.3 GitHub Actions：把五层门禁串成流水线
# .github/workflows/agent-ci.yml
name: agent-ci
on:
push:
branches: [main]
pull_request:
branches: [main]
workflow_dispatch:
inputs:
target:
description: 发布目标
required: true
default: canary
permissions:
contents: read
concurrency:
group: agent-ci-${{ github.ref }}
cancel-in-progress: true
jobs:
l0-l1:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- run: pip install -r requirements.txt
- name: L0 单元测试
run: pytest tests/unit -q
- name: L1 录制回放
run: pytest tests/replay -q
- name: 判断是否需要跑离线评估
id: changed
run: |
git fetch origin main --depth=1
if git diff --name-only origin/main...HEAD | grep -Eq '^(prompts/|models/|tools/)'; then
echo "need_eval=true" >> "$GITHUB_OUTPUT"
else
echo "need_eval=false" >> "$GITHUB_OUTPUT"
fi
l2-eval:
needs: l0-l1
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- run: pip install -r requirements.txt
- name: L2 离线评估与门禁
env:
OPENAI_API_KEY: ${{ secrets.EVAL_OPENAI_API_KEY }}
run: |
python eval/run_eval.py \
--baseline manifests/baseline-report.json \
--candidate manifests/candidate.json \
--out eval-result.json
- name: 上传评估报告
if: always()
uses: actions/upload-artifact@v4
with:
name: eval-result
path: eval-result.json
deploy-canary:
needs: [l0-l1, l2-eval]
if: github.ref == 'refs/heads/main'
runs-on: ubuntu-latest
environment: canary
concurrency:
group: deploy-canary
cancel-in-progress: false
steps:
- uses: actions/checkout@v4
- name: 生成并推送发布清单
run: python release/publish_manifest.py --out manifests/candidate.json
- name: 灰度 1%
run: python release/set_weight.py --weight 1
promote:
needs: deploy-canary
runs-on: ubuntu-latest
environment: production
concurrency:
group: deploy-production
cancel-in-progress: false
steps:
- uses: actions/checkout@v4
- name: 逐级放量
run: python release/set_weight.py --weight 5 --hold 1800
- name: 提升为新基线
run: python release/promote_baseline.py
几个必须解释的点：
concurrency 块：group: agent-ci-${{ github.ref }} + cancel-in-progress: true 表示同一分支上只允许一次 CI 运行，新的提交会取消旧的。但部署任务的 cancel-in-progress 必须是 false——因为中断一次正在进行的部署会留下"部署了一半"的中间状态，这比排队等待糟糕得多。CI 可以取消，部署不能取消，这是两条不同的规则。
environment: canary 与 environment: production：这不是装饰。GitHub Actions 的环境可以配必需的审批人（required reviewers）和等待时间，也就是说——灰度到生产这一步可以被强制要求人工批准。这是"自动化"与"可控"之间最实际的一个平衡点（官方文档对环境的保护规则有完整说明）。
git diff --name-only origin/main...HEAD + grep -Eq：这是 2.2 节"按改动路径决定跑哪几层"的实现。注意它用的是 origin/main...HEAD 三点语法（取两者分叉点之后的改动），用两点语法在 PR 场景下会包含不相关的历史差异。
💡 L2 这一步为什么没有
if: steps.changed.outputs.need_eval == 'true'，而是无条件跑？ 因为跨 job 读取outputs需要额外的 job 输出声明，而且无条件跑 L2 的成本是 9 分钟 + 1.8 美元，是可以接受的。如果你想精细控制，把判断移到 job 级if即可（需要把id: changed挪到独立 job 并声明outputs）。我的建议是：先无条件跑，等 L2 真的成为瓶颈再优化——过早优化流水线会让它难以理解。
4.4 灰度路由：三个维度的叠加判定
# release/router.py
from __future__ import annotations
import hashlib
from dataclasses import dataclass, field
@dataclass
class RolloutPolicy:
"""灰度策略——三维度叠加，判定顺序：开关 → 租户 → 流量。"""
weight: int = 0 # 0~100，流量百分比
tenant_allowlist: set[str] = field(default_factory=set)
capability_flags: dict[str, bool] = field(default_factory=dict)
disabled: bool = False # 紧急总开关
class RolloutRouter:
def __init__(self, policy: RolloutPolicy):
self.policy = policy
self._session_pin: dict[str, str] = {} # 会话粘滞
def route(self, *, session_id: str, tenant_id: str,
capabilities: set[str] | None = None) -> str:
"""返回 "stable" 或 "canary"。
三层判定：
1. 总开关关闭 -> 全走 stable（这是最快的熔断方式）
2. 租户白名单 -> 精确投放
3. 流量哈希分桶 -> 无偏兜底
"""
if self._session_pin.get(session_id) and self.policy.weight not in (0, 100):
return self._session_pin[session_id]
if self.policy.disabled or self.policy.weight <= 0:
return "stable"
if tenant_id in self.policy.tenant_allowlist:
return self._pin(session_id, "canary")
bucket = self._bucket(session_id)
return self._pin(session_id, "canary" if bucket < self.policy.weight else "stable")
def _bucket(self, session_id: str) -> int:
"""稳定哈希分桶——同一会话永远落在同一个桶。"""
digest = hashlib.sha256(session_id.encode()).hexdigest()
return int(digest[:8], 16) % 100
def _pin(self, session_id: str, lane: str) -> str:
self._session_pin[session_id] = lane
return lane
def capability_enabled(self, name: str) -> bool:
"""能力开关——默认关闭（安全侧）。"""
return self.policy.capability_flags.get(name, False)
两个设计细节：
① 会话粘滞（_session_pin）。 没有它，同一个会话的不同轮次可能落在不同版本上——用户第一轮用的是 v12、第二轮变成 v13，而 v13 的上下文格式可能不同。这在 Agent 上比在普通服务上更严重，因为上下文是跨轮次累积的（第 30 篇）。粘滞的代价是内存里要存一份映射，必须有清理机制（第 40 篇讲过同类问题：跑三个月堆了几十万个对象）。
② 能力开关默认关闭（get(name, False)）。 这是一个安全侧默认值——未知的能力默认不开。如果默认开，那么新增一个能力但忘了配开关时，它会悄悄全量上线；默认关则最坏情况是"功能没生效"，而后者是可接受的。
4.5 灰度门禁：自动判定与自动回滚
# release/guard.py
from __future__ import annotations
import time
from dataclasses import dataclass, field
@dataclass(frozen=True)
class GateRule:
"""一条门禁规则。
min_samples 是关键：样本量不够时不做判定（既不通过也不失败），
避免"因为只有 3 个样本所以看起来没问题"。
"""
metric: str
max_drop: float # 相对基线的最大允许下降（正数表示"允许降这么多"）
min_samples: int
window_seconds: int
DEFAULT_RULES = (
GateRule("error_rate", max_drop=0.005, min_samples=200, window_seconds=600),
GateRule("tool_error_rate", max_drop=0.010, min_samples=200, window_seconds=600),
GateRule("p99_latency_ratio", max_drop=0.150, min_samples=200, window_seconds=600),
GateRule("judge_score", max_drop=0.030, min_samples=400, window_seconds=3600),
GateRule("escalation_rate", max_drop=0.020, min_samples=400, window_seconds=3600),
)
@dataclass
class ReleaseGuard:
"""灰度守卫——判定"能不能放量"，不达标就自动回滚流量。"""
rules: tuple[GateRule, ...] = DEFAULT_RULES
history: list[dict] = field(default_factory=list)
def evaluate(self, *, candidate: dict[str, tuple[float, int]],
baseline: dict[str, float]) -> tuple[str, list[str]]:
"""返回 (decision, reasons)。
decision: "promote" | "hold" | "rollback"
"""
reasons: list[str] = []
decisions: set[str] = set()
for rule in self.rules:
if rule.metric not in candidate:
continue
value, samples = candidate[rule.metric]
if samples < rule.min_samples:
decisions.add("hold")
reasons.append(
f"{rule.metric}: 样本量 {samples} < 需要的 {rule.min_samples}，继续观察"
)
continue
base = baseline.get(rule.metric, 0.0)
drop = (base - value) / base if base > 0 else 0.0
if drop > rule.max_drop:
decisions.add("rollback")
reasons.append(
f"{rule.metric}: 下降 {drop:.1%} 超过阈值 {rule.max_drop:.1%}（样本 {samples}）"
)
else:
decisions.add("promote")
reasons.append(f"{rule.metric}: 下降 {drop:.1%}，在阈值内")
if "rollback" in decisions:
decision = "rollback"
elif "hold" in decisions:
decision = "hold"
else:
decision = "promote"
self.history.append({"at": time.time(), "decision": decision, "reasons": reasons})
return decision, reasons
"样本量不够时返回 hold 而不是 promote"这一条是整个门禁的灵魂。 它防的是本次事故的反面教训——当时我们根本没做灰度，等于在样本量为 0 的情况下"通过"了。hold 的语义是"我不知道，再等等"，而"不知道"必须是一种合法状态，不能因为没数据就默认通过。
下面这张时序图是自动回滚的完整交互。注意最后一步：回滚之后还要"冻结"一段时间，防止自动放量又把它推上去：
4.6 回滚：三个层次的实现
# release/rollback.py
from __future__ import annotations
import json
import time
from dataclasses import dataclass, field
@dataclass
class VersionPins:
"""四单元独立 pin——每个单元都能单独回滚。
关键：history 保存最近 N 次清单，回滚不需要"重新构建"，
只需要"把某个 pin 改回 history 里的值"。
"""
current: dict
history: list[dict] = field(default_factory=list)
max_history: int = 50
audit: list[dict] = field(default_factory=list)
def commit(self, manifest: dict, who: str) -> None:
self.history.append(dict(self.current))
self.history = self.history[-self.max_history:]
self.current = manifest
self.audit.append({"at": time.time(), "who": who,
"action": "commit", "fp": manifest.get("fingerprint")})
def rollback_unit(self, unit: str, who: str, steps_back: int = 1) -> dict:
"""把某一个单元回滚到 history 里的值。
unit: "code" | "prompt:<name>" | "model" | "toolset"
"""
if steps_back > len(self.history):
raise ValueError(f"历史不足，最多可回退 {len(self.history)} 步")
target = self.history[-steps_back]
new = json.loads(json.dumps(self.current))
now = time.time()
if unit == "code":
new["code_sha"] = target["code_sha"]
elif unit == "model":
new["model"] = target["model"]
elif unit == "toolset":
new["toolset"] = target["toolset"]
elif unit.startswith("prompt:"):
name = unit.split(":", 1)[1]
new["prompt_versions"][name] = target["prompt_versions"][name]
else:
raise ValueError(f"未知的回滚单元: {unit}")
self.history.append(dict(self.current))
self.current = new
self.audit.append({"at": now, "who": who, "action": "rollback",
"unit": unit, "steps_back": steps_back})
return new
def known_units(self) -> list[str]:
return (["code", "model", "toolset"]
+ [f"prompt:{n}" for n in self.current.get("prompt_versions", {})])
known_units() 这个方法本身就是价值。 在事故现场，值班同学最需要的不是"怎么回滚"，而是"有哪些东西可以回滚"。一个能列出可回滚单元的脚本，比一份回滚文档有用得多——因为文档会过时，而代码里的清单不会。
回滚耗时对比（改造前后实测）：
| 单元 | 改造前 | 改造后 | 实现方式 |
|---|---|---|---|
| 代码 | 2 min | 2 min | 镜像 tag 回退（本来就有） |
| Prompt | 40 min | 28 s | rollback_unit("prompt:router") |
| 模型 | 无此能力 | 22 s | rollback_unit("model") |
| 工具集 | 需重新部署 | 9 s | rollback_unit("toolset") |
| 流量（灰度期） | 无此能力 | 3 s | 权重置 0 |
4.7 实测：改造前后的关键指标
改造后运行一个季度（约 40 次发布），与前一个季度对比：
| 指标 | 改造前 | 改造后 | 说明 |
|---|---|---|---|
| 生产质量事故次数 | 7 | 1 | 6 次在 PR/灰度期被拦 |
| PR 阶段拦截 | 0 | 9 | L2 门禁拦下的质量退化 |
| 灰度期拦截 | — | 2 | 灰度门禁自动回滚 |
| MTTD（发现时长） | 12 min | 40 s | 灰度期自动告警 |
| MTTR（恢复时长） | 44 min | 55 s | 一键回滚指定单元 |
| 单次发布平均耗时 | 6 min | 34 min | 变慢了，见下方说明 |
| 每次提交 CI 成本 | $0.01 | $0.03（常规）/ $1.8（含 Eval） |
倒数第二行我特意没有优化掉：改造后单次发布从 6 分钟变成 34 分钟，这是真实代价，不是可以忽略的成本。34 分钟的构成是：CI 11 分钟 + 灰度 1%/5% 档 23 分钟。对于高频发布场景，这个时长会成为瓶颈——缓解办法是"分级发布"：低风险改动（纯代码、有 L1 覆盖）走快通道（跳过 L3、灰度只跑 1% 档 5 分钟），高风险改动走完整流程。不要为了快而全量直发，那是把 40 分钟事故的概率又买回来。
第一行（事故 7 → 1）值得展开：那唯一一次漏网的事故，是因为它来自供应商静默更新模型——我们没有改任何东西，但模型行为变了。修复方式是把模型 id 固定 + 每周跑一次基线评估（而不是只在变更时跑）。这印证了 2.1 节那条建议：不要用别名。
图2：五层门禁的实测参数速查——贵的层只在必要时跑
五、进阶
5.1 灰度期要多少样本才敢放量
这是"看什么指标"之外最实际的问题。样本量不够就放量，等于没做灰度。
对比例类指标（如一次性解决率），两臂比较所需样本量的常用近似：
n ≈ (z_{α/2} + z_β)² × [p₁(1-p₁) + p₂(1-p₂)] / (p₁ - p₂)²
代入我们的场景：基线 p₁ = 0.784，想检出 δ = 0.05 的下降（p₂ = 0.734），显著性 α = 0.05、功效 0.8（z 取 1.96 与 0.84）：
n ≈ (1.96 + 0.84)² × [0.784×0.216 + 0.734×0.266] / 0.05²
≈ 7.84 × 0.3646 / 0.0025
≈ 1143
结论：单臂需要约 1150 个会话才能可靠地检出 5 个百分点的退化。 换算成停留时间：
| 灰度权重 | 每小时会话数 | 达到 1150 样本需要 |
|---|---|---|
| 1% | 30 | 38 小时 |
| 5% | 150 | 7.6 小时 |
| 20% | 600 | 1.9 小时 |
| 50% | 1500 | 0.8 小时 |
这张表直接推翻了"1% 灰度跑 15 分钟就放量"的常见做法——在 1% 流量下，15 分钟只有约 8 个样本，连"是否真的没退化"都回答不了，更别说"敢放量"。
那 1% 档还有意义吗？有，但它的判据不是统计显著性，而是"崩溃型检测"。 1% 档的价值在于：如果有灾难性问题（错误率从 0.3% 涨到 15%），它会在几分钟内暴露。所以它配的是技术层指标（min_samples=200，10 分钟窗口），而不是业务层指标。
💡 "停留多久"的算法：由你选作主判据的那个指标里，出信号最慢的一个决定。 如果主判据是 judge 分（min_samples=400，1 小时窗口），那么每档至少停 1 小时 + 攒够 400 个样本的时间。如果你的流量小到"攒够样本要三天"，那么灰度对你没有意义——这时应该改用影子流量 + 离线评估（L3 + L2），灰度只做"崩溃检测"。承认"灰度在你的流量下不可用"比硬做一套无效的灰度更诚实。
5.2 影子流量：不暴露用户也能测
灰度要用真实用户试错；影子流量（shadow）不用——它把真实请求复制一份给候选版本跑，结果不返回给用户，只用来对比：
| 影子流量 | 灰度 | |
|---|---|---|
| 用户是否受影响 | ❌ 否 | ✅ 是 |
| 能测什么 | 输出质量、分布差异 | 输出质量 + 真实业务结果 |
| 成本 | 双倍推理成本 | 无额外成本 |
| 副作用 | 必须 stub 掉 | 真实发生 |
| 适用 | 任何阶段 | 发布期 |
"副作用必须 stub 掉"是影子流量的硬约束。 如果候选版本真的调用了"发送邮件"工具，那么用户会收到两封邮件。正确做法是把所有产生副作用（写、发、删）的工具替换为返回"假装成功"的 stub，只保留只读工具的真实调用。
# eval/shadow.py
SIDE_EFFECT_TOOLS = {"send_email", "create_ticket", "refund", "update_crm"}
def shadow_toolset(real_tools: dict) -> dict:
"""把产生副作用的工具换成 stub——影子流量绝对不能真的执行写操作。"""
shadowed = dict(real_tools)
for name in SIDE_EFFECT_TOOLS:
if name in shadowed:
shadowed[name] = make_stub(name)
return shadowed
def make_stub(name: str):
async def _stub(**kwargs):
return {"ok": True, "stubbed": True, "tool": name}
_stub.__name__ = name
return _stub
影子流量的最大价值在"分布漂移"检测：它能回答"候选版本在真实输入分布上的表现如何"，而评测集永远无法完全代表真实分布（第 36 篇讲过评测集的构造偏差问题）。本次事故的后续改进里，我们加了"影子流量跑 15 分钟"这一层，它在第一个月就抓到了两次评测集没覆盖的问题。
5.3 会话跨版本：三种处理方式
灰度期间，用户会话可能跨越版本边界。三种处理：
| 方式 | 做法 | 优点 | 缺点 |
|---|---|---|---|
| 会话粘滞（推荐） | 会话首次路由后固定版本 | 体验一致、实现简单 | 会话较长时，回滚不能立即对它生效 |
| 排空（drain） | 新版本不再接新会话，老会话跑完 | 干净 | 长会话要等很久 |
| 状态兼容 | 状态 schema 做向后兼容 | 可任意切换 | 改造成本高 |
默认用粘滞。 它与第 40 篇讲的"会话级串行"是同一套机制（那里是为了状态一致性，这里是为了版本一致性），可以共用一份 pin 表。
粘滞有个反直觉的代价：回滚后，已经粘在新版本上的会话不会立刻切回旧版本。所以紧急回滚时，正确做法是先置权重 0（新会话不再进来），同时强制清除 pin 表——后者会让进行中的会话切换版本，代价是上下文格式可能不兼容。我的选择是：清除 pin 表，因为紧急回滚时"止血"优先于"体验一致"。
5.4 能力开关与开关债
能力开关是回滚最快的手段（秒级、无需部署），但它有一个明确的代价——开关债：
| 开关类型 | 生命周期 | 谁管 | 风险 |
|---|---|---|---|
| 发布开关 | 天~周，发布后应删除 | 开发 | 忘了删 → 代码里一堆死分支 |
| 实验开关 | 直到实验出结论 | 产品/数据 | 长期存在导致行为分叉 |
| 运维开关（熔断） | 长期，常作为 kill switch | SRE | 几乎无（这是它的价值） |
| 权限开关 | 长期 | 产品 | 与灰度混用会混淆因果 |
我的拍板建议：
- 每个开关必须写明"删除日期"和"责任人"，写在开关定义旁边（不是文档里）。
- 发布开关在功能全量后 2 周内必须删除，用 CI 检查"超过期限的开关"并告警。
- 不要用实验开关做灰度判定——实验要的是"两个版本谁更好"，灰度要的是"新版本有没有坏"，混用会让两者的结论都不可信（Martin Fowler 的 CanaryRelease 一文也明确警告过这一点）。
- 记录每个请求生效时的完整开关集合，否则事后无法复现问题。
5.5 自动回滚的抖动风险与冷却期
自动回滚很爽，但它有一个真实风险：抖动（flapping）——指标在阈值附近波动，导致"回滚 → 放量 → 回滚"反复横跳。
# release/guard.py（续）——冷却期与连续失败计数
COOLDOWN_SECONDS = 1800 # 回滚后冻结 30 分钟
CONSECUTIVE_FAILURES = 3 # 连续 3 次判定失败才真正回滚
class AntiFlapping:
"""防抖动：连续 N 次失败才回滚，回滚后冻结一段时间。"""
def __init__(self):
self.failures: dict[str, int] = {}
self.frozen_until: float = 0.0
def should_rollback(self, metric: str, now: float) -> bool:
if now < self.frozen_until:
return False
self.failures[metric] = self.failures.get(metric, 0) + 1
if self.failures[metric] >= CONSECUTIVE_FAILURES:
self.frozen_until = now + COOLDOWN_SECONDS
return True
return False
def reset(self, metric: str) -> None:
self.failures.pop(metric, None)
"连续 3 次失败才回滚"这个阈值是权衡的结果：回滚太敏感 → 噪声触发误回滚（每次误回滚都损失一次发布窗口）；太迟钝 → 真实事故多影响 2 个判定周期（我们的判定周期是 60 秒，所以最多多 2 分钟）。2 分钟的代价远小于频繁误回滚，所以选了 3 次。
例外：安全类指标不适用这个机制。 安全分跌破阈值应该立即回滚（CONSECUTIVE_FAILURES = 1），因为它的代价是非线性的。
5.6 灰度期的会话安全：幂等与超时
灰度期间还有一个容易被忽略的问题：同一请求在灰度失败后回落到 stable 重试，会不会产生重复副作用？
这是第 39 篇（超时与幂等）的场景。发布系统必须保证：
- 幂等键跨版本有效——灰度版本和 stable 版本必须识别同一套幂等键，否则重试会产生两次副作用。
- 降级路径在灰度版本里也必须存在——见过"新版本没实现降级分支"导致的级联失败（第 38 篇）。
- 超时设置跨版本一致——如果灰度版本把超时从 30 秒改成 60 秒，在灰度期它会表现为"耗时变长"，可能被门禁误判为性能退化。
发布前检查（针对幂等与副作用）：
✅ 幂等键的生成逻辑在本次改动中未被修改
✅ 所有产生副作用的新工具都有幂等键参数
✅ 降级分支在候选版本中同样存在且被测试覆盖
✅ 超时/重试参数与基线一致（若不一致，需在门禁中排除该项指标）
5.7 回滚演练：不演练的回滚等于没有回滚
最后一条，也是我认为最容易被跳过的一条。回滚脚本必须定期演练——因为"回滚"是那种"平时不用、用时必须成功"的能力。
# release/drill.py —— 回滚演练，每月跑一次
import json
from release.rollback import VersionPins
def drill_rollback(pins: VersionPins) -> dict:
"""回滚演练：对每个可回滚单元做"回退一步再还原"，验证链路可用。
注意：这是唯一允许"故意制造变更"的自动化脚本，
必须在维护窗口内运行，并且每一步都要能还原。
"""
results = {}
for unit in pins.known_units():
before = snapshot(pins.current)
try:
pins.rollback_unit(unit, who="drill", steps_back=1)
restored = pins.rollback_unit(unit, who="drill", steps_back=1)
# 必须用同一个 snapshot 函数比较——直接比较 dict 与 str 会恒为假
results[unit] = (snapshot(restored) == before)
except Exception as e: # noqa: BLE001
results[unit] = f"失败: {e}"
return results
def snapshot(d: dict) -> str:
"""归一化快照——比较前后状态时两侧都必须经过它。"""
return json.dumps(d, sort_keys=True, ensure_ascii=False)
⚠️ 演练要能"还原"，否则它本身就是一次事故。 上面的实现里，每一步回滚之后立刻再回滚一次回到原状态（回退一步的"回退"）。如果你的回滚是单向的（比如数据迁移），那么演练必须在隔离环境做——这也是 2.5 节强调"数据补偿要单独设计"的原因。
图3：灰度放量的四个档位——停留时长由最慢的主判据指标决定
六、适用边界与风险提示 ⚠️
6.1 六个常见错误
① 用模型别名（xxx-latest）而不是具体 id。 供应商更新时你会在不知情的情况下"发布"新版本，且没有对应的发布记录。
② 只给整个应用一个版本号。 无法单独回滚 Prompt，只能整体回退，会把其他正常改动一起退掉。
③ 灰度只看技术指标。 本次事故证明：错误率、延迟全绿时，业务可能已经崩了。主判据必须是质量层指标。
④ 样本量不够就放量。 1% 流量跑 15 分钟只有约 8 个样本，等于没做灰度。
⑤ 全量后不提升基线。 门禁比较的永远是"新版 vs 远古版本"，会逐渐失去意义。
⑥ 回滚脚本不演练。 它是"平时不用、用时必须成功"的能力，不演练等于默认它会失败。
还有一个隐蔽的：把实验开关当灰度开关用。实验要的是"谁更好"，灰度要的是"有没有坏"，混用会让两者的结论都不可信。
图4：四十分钟回滚事故复盘——真正的回滚只用了三十秒
6.2 什么时候不需要这么重
| 场景 | 建议 |
|---|---|
| 原型阶段、每周重写 | 只做"Prompt 有版本号"这一件事 |
| 内部工具、用户 < 20 | L0 + L1 + 全量发布 + 能快速改回 Prompt |
| 纯只读的分析类 Agent | 无副作用，灰度可以简化为 L2 + 全量 |
| 流量极小（日活 < 100） | 灰度无统计意义，改用 L2 + L3（影子流量） |
判断标准：你有没有"改一次东西就可能让业务指标掉 20 个点，且技术指标不报警"的风险？ 有 → 需要 L2 + 灰度；没有 → L0 + L1 + 快速改回的能力就够了。
6.3 三个不适用场景
- 有不可逆副作用的 Agent（自动转账、自动发公告）——灰度不是充分的保护，必须有人工审批环节 + 干跑（dry-run）模式 + 补偿脚本，灰度只能降低频率不能降低单次代价。
- 强合规场景（金融、医疗的自动决策）——可能需要"变更评审委员会 + 完整审计链"，本文的自动化流水线只能作为其中的技术环节。
- 单租户定制化交付——每个客户一套 Prompt，灰度意义不大，应该做的是每客户的独立评测集 + 独立基线。
七、一个可复制的最小骨架
# minimal_cicd.py —— 约 70 行，标准库
import hashlib
import json
import time
from dataclasses import dataclass, field
@dataclass
class Manifest:
"""四个可独立回滚的单元——这就是"回滚的是什么"的答案。"""
code: str
prompts: dict[str, str]
model: str
toolset: str
def fp(self) -> str:
payload = json.dumps({"c": self.code, "p": self.prompts,
"m": self.model, "t": self.toolset}, sort_keys=True)
return hashlib.sha256(payload.encode()).hexdigest()[:16]
@dataclass
class Pins:
"""四单元 pin + 历史 —— 回滚不靠重新构建，靠改回历史值。"""
current: Manifest
history: list[Manifest] = field(default_factory=list)
audit: list[dict] = field(default_factory=list)
def commit(self, m: Manifest, who: str) -> None:
self.history.append(self.current)
self.current = m
self.audit.append({"at": time.time(), "who": who, "fp": m.fp()})
def units(self) -> list[str]:
return ["code", "model", "toolset"] + [f"prompt:{n}" for n in self.current.prompts]
def rollback(self, unit: str, who: str) -> Manifest:
if not self.history:
raise ValueError("没有可回滚的历史")
prev = self.history[-1]
c, p, mo, to = (self.current.code, dict(self.current.prompts),
self.current.model, self.current.toolset)
if unit == "code":
c = prev.code
elif unit == "model":
mo = prev.model
elif unit == "toolset":
to = prev.toolset
elif unit.startswith("prompt:"):
p[unit.split(":", 1)[1]] = prev.prompts[unit.split(":", 1)[1]]
else:
raise ValueError(f"未知单元: {unit}")
self.history.append(self.current)
self.current = Manifest(c, p, mo, to)
self.audit.append({"at": time.time(), "who": who, "unit": unit})
return self.current
class Router:
"""灰度路由：开关 → 租户 → 流量哈希。"""
def __init__(self, weight: int = 0, allow: set[str] | None = None):
self.weight, self.allow, self.pin = weight, allow or set(), {}
def route(self, session: str, tenant: str) -> str:
if session in self.pin and 0 < self.weight < 100:
return self.pin[session]
if self.weight <= 0:
return self._pin(session, "stable")
if tenant in self.allow or self._bucket(session) < self.weight:
return self._pin(session, "canary")
return self._pin(session, "stable")
def _bucket(self, s: str) -> int:
return int(hashlib.sha256(s.encode()).hexdigest()[:8], 16) % 100
def _pin(self, s: str, lane: str) -> str:
self.pin[s] = lane
return lane
def gate(candidate: dict[str, tuple[float, int]], baseline: dict[str, float],
rules: dict[str, tuple[float, int]]) -> str:
"""灰度门禁：样本不够返回 hold，任一指标超阈值返回 rollback。"""
verdicts = set()
for metric, (max_drop, min_n) in rules.items():
if metric not in candidate:
continue
value, n = candidate[metric]
if n < min_n:
verdicts.add("hold")
continue
base = baseline.get(metric, 0.0)
drop = (base - value) / base if base > 0 else 0.0
verdicts.add("rollback" if drop > max_drop else "promote")
return "rollback" if "rollback" in verdicts else ("hold" if "hold" in verdicts else "promote")
这 70 行覆盖了本文最核心的三件事：
第一，Manifest 把"回滚的是什么"显式化了。 四个单元，每个都能单独 pin，每个都能单独回滚。这是 1.3 节那张表的代码形态。
第二，Pins.rollback() 实现了"回滚不靠重新构建"。 它只做一件事：把某个字段改回 history[-1] 的值。这正是把 40 分钟压缩到 30 秒的机制——不需要重新构建镜像、不需要重新部署、不需要改数据库，只需要改一个 pin。
第三，gate() 里"样本不够返回 hold"。 这一条是整套质量门禁的灵魂——"不知道"必须是一种合法状态，不能因为数据不够就默认通过。
如果你只能做一件事：把 Prompt 移出数据库，放进版本化的文件 + Pins 里。这一次改动就能把本次事故从 40 分钟降到 1 分钟以内，而它的成本只有半天工作量。
八、总结
核心只有一句：Agent 的 CI/CD 要先设计"回滚的是什么"，再设计"流水线怎么跑"。 那次 40 分钟的事故里，真正的回滚动作只花了 30 秒——其余 39 分半全花在"不知道该回滚什么"和"找不到上一个版本"上。
五个最该记住的结论：
第一，四个单元必须可独立版本化、独立回滚：代码、Prompt、模型、配置与工具集。而且回滚耗时要与变更频率成反比——Prompt 变更最频繁，它的回滚就应该最快。我们当时恰恰是反的。
第二，质量门禁的主力是 L2 离线评估，不是单测。 本次事故里 L0 全绿（代码没毛病），只有评估能发现"结论完整率下降 8.1 个百分点"。代价是 9 分钟和 1.8 美元，对比 40 分钟事故和 24.7 个百分点的业务损失，这个交换极其划算。
第三，灰度期的主判据必须是质量层指标，技术层指标只用来抓崩溃。 错误率、延迟、工具成功率在本次事故中全程正常——答非所问的服务在技术上"工作得很好"。
第四，停留时长由"最慢的主判据指标"决定，不是拍脑袋定。 单臂需要约 1150 个会话才能检出 5 个百分点的退化；在 1% 流量下这需要 38 小时。如果你的流量小到灰度无统计意义，就诚实地说"灰度对我不可用"，改用影子流量 + 离线评估。
第五，全量后必须把新版本提升为基线。 漏掉这一步，门禁比较的就永远是"新版 vs 远古版本"，整套机制会在几个月内慢慢失效——而且失效得很安静，没人会发现。
最后一点顺序上的建议：先有评测集，再有门禁，最后才有灰度。 很多团队想直接上灰度平台，但没有评测集就没有"质量基线"，没有基线门禁就无从判定"有没有退化"——这时灰度只是一套昂贵的流量切换装置，它挡不住本次事故这类问题。第 35、36 篇是本文的前置条件，不是可选项。
落地自检清单（Checklist）
- ✅ 有 四个可独立回滚的单元（代码 / Prompt / 模型 / 工具集）
- ✅ Prompt 在版本化文件里，不在数据库里；版本号 = 内容哈希
- ✅ 模型用具体 id，不用别名
- ✅ 有 发布清单（manifest），能回答"线上跑的是哪四个版本"
- ✅ L0 + L1 每次提交都跑（38 秒 + 2 分钟）
- ✅ Prompt/模型/工具集变更触发 L2 离线评估
- ✅ 门禁里有一票否决类指标（安全分，阈值 0.00）
- ✅ 灰度路由三维度叠加：开关 → 租户 → 流量哈希
- ✅ 灰度有会话粘滞，防止同一会话跨版本
- ✅ 门禁样本量不足返回 hold，不默认通过
- ✅ 灰度主判据是质量层指标，技术层只抓崩溃
- ✅ 每档停留时长由最慢主判据的样本量需求决定
- ✅ 全量后提升为新基线
- ✅ 回滚有防抖动（连续 N 次失败 + 冷却期），但安全指标立即回滚
- ✅ 每月演练一次回滚，且演练可还原
- ✅ 发布前检查幂等键跨版本有效、降级分支存在、超时参数一致
常见问题（FAQ）
Q1：我们团队只有 3 个人，需要全套吗？
不需要。 只做两件事：① 把 Prompt 放进版本化文件（半天工作量）；② 有一个能改回上一个 pin 的脚本（就是第七节的 Pins.rollback）。这两件事能把本次事故从 40 分钟降到 1 分钟以内。 五层门禁等你有了评测集再说——没有评测集就没法判质量，硬上灰度只是买了个流量切换器。
Q2：灰度用 1% 还是 5% 起步？
两个都做，但判据不同。 1% 档配技术层指标（10 分钟窗口，200 样本），只用来抓崩溃；5% 档配质量层指标，才是真正的质量关卡。不要指望 1% 档能判定质量——1% 流量下 15 分钟只有约 8 个样本。
Q3：离线评估跑一次 9 分钟，太慢了怎么办？
先确认它是否只在必要时跑。 如果连纯代码改动也在跑 L2，那就是配置问题（4.3 节的 changed 判断）。如果确实每次都要跑，两个方向：① 用更小的评测集做"快速门禁"（100 例，2 分钟），完整集只在发布候选跑；② 评测集并行化。不要把 9 分钟当成"跳过评估"的理由——那是把 40 分钟事故的概率买回来。
Q4：模型供应商悄悄更新了，我们没改任何东西却退化了，怎么办？
两件事：① 用具体 model id 而非别名，把"升级模型"变成一次显式的发布；② 每周跑一次基线评估（不是只在变更时跑），主动发现供应商侧的漂移。我们的第 7 次事故（改造后唯一一次漏网）就是这么补上的。
Q5：自动回滚会不会误杀？
会，所以用"连续 3 次失败才回滚 + 30 分钟冷却"来防抖动。 代价是真实事故最多多影响 2 分钟（判定周期 60 秒），这远小于频繁误回滚损失的发布窗口。例外是安全类指标——它应该立即回滚。
Q6：灰度期间用户会话跨版本怎么办？
默认用会话粘滞（首次路由后固定版本）。紧急回滚时，先置权重 0 阻止新会话进入，同时强制清除 pin 表——这时"止血"优先于"体验一致"。粘滞表必须有清理机制，否则内存里会堆积几十万条。
Q7：能力开关和灰度开关能共用一个吗？
不能，语义不同。 灰度要回答"新版有没有坏"，实验要回答"哪个版本更好"。混用会让两边的结论都不可信——实验的分流会污染灰度的对照，灰度的自动回滚会破坏实验的样本。这是 Martin Fowler 在 CanaryRelease 里明确警告过的一点。
Q8：发布变慢了（6 分钟 → 34 分钟），值得吗？
值得，但要分级。 34 分钟里的大头是灰度 1%/5% 档的 23 分钟。解法是分级通道：低风险改动（纯代码、有 L1 覆盖）走快通道（跳过 L3、灰度只跑 1% 档 5 分钟），高风险改动走完整流程。不要为了快而全量直发——那正是本次事故的做法。
Q9：影子流量和灰度选哪个？
不是二选一，是互补。 影子流量不暴露用户、能测分布漂移，但测不出真实业务结果、且要 stub 掉副作用；灰度用真实用户、能测业务结果，但要承担暴露风险。建议：L3 影子流量每次发布候选都跑（15 分钟、6 美元），灰度只做质量确认与崩溃检测。流量特别小的时候，影子流量是唯一有意义的选择。
参考资料
- Google SRE Workbook — Canarying Releases（灰度的规模、时长、流量与指标选择原则）：https://sre.google/workbook/canarying-releases
- Martin Fowler — bliki: Canary Release（金丝雀发布的定义、与蓝绿部署和 A/B 测试的区别）：https://martinfowler.com/bliki/CanaryRelease.html
- Martin Fowler / Pete Hodgson — Feature Toggles (aka Feature Flags)（开关分类与开关债）：https://martinfowler.com/articles/feature-toggles.html
- Argo Rollouts 官方文档 — Canary Deployment Strategy（
setWeight与pause步骤、分析模板）：https://argoproj.github.io/argo-rollouts/features/canary/ - OpenFeature — 厂商无关的开关标准（CNCF 孵化项目，避免开关实现锁定）：https://openfeature.dev/
- GitHub Docs — Workflow syntax for GitHub Actions（
on、jobs、concurrency、if表达式）：https://docs.github.com/en/actions/writing-workflows/workflow-syntax-for-github-actions - GitHub Docs — Using environments for deployment（环境的必需审批人、等待时间与分支限制）：https://docs.github.com/actions/deployment/targeting-different-environments/using-environments-for-deployment
- GitHub Docs — Using concurrency（并发组与
cancel-in-progress的行为）：https://docs.github.com/en/actions/using-jobs/using-concurrency
更多推荐
11
0- 0
已为社区贡献13条内容
所有评论(0)