前段时间遇到一个小问题，后来发现这是个挺常见的坑，顺手整理一篇笔记。
目录
- 前言
- 一、问题定义：零件都管好了，为什么还会翻车
- 二、整体方案：一张上位图、四个站、三条纪律
- 2.1 零件表：前 12 篇的产出挂在这条流水线的哪一站
- 2.2 发布六站：一次 LLM 应用发布要过哪几关
- 2.3 变更决策表：什么改动走多重的流程
- 2.4 逐条对开篇 4.2 的六条
- 2.5 三个取舍点
前言
第 12 篇《数据工程底座》结尾我写下一句：“数据底座是地基，LLMOps 是地基建起来之后，怎么把整栋楼管起来。粮仓的地基在本篇，楼怎么盖，在第 13 篇。”写那句话的时候我以为我懂 LLMOps——不就是给模型加个版本号、给 Prompt 加个管理面板吗。结果没过两周，现实就让我给这句话补了个注脚——补在会议室里，当着人的面，我答不上话。
那天下午我把网关里的 model 名从旧模型换成榜分更高的新版。一行配置，影子里看着挺好，放 5% 小流量。两个小时之后，第 7 篇那个 schema 闸开始隔三差五降级，有几类意图答非所问。回滚，恢复，前后二十分钟。
复盘的时候我卡在最简单的那一问上：我要回滚到哪一版？ 模型名我知道，改回去就行；可“当时那一版 Prompt 配的是哪个数据版本”——我答不上来。不是记不清，是从来没有一个地方把这三件记在一起过。那一刻我认了：我不是不会换模型，我是从来没有“一次发布”这个东西。LLMOps 要管的从来不是模型，是那一次可信发布的完整身份。
一、问题定义：零件都管好了，为什么还会翻车
前 12 篇把该修的零件全修了：第 2 篇有评测门、第 3 篇有灰度熔断、第 4 篇有 Prompt 内容哈希身份、第 5 篇有网关路由、第 6 篇有安全护栏、第 7 篇有 schema 闸、第 8 篇有全链路 Trace、第 9 篇有成本账、第 10 篇有反馈闭环、第 11 篇有换版纪律、第 12 篇有数据底座。可它们躺在各自的抽屉里，谁也管不了“把它们凑在一起的那一次发布”。我踩过三个误区，一个比一个贵。
第一，把 LLMOps 当“MLOps + 一个 Prompt 管理面板”。 这是最根深蒂固的一个：模型注册表、训练流水线、监控看板那套搬过来，再给 Prompt 加个版本号，齐活。可这套是模型为中心的旧框架，它的发布单位是一个模型权重。真出事的从来不是“单个零件坏了”，是“零件换了一个、你以为没事”——同一版 Prompt 换个模型不是同一件事，同一版模型换个数据版本也不是同一件事。
第二，把“换模型”当升级。 直觉上，新模型更强、榜分更高，换上去应该更好，改一行配置的事。可换模型同时改了四件事：Prompt 的效果（旧模型的“咒语”新模型可能不认）、schema 遵守率（第 7 篇的闸从“偶尔重试”变成“天天降级”）、成本结构（第 9 篇的账本整个重算）、缓存与上下文行为（第 9 篇的语义缓存前缀失效）。我只看了榜分。
第三，把微调当第一杠杆。 一提“模型全生命周期管理”，第一反应往往是上微调、上 GPU、上训练流水线，听着最硬核、最像“正经的 MLOps”。但它把一个“改一行 Prompt、跑一次评测门就能验证”的问题，变成了“攒数据 → 重训 → 重评 → 重发 → 回滚更难”的问题，而且这是它最贵的地方：Prompt 回滚是切指针，模型回滚是切权重再重测所有下游。
先把结论撂这儿：LLMOps = 把 Prompt / 数据 / 模型三条线的变更收进同一条流水线，发布的原子单位不是“一个模型版本”，是“一组被一起评测过、被锁死的版本三元组”；管好单个零件，不等于管好一次发布。 这一篇不打算讲“MLOps 全生命周期科普”（那是训练→部署→监控的旧框架，发布原子没变），也不横向比 30 个 LLMOps 工具（工具清单放附录 C）。我要讲的只有一件：一堆已经造好的零件，怎么串成一条能跑、能回滚、能复现的发布流水线。
先把那个自造的词说圆，免得你去搜一个搜不到的东西：版本三元组 = Prompt 内容哈希 × 数据版本号 × 模型 ID，锁成一组；这三样拼起来再算一个指纹，那串指纹就是“那一次发布”的身份。 后面我全程只叫它“版本三元组”，不换词。
二、整体方案：一张上位图、四个站、三条纪律
开篇《四大支柱》里 4.2 那张“MLOps/LLMOps 模型全生命周期管理”上位图，给的就是本篇的主框架——六条：第 1 条模型注册与版本管理、第 2 条标准化训练/微调流水线、第 3 条模型 A/B 对比测试、第 4 条模型性能衰退自动监控（Data Drift / Response Drift）、第 5 条 GPU 算力弹性调度、第 6 条模型退役规范。同时 4.2 开头那句“传统 MLOps 聚焦训练-部署-监控……LLMOps 还需额外管理 Prompt 版本、非确定性输出质量、在线人类反馈、Token 成本”，这一篇就是接着它往下写的（这四个分项各有公开出处；整句话是本专栏对多家口径的归纳，不是逐字引用）。
整篇的主线就一句：前 12 篇已经把零件造好了——零件都在，可谁也没给它们排过队。LLMOps 就是那张把它们串起来的图：三条线（Prompt / 数据 / 模型）的变更从同一个入口进来，锁成一个不可变的版本三元组，过评测门、过发布门，被 Trace 和成本账盯着，被反馈闭环推着，被退化监控和退役扫描兜着——转一圈回到变更入口，这就是飞轮。
零件层（前 12 篇造好、各自独立的零件）
Prompt 身份（第 4 篇：content_hash 当身份，semver 只是标签）
数据版本（第 12 篇：ReleaseBundle V_k -> V_{k+1}，不可变）
模型（第 5 篇网关：多模型路由 + fallback；本篇补：模型注册与版本）
常驻件：护栏（第 6）、schema 闸（第 7）、Trace（第 8）、成本账（第 9）
│
▼ 锁成一个不可变的「版本三元组」ReleaseTriple(prompt_id, data_version, model_id)
│ —— 三件各自有身份，凑一起再算一个三元组指纹，这个指纹才是「那次发布」
▼
变更入口（三条线都从这一个口进）
│ 单一变更原则：一次只动一件，三元组只在一个维度上 +1（两维同改 = 出事了不知道谁的锅）
▼
① 注册：零件版本登记（模型进 registry、Prompt 进 registry、数据出新版本快照）
▼
② 评测门（第 2 篇）：候选三元组过 Golden Set -> 不过门不许出 staging
▼
③ 发布门（第 3 篇）：影子 -> 小流量金丝雀 -> 分层放量；任一阶段劣化 -> 自动回滚（指针指回旧三元组，零重建）
▼
④ 观测（第 8 篇 Trace + 第 9 篇成本账）：发布后盯着，trace 按三元组指纹打标
▼
⑤ 反馈（第 10 篇 HITL）：差评/点踩 -> 三段筛选仲裁 -> 回灌 golden set -> 触发下一轮评测
▼
⑥ 退化监控与退役（本篇新增）：Data Drift / Response Drift 超阈值 -> 告警 -> 回变更入口；
供应商下线旧模型 -> 扫出「还钉着它的三元组」-> 待迁移清单（对应开篇 4.2 的第 4 条衰退监控、第 6 条退役规范）
│
└──────────► 回到变更入口：一圈 = 飞轮（= Google Cloud Agent Quality Flywheel 的三阶段）
那条环形主线不是我自己编的。开篇支柱四那张 4.2 飞轮图的三阶段 Build & Test → Ship & Monitor → Learn & Refine，出处是 Google Cloud 的 Agent Quality Flywheel（2026，Cloud Next '26 的 BRK3-023 场次 + Google Developers Blog 成文），跟经典那本《Practitioners Guide to MLOps》白皮书没关系——开篇里把它写成“MLOps 实践框架”，这里更正一下。这个飞轮里有一条设计原则我特别喜欢，本篇的一条纪律就是从它这儿来的：“提出修复的永远不给自己打分”——防止 metric-gaming，打分交给独立评测服务。翻译成我们的话：谁改的谁别自己判过没过，判过没过是评测门的事。
2.1 零件表：前 12 篇的产出挂在这条流水线的哪一站
先说清楚一件事：这篇没有新零件，全是把旧零件串起来。 下面这张表你可以直接拿去对着自己的代码库点一遍——哪一站的零件你有了、哪一站还空着。
前篇产出是什么零件挂在流水线哪一站本篇给它补的一句话第 4 篇 Prompt registry（内容哈希身份）三元组第 1 维变更入口 → ① 注册身份从 Prompt 扩到三件，凑一起再算三元组指纹第 12 篇数据底座 ReleaseBundle三元组第 2 维变更入口 → ① 注册数据版本进三元组，改数据 = 出新的发布第 5 篇网关（模型路由/fallback）三元组第 3 维变更入口 → ① 注册网关管路由，管不了“这模型配这 Prompt 能不能上”第 2 篇 Golden Set评测门② 评测门评测对象从“一个 Prompt”改成“一个三元组”第 3 篇灰度 + 熔断发布门③ 发布门回滚的粒度从“服务版本”改成“三元组指针”第 8 篇全链路 Trace观测④ 观测trace 按三元组指纹打标第 9 篇成本六维成本账④ 观测换一维就要重算一遍成本账第 10 篇反馈闭环飞轮的推力⑤ 反馈反馈挂 trace_id + 三元组指纹双标第 11 篇换版纪律不可变纪律⑥ 退化监控与退役“不能 UPDATE 只能换版本”升格为“三件都不能原地改”第 6 篇护栏 / 第 7 篇 schema 闸常驻件全程换模型/换 Prompt 要重过护栏与遵守率；护栏版本化但先不塞三元组
2.2 发布六站：一次 LLM 应用发布要过哪几关
这张表是本篇最实用的一张，你可以直接抄成自己的建设清单。
站回答的问题拦的是什么 / 兜的是什么代价 / 留意① 注册这三件各自是哪一版？拦“无身份的零件”——没进 registry 的模型、没版本号的 Prompt、没快照的数据，一律锁不进三元组注册表要跟发布账本分开（一个管“有哪些版本”，一个管“哪一版被发过”）② 评测门这一组凑一起，行不行？拦“没被一起评测过的组合”——强模型 + 老 Prompt 也可能回归每加一个零件版本就多一组要评测的对象，组合会膨胀，靠单一变更原则收敛③ 发布门放量放得安全吗？拦“放量中劣化还硬推”——影子/金丝雀/放量任一阶段劣化即回滚回滚必须是指针级的（指回旧三元组），不是重跑流程④ 观测上线后在往哪走？拦“静默退化”——不盯着，劣化要等客诉才发现trace/成本都要按三元组打标，否则出事了拼不出图⑤ 反馈线上真实反馈怎么用？拦“把差评直接当结论”——要过分流/筛选/仲裁才回灌反馈是飞轮的推力，但它是“弱信号”，不仲裁不许进 golden set⑥ 退化监控与退役不换也会坏吗？兜“数据漂移 / 响应漂移 / 供应商下线”——不发布也会退化监控只告警不自动改，改还是走变更入口（自动改 = 绕开评测门）
挑几个站展开说：
① 注册：模型进 registry 要记四件标签。 Prompt 用第 4 篇的内容哈希定身份、数据用第 12 篇的 ReleaseBundle 快照定版本——这两件前 12 篇都讲过了，本篇补的是“模型”这一维。模型进 registry 不用记太多，但四件标签跑不掉：能力（榜分/尺寸/上下文窗口）、schema 遵守率（第 7 篇那个闸的实测值）、成本（每千 token 单价，第 9 篇要用）、合规（能不能出境、数据驻留要求）。这四栏不是凑数：换模型至少要重估的正好就是这四类，对得上 registry 该记的字段——别省。真实工具里 MLflow Model Registry 3.16.1 的 LoggedModel 就是干这个的，把模型↔代码版本↔Prompt↔评测链接起来（版本与状态见附录 C）。
② 评测门：第 2 篇的 harness 一行不改，只换对象。 这是本篇最容易漏、也最反直觉的一条——评测对象不是“一个 Prompt 版本”，是“一个三元组”。同一版 Prompt 换个模型、换个数据版本，评测结论不作数，得重跑。于是第 2 篇那套 Golden Set + CI 自动化评测，在本篇原样调用，只把输入从 Prompt 换成三元组。评测过了，才把这个三元组指纹写进“已验证组合表”；没进这张表的，版本号再新也不许上 serving。
③ 发布门：第 3 篇灰度也一行不改地复用，只换回滚粒度。 影子 → 小流量金丝雀 → 分层放量 → 秒级回滚，这套第 3 篇讲透了。本篇只说一件事：第 3 篇回滚的是服务，要回滚得动，得先有本篇的“三元组指针”——指针指回旧三元组，零重建。不然你回滚了服务版本，底层的数据版本和 Prompt 版本还停在新的一版上。
⑥ 退化监控与退役：本篇真正新增的一段。 就两件事，先说漂移监控：Data Drift 是输入分布 P(X) 漂了（用户话题、问法、语言风格变了），Concept Drift 是 P(Y|X) 漂了（输入长得差不多，但“正确答案”变了——举个例子新法规让原来对的答案变错），Response Drift 是 LLM 特有的一类：供应商静默换了权重 / 安全过滤器 / 系统 prompt，导致输出格式、语气、啰嗦度、拒答率变了——这是“换模型 API 端”最隐蔽的一类，你什么都没改，它也漂。再说退役：供应商公告下线旧模型，你得能一次扫出“线上还有哪些三元组钉着它”，出一张待迁移清单。监控这条我给死一句：只告警，不自动改。 自动改 = 绕开评测门，等于把门拆了。
2.3 变更决策表：什么改动走多重的流程
变更类型走几站为什么代价 / 留意改一句 Prompt 措辞① ② ③ ④（全套）Prompt 字节敏感，一改就可能变行为；内容哈希变了就是新三元组最频繁的一类变更，最容易被“就改一句话”漏掉门换知识库数据版本（第 12 篇 V_{k+1}）① ② ③ ④（全套）数据换了，同一版 Prompt 的检索上下文全变第 11 篇巡检告警常是它的触发源换模型（供应商 / 版本 / 微调权重）① ② ③ ④（全套，且更重）同模型换版本 = 换了一套行为分布，Prompt 效果/schema 遵守率/成本/缓存全变最强反直觉点：越强的模型可能越差；换模型走发布，不走升级改路由 / 温度 / 采样参数① ② ③（可只跑最小回归）不换零件、只换调用方式，风险面比换零件小，但仍是行为变更温度/路由不是三元组的维度，不进账本、不产生新指纹，但它改了输出分布，仍要过最小回归 + 记一条配置变更账；别把它当“配置”绕过门高频小批新料入库（第 12 篇在线增量）不走发布（走增量）只 append、不动全库、不改 Prompt/模型对应第 12 篇在线增量链路：新料不该触发一次全量发布；但增量漂移要靠定期对齐收敛，对齐出 V_{k+1} 时再走一次完整发布，否则线上三元组的 data_version 会慢慢说谎换护栏 / 换 schema① ② ③（护栏不算三元组，但要重跑对抗回归）安全防护效果跟模型和 Prompt 强相关边界：护栏/闸先版本化、独立发布，别塞进三元组免得组合爆炸
2.4 逐条对开篇 4.2 的六条
第 1 条注册、第 3 条 A/B、第 4 条衰退监控、第 6 条退役，这四条落进主线了（第 3 条 A/B 落在 ③ 发布门，第 3 篇的影子/灰度就是它的落地）。第 2 条标准化训练/微调流水线，落成下一节那个取舍——“微调是最终一颗子弹”。第 5 条 GPU 算力弹性调度，这块我没踩过，只给边界：vLLM 0.26.0 / SGLang 0.5.16 / TensorRT-LLM 1.2.1 这些推理引擎解决的是“推理成本/吞吐/显存”的问题，跟发布纪律是两件事，别混谈——你不会因为换了推理引擎就要重跑评测门（除非它改了输出分布），也不会因为发布流水线搭好了就自动省钱。“不完全指南”这四个字不是谦虚：GPU 调度、分布式训练、RLHF 全流程，这三块我没真踩过，只能指个方向，不硬编。
2.5 三个取舍点
取舍一：LLMOps 的发布原子不是“模型版本”，是“版本三元组”。
一提 LLMOps，大多数人的第一反应是“MLOps 那套 + 一个 Prompt 管理面板”。但 LLM 应用真正翻车的地方从来不在“单个零件坏了”，在“零件换了一个、你以为没事”。MLOps 的发布单位是一个模型权重（换权重 = 一次发布），LLMOps 的发布单位是三个零件的一个组合（换任何一件 = 一次发布）。 我的做法：三件各自用内容哈希定身份，凑一起再算一个三元组指纹；只有出现在“已验证组合表”里的指纹才允许上 serving。代价我认——组合会膨胀。Prompt 10 版 × 数据 8 版 × 模型 5 版 = 400 组，全测不现实（这个 400 是我拿自己场景算的估算，不是哪个权威口径，见附录 B）；得靠“单一变更原则 + 只锁被评测过的组合 + 定期清理过期组合”收敛。但省这笔钱的代价更贵：你回滚得了模型名，回滚不了“当时那一版 Prompt 配的是哪个数据版本”。
取舍二：换模型不是升级，是发布——越强的模型不一定让你的应用更好。
这是我最想掰的一个认知。榜分更高 ≠ 你的应用更好，这不是我一个人的体感，是公开的数据：GSM1k（NeurIPS 2024）把 GSM8k 换题重出，模型在两张榜之间最高掉 8 个百分点，而且“模型生成 GSM8k 样例的可能性”与“过拟合差”的相关是 0.36——分数里有见过题的水分；Goodhart 通道那篇（arXiv 2609.00064）把模型往一个代理指标上微调，代理指标逼近天花板的同时 MMLU 从 0.371 掉到 0.279。schema 遵守率也一样，SO-Bench（CVPR 2026）里 GPT-4o 指令跟随 92.36% 对结构化 API 99.67%，Claude-3.5-Sonnet 无约束约 74.5%，GPT-4-0613 硬约不到 40%——遵守率不是模型的固定属性，它随“用哪个模型 + 用不用结构化输出 API + schema 多深”一起变。于是模型这一维是唯一“越强越贵、越换越险”的零件，它得走完整发布流程（重跑 golden set → 影子 → 灰度 → 熔断），不能走“改个配置名”的升级流程。代价是慢、贵、要维护一段多模型并行期；但“它榜分高”和“我的应用变好了”中间隔着一次评测，省掉那次评测，就是拿线上流量赌。
取舍三：微调是最终一颗子弹，不是第一杠杆——先把 Prompt、数据、路由榨干，再谈重训。
对绝大多数团队，微调是 ROI 最低的第一步：它把“改一行 Prompt、跑一次评测门就能验证”的问题，变成“攒数据 → 重训 → 重评 → 重发 → 回滚更难”，而且它最难回滚（Prompt 回滚是切个指针，模型回滚是切权重 + 重测所有下游）。量级上，LoRA/QLoRA 一次实验大概 5 到 500 美元（7B LoRA 单卡 RTX 4090 约 1.36 到 2.72 美元、70B LoRA 单卡 H100 约 24 到 48 美元、QLoRA 13B 单卡 L4 约 3.52 到 7.04 美元），full fine-tuning 是 100 到 10000 美元往上（口径见附录 B）——是“几十到几百美元”级，不是“训一个大模型”级，但对大多数团队仍是“最贵的一步”。2025 年那几份共识给的都是同一条顺序：Prompt → Few-shot → RAG → LoRA/QLoRA → Full FT，微调排倒数第二，跟第 9 篇那套“先缓存 → 后压缩 → 再小模型 → 最终蒸馏”是同一个逻辑：便宜的杠杆先用尽。我给微调留了个明确的“上场判据”：① Prompt/数据/路由都试过且到了天花板；② 问题是“风格/领域术语/格式稳定性”这类靠 Prompt 说不清的；③ 第 10 篇的反馈闭环已经攒出一批干净、仲裁过的训练样本（没有这批样本就谈微调，等于拿噪声训模型）。代价是：有些问题确实只有微调能解，拖到最后照样要吃这一口，而且拖久了团队会催“怎么老不上微调”。我认，因为顺序错了的代价不对称——先微调后 Prompt，花的钱收不回来；先 Prompt 后微调，最多晚一点，每一步都可回滚。
三、代码实战：一个可离线跑的最小发布流水线
先说清楚 demo 为什么自包含：第 2/3/4/5/8/9/10/11/12 篇的 demo 都嵌在各自这篇里，磁盘上没有那些文件，mini_llmops.py 不 import 任何不存在的东西。LLMOps 这条线最容易写成纯概念（满篇“生命周期”“飞轮”“治理”），可它真正吃劲的那几件事——零件定身份、锁三元组、单一变更校验、评测门、灰度回滚指针、漂移告警、退役扫描、发布账本——全是纯数据加纯判断的活，唯一碰外部世界的是“真实评测 / 真实灰度平台 / 真实监控 / 真实模型注册中心”，全都藏在可替换的接口后。demo 只用标准库：发布账本用 sqlite3，零件身份和三元组指纹用 hashlib，评测门用一张确定的 mock 评分表（离线、可复现、能断言）。不 import openai/requests/langchain 任何一个，不调 API、不花钱、离线可跑。
mock 数据刻意埋了活靶子：两版 Prompt、两版数据、两个模型——里面新模型“榜分更高但应用侧 schema 遵守率更差”。release/mini_llmops.py：
"""
迷你 LLMOps 发布流水线：把前 12 篇拆散的零件（Prompt 身份 / 数据版本 / 模型路由）
抽象成三个零件指针，跑通 注册零件 -> 锁版本三元组 -> 单一变更校验 -> 评测门 ->
灰度发布 + 指针级自动回滚 -> 退化监控 -> 模型退役扫描 -> 审计账本 一条线。
纯标准库、无第三方依赖、不调 API、离线可跑（发布账本用标准库 sqlite3，
零件身份与三元组指纹用标准库 hashlib，评测门用一张确定性 mock 评分表）。
依赖安装：无（Python >= 3.10，只用到标准库 sqlite3 / hashlib / json / datetime）
运行：python mini_llmops.py / python test_mini_llmops.py
（Windows 控制台打印中文若报 GBK 编码错：PowerShell 用 $env:PYTHONUTF8=1; python mini_llmops.py，
cmd 用 set PYTHONUTF8=1 && python mini_llmops.py）
主线一句话：LLMOps 的发布原子不是"一个模型版本"，是"一组被一起评测过、被锁死的版本三元组"
（Prompt 版本 x 数据版本 x 模型版本）。三条线（Prompt / 数据 / 模型）的变更从同一个入口进来，
一次只许动一维（single_change_from），锁成指纹（ReleaseTriple.fingerprint），
过评测门（eval_gate，没进 verified 表不许上 serving），过发布门（canary，劣化即指针级回滚），
被退化监控盯着（drift_check，只告警不自动改），被退役扫描兜着（retire_model，一次 SELECT 出清单），
每一次动作落审计账本（audit 表）。
七个"为什么"是整份代码的骨架：
1. 为什么三元组换一维指纹必变：三件里任何一件的内容变一个字节，身份就该变——不然你会拿
"旧 Prompt + 新模型"当成"同一次发布"，出事时说不清是哪三件。（ReleaseTriple.fingerprint）
2. 为什么两维同改要被拒：一次动两件，指标掉了归因不到任何一维。single_change_from()
返回变化的维度，长度 >1 直接拒收、要求拆成两次发布。（对应踩坑②）
3. 为什么没进 verified 表的三元组 set_serving 要抛错：没被一起评测过的组合，版本号再新也不许发——
评测门是准入不是仪式。（ReleaseStore.set_serving）
4. 为什么 MockScorer 要能造出"更强的模型反而 schema 遵守率掉、被评测门拦"：榜单能力与应用侧
golden set 是两回事（GSM1k 榜间掉 8 点、Goodhart 通道 MMLU 0.371->0.279 是公开背书）；
换模型不是升级是发布，得走门。（build_mock_world 里的评分表）
5. 为什么 canary 劣化要指针级回滚、零重建：回滚 = 把 serving 指针指回旧三元组，不是重跑流程。
（canary -> ReleaseStore.rollback）
6. 为什么 drift_check 只告警不自动改：自动改 = 绕开评测门，等于把门拆了。漂移超阈值只发告警 +
建议走变更入口，改还是得从入口进。（drift_check）
7. 为什么退役扫描必须是一次 SELECT：模型是耗材不是资产，退役是必然事件，它得能一次查出来，
不是翻配置文件考古。（retire_model）
demo 的"三元组换一维指纹就变 / 拦下一个榜分更高的模型 / 两维同改被拒 / 灰度第二阶段劣化自动回滚 /
漂移告警 / 退役扫出 2 个"是对 mock 数据的确定性结论，不是对真实发布流水线的承诺——真实门禁阈值
取决于你的业务 SLA（量级锚点见文章附录 B）。真实生产把 PartRegistry 换成 MLflow / HF Hub 管模型
+ Langfuse 管 Prompt + DVC 管数据，把 eval_gate 换成第 2 篇 harness、canary 换成第 3 篇灰度、
drift_check 换成监控告警系统，流水线逻辑一行不改（同第 8/12 篇"生产换 SDK、骨架一行不动"的替换心智）。
"""
import datetime
import hashlib
import json
import sqlite3
from dataclasses import dataclass, field
def _now_iso():
return datetime.datetime.now(datetime.timezone.utc).isoformat()
# ---------- 数据契约 ----------
@dataclass
class PartVersion:
"""一个零件版本：kind（prompt / data / model）、version_id、content_hash、meta。
为什么身份是 content_hash 不是 semver：第 4 篇那条纪律——内容哈希才是身份，semver 只是标签。
标签可以不变而内容被人悄悄改了，哈希不会。三件里任何一件的内容变一个字节，身份就该跟着变，
才有资格进一个新的三元组。"""
kind: str
version_id: str
content_hash: str
meta: dict = field(default_factory=dict)
@dataclass(frozen=True)
class ReleaseTriple:
"""版本三元组：prompt_id x data_version x model_id，加一个 fingerprint()。
为什么发布原子是它、不是"一个模型版本"：单看任何一件的版本号都没意义，只有"被一起评测过的
那一组"才有意义。换一维，指纹必变——这就是"另一次发布"。同一版 Prompt 换个模型、换个数据版本，
都不是同一件事，评测结论也不作数。"""
prompt_id: str
data_version: str
model_id: str
def fingerprint(self) -> str:
raw = f"{self.prompt_id}|{self.data_version}|{self.model_id}"
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
@dataclass
class RollbackRecord:
"""一次回滚：from_triple 是劣化的那个、to_triple 是指针要指回的那个。
为什么回滚要落成一条记录、而不是只把指针拨回去：历史里要有"回滚"这个动作本身——
谁在什么时候把指针从谁指回了谁、为什么，必须查得到（第 4 篇 rollback 心智的上位版）。
to_triple 允许为 None：首次发布就劣化时没有可指回的旧三元组，账上落 NULL 而不是空串。"""
from_triple: str
to_triple: str | None = None
reason: str = ""
operator: str = "auto-canary"
at: str = ""
# ---------- 零件注册表：三件各自定身份 ----------
class PartRegistry:
"""零件注册表：三件（prompt / data / model）同表不同 kind。
为什么注册表和发布账本要分开：一个管"有哪些版本"，一个管"哪一版被发过"。混在一起，
就会把"登记过"当成"发布过"——登记只是这组三件存在，不等于它被一起评测过、被发过。
（对应正文的"注册表要跟发布账本分开"。）"""
def __init__(self):
self._parts = {}
def put(self, kind, content, meta=None) -> str:
"""按内容哈希定 version_id——同一份内容反复 put 得到同一个 id（幂等、可复现）。
为什么幂等很重要：CI 每天跑一遍注册，不该因为跑了几次就多出几个版本号。
为什么空内容要拒：None 和 "" 会折叠成同一个哈希，两个语义不同的输入拿到同一个身份，
正好是"内容哈希才是身份"最想避免的反例。"""
if kind not in ("prompt", "data", "model"):
raise ValueError(f"未知零件类型 {kind!r}：只认 prompt / data / model")
if content in (None, ""):
raise ValueError(f"{kind} 零件内容不能为空：空内容定不出身份")
h = hashlib.sha256(content.encode("utf-8")).hexdigest()
version_id = f"{kind}-{h[:12]}"
self._parts[(kind, version_id)] = PartVersion(
kind=kind, version_id=version_id, content_hash=h, meta=dict(meta or {}))
return version_id
def get(self, kind, version_id):
return self._parts.get((kind, version_id))
def list(self, kind):
return [pv for (k, _), pv in sorted(self._parts.items()) if k == kind]
# ---------- 变更入口：单一变更原则 ----------
def single_change_from(current, candidate) -> list:
"""单一变更原则：返回 current -> candidate 变化的维度列表。
为什么长度 >1 就是违规：一次动两件，指标掉了归因不到任何一维——又调 Prompt 又换数据版本，
你只能花半天在两者之间二选一，最后发现两个都有问题还互相掩盖。
为什么返回列表而不是布尔：调用方要能告诉用户"你动了哪几维、该拆成几次"。"""
changed = []
for dim, a, b in (("prompt", current.prompt_id, candidate.prompt_id),
("data", current.data_version, candidate.data_version),
("model", current.model_id, candidate.model_id)):
if a != b:
changed.append(dim)
return changed
# ---------- 发布账本：sqlite3 三张表 ----------
class ReleaseStore:
"""发布账本：sqlite3 三张表，把"这一次发布是哪三件、被一起评测过没有、动过几次"钉死。
- triples：三元组指纹 + 三件 id + 状态（staging / serving / retained / rolled_back）。
指纹当主键——同一组三件只可能有一行，这就是"这次发布"的身份。
- verified：已验证组合表。只有评测门跑过、lock() 过的指纹才进这张表。set_serving()
只认这张表——没被一起评测过的组合，版本号再新也不许上 serving。
- audit：每次 publish / rollback / lock 落一行（时间、动作、从哪个三元组到哪个、
原因、操作人）。回滚不是"指针拨回去"就完了，账上得留下"回滚"这个动作本身。
为什么三张表用 sqlite3 而不是真数据库：demo 要离线可跑、能断言。生产就是
"MLflow / HF Hub 管模型 + Langfuse 管 Prompt + DVC 管数据 + 一张发布账本"，
纪律一条不改（附录 C）。"""
def __init__(self, path=":memory:"):
self.conn = sqlite3.connect(path)
self.conn.row_factory = sqlite3.Row
self._init_schema()
def _init_schema(self):
self.conn.executescript(
"""
CREATE TABLE IF NOT EXISTS triples(
triple_id TEXT PRIMARY KEY NOT NULL,
prompt_id TEXT NOT NULL,
data_version TEXT NOT NULL,
model_id TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'staging',
created_at TEXT
);
CREATE TABLE IF NOT EXISTS verified(
triple_id TEXT PRIMARY KEY NOT NULL,
verified_at TEXT,
report TEXT
);
CREATE TABLE IF NOT EXISTS audit(
id INTEGER PRIMARY KEY AUTOINCREMENT,
at TEXT,
action TEXT,
from_triple TEXT,
to_triple TEXT,
reason TEXT,
operator TEXT
);
"""
)
self.conn.commit()
def _audit(self, action, frm, to, reason, operator):
self.conn.execute(
"INSERT INTO audit(at, action, from_triple, to_triple, reason, operator)"
" VALUES (?,?,?,?,?,?)", (_now_iso(), action, frm, to, reason, operator))
self.conn.commit()
# ---- 三元组表 ----
def register_triple(self, triple, state="staging") -> str:
"""把三元组登记进账本（staging）。登记只是"这组三件存在"，不等于"被发过"——
上 serving 还得过 verified 那张表。INSERT OR IGNORE 让重复登记幂等，
但不许把已有状态冲掉（一个已经 serving 的三元组不该因为重跑注册就掉回 staging）。"""
tid = triple.fingerprint()
self.conn.execute(
"INSERT OR IGNORE INTO triples"
"(triple_id, prompt_id, data_version, model_id, state, created_at) VALUES (?,?,?,?,?,?)",
(tid, triple.prompt_id, triple.data_version, triple.model_id, state, _now_iso()))
self.conn.commit()
return tid
def list_triples(self):
return [dict(r) for r in self.conn.execute(
"SELECT * FROM triples ORDER BY created_at, triple_id")]
def triple_of(self, triple_id):
row = self.conn.execute("SELECT * FROM triples WHERE triple_id=?", (triple_id,)).fetchone()
if row is None:
return None
return ReleaseTriple(row["prompt_id"], row["data_version"], row["model_id"])
# ---- 已验证组合表：上 serving 的唯一准入 ----
def lock(self, triple, report=None) -> str:
"""评测门通过之后，把三元组指纹登记进已验证组合表——这是"这一组被一起评测过"的唯一凭证。
为什么 lock 要单独一张表、而不是在 triples 上打个标：verified 是准入名单，triples 是流水账；
混在一起，一个误操作就能把"没评过的"标成"评过的"，守门员就没了。"""
tid = self.register_triple(triple)
self.conn.execute(
"INSERT OR REPLACE INTO verified(triple_id, verified_at, report) VALUES (?,?,?)",
(tid, _now_iso(), json.dumps(report or {}, ensure_ascii=False)))
self.conn.commit()
self._audit("lock", None, tid, "评测门通过：登记为已验证组合", "eval-gate")
return tid
def is_verified(self, triple_id) -> bool:
row = self.conn.execute("SELECT 1 FROM verified WHERE triple_id=?", (triple_id,)).fetchone()
return row is not None
def list_verified(self):
return [dict(r) for r in self.conn.execute(
"SELECT triple_id, verified_at FROM verified ORDER BY verified_at, triple_id")]
# ---- 发布指针 ----
def set_serving(self, triple_id, reason="发布门通过，指针切到新三元组", operator="release-bot"):
"""把 serving 指针切到某个三元组：旧 serving 降为 retained，只读保留、可回滚。
为什么没进 verified 表就抛错：评测门是准入不是仪式——没被一起评测过的组合，
版本号再新也不许发。这条硬约束是整条流水线的守门员，也是 demo 断言③的落点。
为什么旧 serving 不删：回滚 = 指针指回旧版，零重建（第 3 篇回滚粒度在本篇的上位化）。"""
if not triple_id:
raise ValueError("triple_id 不能为空：这压根不是个三元组 id")
if not self.is_verified(triple_id):
raise ValueError(
f"三元组 {triple_id} 没进已验证组合表：没被一起评测过的组合不许上 serving")
if self.conn.execute("SELECT 1 FROM triples WHERE triple_id=?", (triple_id,)).fetchone() is None:
raise KeyError(f"三元组 {triple_id} 没登记过：先 register_triple 再谈发布")
old = self.current_serving()
if old and old != triple_id:
self.conn.execute("UPDATE triples SET state='retained' WHERE triple_id=?", (old,))
self.conn.execute("UPDATE triples SET state='serving' WHERE triple_id=?", (triple_id,))
self.conn.commit()
self._audit("publish", old, triple_id, reason, operator)
return old
def current_serving(self):
row = self.conn.execute(
"SELECT triple_id FROM triples WHERE state='serving'"
" ORDER BY created_at DESC, triple_id LIMIT 1").fetchone()
return row["triple_id"] if row else None
def rollback(self, record: RollbackRecord):
"""指针级回滚：把 serving 指回旧三元组，零重建。
为什么回滚粒度是指针不是流程：重跑一遍发布流程是分钟到小时级，指针切回去是毫秒级；
第 3 篇回滚的是服务，要回滚得动，得先有这个"三元组指针"——不然你回滚了服务版本，
底层的数据版本和 Prompt 版本还停在新的一版上。
为什么回滚前要先校验端点存在：不然回滚到不存在的三元组会静默成功，账本里记一条"假回滚"
——对一个以"审计可信"为卖点的模块是硬伤。"""
if not record.at:
record.at = _now_iso()
from_id = record.from_triple or None
to_id = record.to_triple or None
if from_id and self.triple_of(from_id) is None:
raise KeyError(f"三元组 {from_id} 没登记过，回滚无从谈起")
if to_id and self.triple_of(to_id) is None:
raise KeyError(f"三元组 {to_id} 没登记过，指针无从指回")
# 先降级当前 serving（若不在回滚目标里），避免出现两行 state='serving'
cur = self.current_serving()
if cur and cur != to_id:
self.conn.execute("UPDATE triples SET state='retained' WHERE triple_id=?", (cur,))
if from_id:
self.conn.execute("UPDATE triples SET state='rolled_back' WHERE triple_id=?",
(from_id,))
if to_id:
self.conn.execute("UPDATE triples SET state='serving' WHERE triple_id=?",
(to_id,))
self.conn.commit()
self._audit("rollback", from_id, to_id, record.reason, record.operator)
record.from_triple = from_id
record.to_triple = to_id
return record
# ---- 审计 ----
def list_audit(self):
return [dict(r) for r in self.conn.execute("SELECT * FROM audit ORDER BY id")]
# ---- 退役扫描 ----
def triples_pinning(self, model_id):
"""按 model_id 一次扫出所有钉着它的三元组，serving 排最前——退役清单就靠这一条 SELECT。
为什么不做成"遍历配置文件"：模型 ID 散在配置里，退役那天就成了考古；进了账本，
退役就是一次查询。"""
rows = self.conn.execute(
"SELECT * FROM triples WHERE model_id=?"
" ORDER BY CASE state WHEN 'serving' THEN 0 WHEN 'retained' THEN 1 ELSE 2 END, created_at",
(model_id,)).fetchall()
return [dict(r) for r in rows]
# ---------- 评测门：确定性 mock 评分桩 ----------
class MockScorer:
"""确定性评分桩：按 (model_id, prompt_id, case_id) 查一张显式评分表。
为什么要一张"显式表"而不是随机数：demo 要能确定性地演示"榜分更高的模型在这套应用里
schema 遵守率掉、被门拦下"，随机数做不到"每次都拦得住"。
为什么索引要带上 prompt_id：同一版 Prompt 换个模型不是同一件事——只按 model_id 索引，
就锁不住"旧模型配新 Prompt"和"新模型配旧 Prompt"是两种不同的行为分布。
表里没配的组合走模型默认画像，保证任何三元组都跑得完。"""
def __init__(self, table=None, profiles=None):
self.table = dict(table or {})
self.profiles = dict(profiles or {})
self._fallback = {"passed": True, "schema_ok": True, "cost": 0.001}
def score(self, triple, case_id) -> dict:
key = (triple.model_id, triple.prompt_id, case_id)
if key in self.table:
return dict(self.table[key])
return dict(self.profiles.get(triple.model_id, self._fallback))
def evaluate(self, triple, cases) -> dict:
"""逐用例打分 + 汇总（通过率 / schema 遵守率 / 成本估算）。
为什么成本也进汇总：换模型同时改了成本结构——一个更强的模型配上更啰嗦的输出，
账单能翻倍（第 9 篇成本账本整个重算）。成本不拦发布，但要看得见。"""
rows = []
for cid in cases:
r = self.score(triple, cid)
rows.append({"case_id": cid, "passed": bool(r.get("passed", True)),
"schema_ok": bool(r.get("schema_ok", True)),
"cost": float(r.get("cost", 0.0))})
n = len(rows)
n_pass = sum(1 for r in rows if r["passed"])
n_schema = sum(1 for r in rows if r["schema_ok"])
return {
"triple_id": triple.fingerprint(), "model_id": triple.model_id,
"prompt_id": triple.prompt_id, "data_version": triple.data_version,
"cases": rows, "total": n, "passed_cases": n_pass,
"pass_rate": round(n_pass / n, 4) if n else 0.0,
"schema_rate": round(n_schema / n, 4) if n else 0.0,
"cost": round(sum(r["cost"] for r in rows), 6),
}
def eval_gate(candidate, store, cases, scorer, baseline=None, baseline_triple=None, tol=0.0) -> dict:
"""评测门（第 2 篇 harness 的位置）：跑候选三元组、对比基线、过了才 lock。
为什么评测对象是三元组不是 Prompt：同一版 Prompt 换个模型、换个数据版本，评测结论不作数，
得重跑——所以第 2 篇的 harness 在本篇不改一行，只换对象。
为什么基线要显式传进来：第一次发布没有前序基线，自己跟自己比；之后每次候选都要跟
"当前 serving 的三元组"比，不然你比的是上一版候选，不是线上正在跑的那一版。"""
cand = scorer.evaluate(candidate, cases)
if baseline is None:
if baseline_triple is not None:
baseline = scorer.evaluate(baseline_triple, cases)
else:
baseline = cand # 首次发布：没有前序基线，自己跟自己比
regressions = []
for key in ("pass_rate", "schema_rate"):
if cand[key] < baseline[key] - tol:
regressions.append(f"{key} {cand[key]:.3f} < 基线 {baseline[key]:.3f}")
passed = not regressions
cost_ratio = round(cand["cost"] / baseline["cost"], 3) if baseline["cost"] else 0.0
report = {
"triple_id": candidate.fingerprint(), "passed": passed,
"regressions": regressions, "candidate": cand, "baseline": baseline,
"cost_ratio": cost_ratio,
# 逐用例差异：出事了能一眼看出是哪些用例回归，而不是只看到一个总数
"diff": [{"case_id": c["case_id"], "passed": c["passed"], "schema_ok": c["schema_ok"]}
for c in cand["cases"]],
}
if passed:
store.lock(candidate, report) # 过了才登记已验证组合，没过连门都进不去
return report
# ---------- 发布门：灰度 + 指针级自动回滚 ----------
DEFAULT_CANARY_STAGES = [
{"name": "shadow", "traffic": 0.0}, # 影子：不接真实流量，只看指标
{"name": "canary", "traffic": 0.05}, # 小流量金丝雀
{"name": "ramp", "traffic": 1.0}, # 分层放量
]
def canary(store, candidate, stages=None, metrics=None, *,
metric_keys=("quality", "schema_rate"), margin=0.0, operator="auto-canary") -> dict:
"""发布门（第 3 篇灰度的位置）：影子 -> 小流量 -> 放量，任一阶段劣化即自动回滚。
为什么回滚在这里是"自动"的：等客诉才发现就晚了；灰度阶段的指标劣化是机器能判的，
判到就滚，不用等人拍板。
为什么回滚是指针级的：store.rollback() 只动 serving 指针，不重建任何东西——
新三元组的零件（Prompt / 数据 / 模型）本来就都还在，指回去就行。
为什么缺指标要 fail-closed：忘传指标就等于过门，等于把灰度架空了，宁可少放量不硬放。"""
stages = stages or DEFAULT_CANARY_STAGES
metrics = metrics or {}
tid = candidate.fingerprint()
old = store.current_serving()
trail = []
for st in stages:
name = st.get("name")
m = metrics.get(name, {})
degraded = []
for k in metric_keys:
cur_v, base_v = m.get(k), m.get(k + "_baseline")
if cur_v is None or base_v is None:
continue
if cur_v < base_v - margin:
degraded.append(f"{k} {cur_v:.2f} < 基线 {base_v:.2f}")
if not m:
degraded.append(f"阶段 {name} 无指标：放量证据不足，fail-closed 不硬放")
trail.append({"stage": name, "traffic": st.get("traffic"), "degraded": degraded,
"ok": not degraded})
if degraded:
rec = RollbackRecord(from_triple=tid, to_triple=old,
reason=f"灰度阶段 {name} 劣化：" + "；".join(degraded),
operator=operator)
store.rollback(rec)
return {"stages": trail, "verdict": "rolled_back", "rollback": rec,
"serving": store.current_serving()}
store.set_serving(tid, reason="灰度三阶段全过，放量", operator=operator)
return {"stages": trail, "verdict": "published", "rollback": None,
"serving": store.current_serving()}
# ---------- 退化监控：只告警不自动改 ----------
DRIFT_THRESHOLDS = {
"data_drift_psi": {"warn": 0.10, "crit": 0.25}, # PSI 方向性口径，按业务校准
"response_drift_psi": {"warn": 0.10, "crit": 0.25}, # 上游静默改权重 / 安全过滤器的信号
"schema_fail_rate": {"warn": 0.005, "crit": 0.05}, # 0.5% -> 5% 跳变是"上游静默更新"强信号
}
def drift_check(online_metrics, thresholds) -> list:
"""退化监控：对 Data Drift / Response Drift 各判一条，超阈值返回告警 + 建议动作。
为什么只告警不自动改：自动改 = 绕开评测门，等于把门拆了。漂移告警是"信号源"不是"执行器"，
它告诉你该去变更入口排一次发布（换数据版本 / 换 Prompt / 换模型），改还是得走门。
为什么漂移监控独立于发布：不发布也会退化——供应商静默改了权重、用户问法变了，
这些都不由你的发布触发。同一个入口，两个信号源：第 11 篇巡检管数据维，这里管模型/响应维。"""
alerts = []
for name, spec in thresholds.items():
if name not in online_metrics:
continue
v = online_metrics[name]
warn, crit = spec.get("warn"), spec.get("crit")
if crit is not None and v > crit:
level = "crit"
elif warn is not None and v > warn:
level = "warn"
else:
continue
alerts.append({
"metric": name, "value": v, "level": level, "thresholds": dict(spec),
"suggestion": (f"{name}={v} 超阈值（warn {warn} / crit {crit}）：只告警不自动改，"
f"请从变更入口走发布流程（换数据版本 / 换 Prompt / 换模型）"),
})
return alerts
# ---------- 退役扫描：一次 SELECT 出待迁移清单 ----------
def retire_model(store, model_id) -> list:
"""退役扫描：扫出所有还钉着这个 model_id 的三元组，返回待迁移清单。
为什么必须是一次查询：模型是耗材不是资产，退役是必然事件——供应商公告到下线典型 60 天到
6 个月、preview 可短至 2 周。它得能一次查出来，不是翻配置文件考古。
serving 排最前——线上正在跑的那个先迁移，别等它先崩。"""
plan = []
for r in store.triples_pinning(model_id):
plan.append({
"triple_id": r["triple_id"], "prompt_id": r["prompt_id"],
"data_version": r["data_version"], "model_id": r["model_id"],
"state": r["state"], "created_at": r["created_at"],
"migration": ("只换 model 一维 -> 出新三元组 -> 过评测门 -> 灰度"
if r["state"] == "serving" else "随下次发布顺带迁移"),
})
return plan
# ---------- 编排：一条线跑完 注册 -> 评测门 -> 发布门 -> 观测 四站 ----------
def publish(registry, store, kinds_content, cfg=None) -> dict:
"""一条线跑完：注册 -> 单一变更校验 -> 评测门 -> 发布门 -> 观测打标 -> 审计落账。
kinds_content: {"prompt": (content, meta), "data": (content, meta), "model": (content, meta)}
为什么把"单一变更校验"放在评测门之前、而不是之后：两维同改的组合根本不该进评测门——
测了也归因不到哪一维，白测。先在门口拒掉，让用户拆成两次发布（demo 断言②的落点）。"""
cfg = dict(cfg or {})
cases = list(cfg.get("cases", []))
scorer = cfg.get("scorer")
if scorer is None:
raise ValueError("publish 需要 cfg['scorer']：评测门没有评分桩就没法放行")
operator = cfg.get("operator", "release-bot")
# ① 注册：三件各自定身份，凑一起算三元组指纹
ids = {k: registry.put(k, kinds_content[k][0], kinds_content[k][1])
for k in ("prompt", "data", "model")}
triple = ReleaseTriple(ids["prompt"], ids["data"], ids["model"])
out = {"triple": triple, "triple_id": triple.fingerprint(), "changed": [], "rejected": None,
"report": None, "canary": None, "serving": store.current_serving(), "stations": []}
out["stations"].append(("① 注册", f"prompt={ids['prompt']} data={ids['data']} "
f"model={ids['model']} -> 指纹 {triple.fingerprint()}"))
# 变更入口：单一变更原则（相对当前 serving）
cur_id = store.current_serving()
if cur_id:
changed = single_change_from(store.triple_of(cur_id), triple)
out["changed"] = changed
if len(changed) > 1:
out["rejected"] = f"一次动了 {len(changed)} 维（{'+'.join(changed)}）：拆成 {len(changed)} 次发布"
out["stations"].append(("变更入口", "拒收：" + out["rejected"]))
return out
out["stations"].append(("变更入口", f"单一变更校验通过（变化维度：{changed}）"))
else:
out["stations"].append(("变更入口", "首次发布：没有前序 serving，跳过单一变更校验"))
store.register_triple(triple)
# ② 评测门（第 2 篇 harness）：评测对象是三元组，不是 Prompt
baseline = scorer.evaluate(store.triple_of(cur_id), cases) if (scorer and cur_id) else None
report = eval_gate(triple, store, cases, scorer, baseline=baseline)
out["report"] = report
if not report["passed"]:
out["rejected"] = "评测门未通过：" + "；".join(report["regressions"])
out["stations"].append(("② 评测门", "拦下 -> " + out["rejected"]))
return out
out["stations"].append(("② 评测门", f"通过：pass {report['candidate']['pass_rate']:.2f} / "
f"schema {report['candidate']['schema_rate']:.2f} / "
f"成本 x{report['cost_ratio']}"))
# ③ 发布门（第 3 篇灰度）：影子 -> 小流量 -> 放量，劣化即指针级回滚
canary_res = canary(store, triple, cfg.get("canary_stages"), cfg.get("canary_metrics"),
operator=operator)
out["canary"] = canary_res
out["serving"] = store.current_serving()
if canary_res["verdict"] == "rolled_back":
out["rejected"] = "发布门回滚：" + canary_res["rollback"].reason
out["stations"].append(("③ 发布门", "自动回滚 -> 指针指回 "
+ (canary_res["rollback"].to_triple or "(无)")))
return out
out["stations"].append(("③ 发布门", "三阶段全过，放量：指针切到新三元组"))
# ④ 观测：trace / 成本按三元组指纹打标
out["stations"].append(("④ 观测", f"trace 与成本账按指纹 {triple.fingerprint()} 打标"))
return out
# ---------- mock 世界：两版 Prompt、两版数据、两个模型 ----------
PROMPT_V1 = ("你是 X200 客服助手。只依据检索到的保修政策回答。"
"输出 JSON：{\"answer\": string, \"months\": int}。不确定就答不知道。")
PROMPT_V2 = ("你是 X200 客服助手。只依据检索到的保修政策回答。"
"输出 JSON：{\"answer\": string, \"months\": int}。不确定就答不知道，别猜。")
DATA_V1 = "X200 保修政策 v1 切片：整机保修 12 个月，电池保修 24 个月。"
DATA_V2 = "X200 保修政策 v2 切片：整机保修 12 个月，电池保修 24 个月，新增 X210 延保条款。"
MODEL_OLD = "mock://flagship-old" # 模型这一维的内容就是它的 ID / 权重指针
MODEL_NEW = "mock://flagship-new"
MODEL_META = {
"old": {"name": "old-flagship", "leaderboard": 0.71, "cost_per_1k": 0.6},
"new": {"name": "new-flagship", "leaderboard": 0.83, "cost_per_1k": 1.8},
}
CASES = ["case_warranty_months", "case_refund_policy", "case_json_schema", "case_injection"]
def _score_table(m_old, m_new, p1, p2) -> dict:
"""造一张显式评分表：旧模型配哪版 Prompt 都稳；新模型配旧 Prompt 时 schema 遵守率掉、
两类意图答非所问——这就是"榜分更高 != 我的应用更好"的确定性复刻。
为什么按 (model_id, prompt_id, case_id) 索引：换模型、换 Prompt 都可能让同一批用例的结论变，
索引少任何一维，都锁不住"换个模型、同一版 Prompt 就不是同一件事"。"""
ok_old = {"passed": True, "schema_ok": True, "cost": 0.0009}
ok_new = {"passed": True, "schema_ok": True, "cost": 0.0027}
table = {}
for p in (p1, p2):
for c in CASES:
table[(m_old, p, c)] = dict(ok_old)
for c in CASES:
table[(m_new, p1, c)] = dict(ok_new)
# 新模型两个"更聪明反被聪明误"的用例：JSON 契约没守住、退款意图答非所问
table[(m_new, p1, "case_json_schema")] = {"passed": False, "schema_ok": False, "cost": 0.0031}
table[(m_new, p1, "case_refund_policy")] = {"passed": False, "schema_ok": True, "cost": 0.0029}
return table
def clean_canary_metrics() -> dict:
"""三阶段都稳：全过、放量。"""
base = {"quality_baseline": 0.92, "schema_rate_baseline": 0.96}
return {name: {"quality": 0.92, "schema_rate": 0.96, **base}
for name in ("shadow", "canary", "ramp")}
def degrading_canary_metrics() -> dict:
"""影子阶段稳、小流量阶段开始劣化：用来复现"灰度第二阶段劣化、自动回滚"。"""
base = {"quality_baseline": 0.92, "schema_rate_baseline": 0.96}
return {
"shadow": {"quality": 0.92, "schema_rate": 0.96, **base},
"canary": {"quality": 0.84, "schema_rate": 0.88, **base}, # 劣化：质量 -0.08、遵守率 -0.08
"ramp": {"quality": 0.84, "schema_rate": 0.88, **base},
}
class MockWorld:
"""一份 mock 世界：两版 Prompt、两版数据、两个模型（新模型榜分更高但应用侧 schema 遵守率更差）。
为什么 mock 数据要做成一个对象：build_demo() 和 7 个断言必须共用同一份确定性输入——
测试和演示看到的得是同一个世界，否则断言锁不住演示里那个结论。"""
def __init__(self):
self.registry = PartRegistry()
self.prompt_text = {"v1": PROMPT_V1, "v2": PROMPT_V2}
self.data_text = {"v1": DATA_V1, "v2": DATA_V2}
self.model_text = {"old": MODEL_OLD, "new": MODEL_NEW}
self.prompt_id = {k: self.registry.put("prompt", v, {"label": f"prompt-{k}"})
for k, v in self.prompt_text.items()}
self.data_id = {k: self.registry.put("data", v, {"label": f"data-{k}"})
for k, v in self.data_text.items()}
self.model_id = {k: self.registry.put("model", v, dict(MODEL_META[k]))
for k, v in self.model_text.items()}
self.cases = list(CASES)
self.leaderboard = {self.model_id[k]: MODEL_META[k]["leaderboard"] for k in MODEL_META}
self.model_name = {k: MODEL_META[k]["name"] for k in MODEL_META}
self.scorer = MockScorer(
_score_table(self.model_id["old"], self.model_id["new"],
self.prompt_id["v1"], self.prompt_id["v2"]),
profiles={self.model_id["old"]: {"passed": True, "schema_ok": True, "cost": 0.0009},
self.model_id["new"]: {"passed": True, "schema_ok": True, "cost": 0.0027}})
def triple(self, prompt="v1", data="v1", model="old") -> ReleaseTriple:
return ReleaseTriple(self.prompt_id[prompt], self.data_id[data], self.model_id[model])
def parts(self, prompt="v1", data="v1", model="old") -> dict:
return {"prompt": (self.prompt_text[prompt], {"label": f"prompt-{prompt}"}),
"data": (self.data_text[data], {"label": f"data-{data}"}),
"model": (self.model_text[model], dict(MODEL_META[model]))}
def cfg(self, canary_metrics=None) -> dict:
return {
"cases": self.cases,
"scorer": self.scorer,
"canary_stages": list(DEFAULT_CANARY_STAGES),
"canary_metrics": (canary_metrics if canary_metrics is not None
else clean_canary_metrics()),
"operator": "release-bot",
}
def build_mock_world() -> MockWorld:
"""测试和演示共用的那一份确定性输入。"""
return MockWorld()
# ---------- demo ----------
def build_demo():
"""跑一条完整主线：首次发布 -> 换模型被门拦 -> 两维同改被拒 -> 换 Prompt 灰度回滚 ->
漂移告警 -> 退役扫描。下面这些结论是对这份 mock 数据的确定性结论，不是对真实发布流水线的
承诺——真实门禁阈值取决于你的业务 SLA（量级锚点见文章附录 B）。"""
w = build_mock_world()
store = ReleaseStore()
print("== LLMOps 发布流水线：注册 -> 锁三元组 -> 单一变更 -> 评测门 -> 灰度 -> 观测 -> 退役 ==")
print(f"零件层：Prompt {len(w.prompt_id)} 版 | 数据 {len(w.data_id)} 版 | 模型 {len(w.model_id)} 个"
f"（{w.model_name['old']} 榜分 {w.leaderboard[w.model_id['old']]}，"
f"{w.model_name['new']} 榜分 {w.leaderboard[w.model_id['new']]} " + "；".join(row["degraded"])) if row["degraded"] else "稳"
print(f" 阶段 {row['stage']}（放量 {row['traffic']}）：{tag}")
print(f" 结论：{out3['rejected']}")
print(f" serving 指回 {store.current_serving()}（指针级回滚，零重建）")
online = {"data_drift_psi": 0.31, "response_drift_psi": 0.08, "schema_fail_rate": 0.012}
alerts = drift_check(online, DRIFT_THRESHOLDS)
print(f"\n[退化监控] 线上指标 {online}")
for a in alerts:
print(f" {a['level'].upper()} {a['metric']}={a['value']} -> {a['suggestion'][:36]}...")
print(f" serving 仍为 {store.current_serving()}（只告警不自动改：自动改 = 绕开评测门）")
plan = retire_model(store, w.model_id["old"])
print(f"\n[模型退役] 扫 {w.model_name['old']}：待迁移 {len(plan)} 个三元组")
for t in plan:
print(f" {t['triple_id']} [{t['state']}] -> {t['migration']}")
rows = store.list_audit()
print(f"\n审计账本（最后 {len(rows)} 行）：")
for r in rows:
print(f" {r['action']:9s} {(r['from_triple'] or '-'):>8s} -> {r['to_triple']}"
f" | {r['reason'][:30]}")
print("\n确定性结论：三元组换一维指纹必变；两维同改被拒；榜分更高的新模型被评测门拦；"
"换 Prompt 灰度第二阶段劣化后指针级回滚；漂移超阈值只告警不动 serving；"
"退役一次 SELECT 扫出 2 个待迁移三元组。")
if __name__ == "__main__":
import sys
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
build_demo()
跑 python mini_llmops.py，六个站点的收/拦/放行清清楚楚：
== LLMOps 发布流水线：注册 -> 锁三元组 -> 单一变更 -> 评测门 -> 灰度 -> 观测 -> 退役 ==
零件层：Prompt 2 版 | 数据 2 版 | 模型 2 个（old-flagship 榜分 0.71，new-flagship 榜分 0.83 指纹
变更入口：首次发布：没有前序 serving，跳过单一变更校验
② 评测门：通过：pass 1.00 / schema 1.00 / 成本 x1.0
③ 发布门：三阶段全过，放量：指针切到新三元组
④ 观测：trace 与成本账按指纹 打标
serving =
[换模型] 把 model 换成榜分更高的 new-flagship，Prompt / 数据都没动
变更维度：['model']（只有 model 这一维）
评测门：pass 0.50（基线 1.00） | schema 遵守率 0.75（基线 1.00） | 成本 x3.167
结论：评测门未通过：pass_rate 0.500 < 基线 1.000；schema_rate 0.750 < 基线 1.000
serving 仍为 （榜分更高也上不去）
[两维同改] 同时改 Prompt 措辞 + 换数据版本
变更维度：['prompt', 'data']（2 维）
结论：一次动了 2 维（prompt+data）：拆成 2 次发布
[换 Prompt] 只改 Prompt 一维，评测门过了，灰度第二阶段劣化
阶段 shadow（放量 0.0）：稳
阶段 canary（放量 0.05）：劣化 -> quality 0.84 < 基线 0.92；schema_rate 0.88 < 基线 0.96
结论：发布门回滚：灰度阶段 canary 劣化：quality 0.84 < 基线 0.92；schema_rate 0.88 < 基线 0.96
serving 指回 （指针级回滚，零重建）
[退化监控] 线上指标 {'data_drift_psi': 0.31, 'response_drift_psi': 0.08, 'schema_fail_rate': 0.012}
CRIT data_drift_psi=0.31 -> data_drift_psi=0.31 超阈值（warn 0.1...
WARN schema_fail_rate=0.012 -> schema_fail_rate=0.012 超阈值（warn 0.0...
serving 仍为 （只告警不自动改：自动改 = 绕开评测门）
[模型退役] 扫 old-flagship：待迁移 2 个三元组
[serving] -> 只换 model 一维 -> 出新三元组 -> 过评测门 -> 灰度
[rolled_back] -> 随下次发布顺带迁移
审计账本（最后 4 行）：
lock - -> | 评测门通过：登记为已验证组合
publish - -> | 灰度三阶段全过，放量
lock - -> | 评测门通过：登记为已验证组合
rollback -> | 灰度阶段 canary 劣化：quality 0.84 < ...
确定性结论：三元组换一维指纹必变；两维同改被拒；榜分更高的新模型被评测门拦；换 Prompt 灰度第二阶段劣化后指针级回滚；漂移超阈值只告警不动 serving；退役一次 SELECT 扫出 2 个待迁移三元组。
读一遍这个输出，主线就出来了：第一次发布三件锁成基线三元组、过了六站；换模型只动一维、单一变更校验放过，但评测门把它拦下来了——榜分从 0.71 到 0.83 的“更强”，换来的是 schema 遵守率从 1.00 掉到 0.75、成本还涨到 3.167 倍；两维同改在变更入口就被拒收；换 Prompt过了评测门，但在灰度第二阶段劣化，指针级回滚，账本上留下一行 rollback；漂移监控告警但没动 serving 指针；退役扫描一次 SELECT 扫出 2 个还钉着旧模型的三元组。
接着看 7 个断言怎么把这些结论钉死。release/test_mini_llmops.py：
"""
7 个断言锁死"LLMOps 不是 MLOps 加个 L，发布原子是版本三元组"：
三元组换一维指纹必变 / 单一变更原则 / 未评测组合不许上 serving /
更强的模型也可能被评测门拦 / 灰度劣化自动回滚 / 漂移只告警不自动改 / 退役一次扫出清单。
纯标准库 + assert：
python test_mini_llmops.py
# 或 pytest test_mini_llmops.py
"""
from mini_llmops import (
DRIFT_THRESHOLDS, ReleaseStore, ReleaseTriple, build_mock_world,
degrading_canary_metrics, drift_check, eval_gate, publish, retire_model,
single_change_from,
)
def test_triple_identity_changes_with_any_dim():
"""断言①：三元组身份——换任何一维，指纹必变（呼应第 4 篇"内容哈希才是身份"）。
改 Prompt 里一个标点就该出一个新 Prompt 版本、一个新三元组指纹；只改 model_id、
只改 data_version 同理。三件任意一件动了，都是"另一次发布"。"""
w = build_mock_world()
base = w.triple("v1", "v1", "old")
# 改一个标点，就是一个新的 Prompt 版本（哈希身份：内容变一个字节，身份就变）
tweaked = w.registry.put("prompt", w.prompt_text["v1"].replace("。", ".", 1), {})
assert tweaked != w.prompt_id["v1"]
candidates = [
ReleaseTriple(tweaked, base.data_version, base.model_id), # 只动 Prompt
ReleaseTriple(base.prompt_id, w.data_id["v2"], base.model_id), # 只动数据
ReleaseTriple(base.prompt_id, base.data_version, w.model_id["new"]), # 只动模型
]
for cand in candidates:
assert cand.fingerprint() != base.fingerprint() # 换一维，指纹必变
assert len(single_change_from(base, cand)) == 1 # 而且正好是"一维"
def test_single_change_principle():
"""断言②：单一变更原则——相对当前 serving，一次只允许一维变化（对应取舍点①的纪律）。
single_change_from() 对"只改 Prompt"返回 1 维、对"同时改 Prompt + 数据"返回 2 维；
publish() 对 2 维直接拒收、提示拆成两次发布。这是"两维同改不知道谁的锅"的可复现化。"""
w = build_mock_world()
cur = w.triple("v1", "v1", "old")
assert single_change_from(cur, w.triple("v2", "v1", "old")) == ["prompt"]
assert single_change_from(cur, w.triple("v1", "v2", "old")) == ["data"]
assert single_change_from(cur, w.triple("v1", "v1", "new")) == ["model"]
assert len(single_change_from(cur, w.triple("v2", "v2", "old"))) == 2
store = ReleaseStore()
publish(w.registry, store, w.parts("v1", "v1", "old"), w.cfg())
out = publish(w.registry, store, w.parts("v2", "v2", "old"), w.cfg())
assert out["rejected"] and "拆成 2 次发布" in out["rejected"]
# 被拒的三元组没上 serving：指针还停在基线上
assert store.current_serving() == cur.fingerprint()
def test_unverified_triple_cannot_serve():
"""断言③：没被锁进"已验证组合表"的三元组不许上 serving（评测门是准入不是仪式）。
set_serving() 一个没经过 eval_gate -> lock() 的三元组直接抛错；跑过门的能上——
版本号再新、零件再强，没一起评测过就不算数。"""
w = build_mock_world()
store = ReleaseStore()
t = w.triple("v1", "v1", "new")
store.register_triple(t)
try:
store.set_serving(t.fingerprint())
raise AssertionError("没进 verified 表的三元组不该上 serving")
except ValueError:
pass
# 跑过评测门（内部 lock 登记已验证组合）的能上
report = eval_gate(t, store, w.cases, w.scorer)
assert report["passed"]
assert store.is_verified(t.fingerprint())
store.set_serving(t.fingerprint())
assert store.current_serving() == t.fingerprint()
def test_stronger_model_can_be_blocked():
"""断言④：换模型是发布不是升级——更强的模型也可能被评测门拦（对应取舍点②）。
mock 里"榜分更高"的新模型在 golden set 上 schema 遵守率掉、整体 regressed；
eval_gate 判不过 -> 新三元组进不了 serving，旧三元组继续服务。
把"越强 != 我的应用更好"钉成断言。"""
w = build_mock_world()
store = ReleaseStore()
publish(w.registry, store, w.parts("v1", "v1", "old"), w.cfg())
base_id = w.triple("v1", "v1", "old").fingerprint()
assert store.current_serving() == base_id
out = publish(w.registry, store, w.parts("v1", "v1", "new"), w.cfg())
assert out["changed"] == ["model"] # 只动了模型这一维
assert out["rejected"] and "评测门" in out["rejected"]
assert store.current_serving() == base_id # 旧三元组继续服务
# 榜分更高，却在应用侧被门拦——越强不等于我的应用更好
assert w.leaderboard[w.model_id["new"]] > w.leaderboard[w.model_id["old"]]
cand, base = out["report"]["candidate"], out["report"]["baseline"]
assert cand["schema_rate"] < base["schema_rate"] # schema 遵守率掉了
assert not store.is_verified(out["triple_id"]) # 没进已验证组合表
assert out["report"]["cost_ratio"] > 1 # 成本还更高，第 9 篇账本整个重算
def test_canary_degradation_rolls_back():
"""断言⑤：灰度任一阶段劣化 -> 自动回滚到旧三元组，指针级、零重建。
canary 走到第二阶段 mock 指标劣化 -> serving 指回旧三元组、audit 表多一行 rollback
（含 from/to/reason）——回滚是结构里长出来的，不是靠人记。"""
w = build_mock_world()
store = ReleaseStore()
publish(w.registry, store, w.parts("v1", "v1", "old"), w.cfg())
base_id = w.triple("v1", "v1", "old").fingerprint()
# 只改 Prompt 一维：评测门能过（mock 里旧模型对新 Prompt 是稳的），灰度第二阶段劣化
out = publish(w.registry, store, w.parts("v2", "v1", "old"),
w.cfg(canary_metrics=degrading_canary_metrics()))
assert out["report"]["passed"] # 门是过了的
assert out["rejected"] and "回滚" in out["rejected"]
stages = [r["stage"] for r in out["canary"]["stages"]]
assert stages[:2] == ["shadow", "canary"] # 走到第二阶段才劣化
assert store.current_serving() == base_id # 指针指回旧三元组，零重建
rows = [r for r in store.list_audit() if r["action"] == "rollback"]
assert len(rows) == 1
assert rows[0]["from_triple"] == out["triple_id"] # 从哪个三元组
assert rows[0]["to_triple"] == base_id # 到哪个三元组
assert rows[0]["reason"] # 为什么
def test_drift_alerts_without_auto_change():
"""断言⑥：退化监控触发再发布建议（只告警、不自动改）。
mock 线上指标造出 Data Drift 超阈值 -> drift_check 返回告警 + "走变更入口"；
断言它没有动 serving 指针——自动改 = 绕开评测门，门就白建了。
再断言告警里只带信号、不带动作：漂移监控是"信号源"不是"执行器"，物理上没有改 serving 的接口。"""
w = build_mock_world()
store = ReleaseStore()
publish(w.registry, store, w.parts("v1", "v1", "old"), w.cfg())
before = store.current_serving()
alerts = drift_check(
{"data_drift_psi": 0.31, "response_drift_psi": 0.08, "schema_fail_rate": 0.012},
DRIFT_THRESHOLDS)
assert alerts
by_metric = {a["metric"]: a for a in alerts}
assert by_metric["data_drift_psi"]["level"] == "crit" # 0.31 > 0.25，显著
assert by_metric["schema_fail_rate"]["level"] == "warn" # 0.012 > 0.005，警告
assert "response_drift_psi" not in by_metric # 0.08 < 0.10，没超
assert all("变更入口" in a["suggestion"] for a in alerts)
# 告警只带信号、不带可执行动作字段：漂移监控是信号源不是执行器
assert all(set(a) = 2
assert plan[0]["state"] == "serving" # serving 优先，先迁移线上的
assert any(t["triple_id"] == store.current_serving() for t in plan)
assert all(t["model_id"] == w.model_id["old"] for t in plan)
assert plan[0]["prompt_id"] and plan[0]["data_version"] # 能对账到具体版本
assert plan[0]["migration"] # 每条都带迁移动作
if __name__ == "__main__":
import sys
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
test_triple_identity_changes_with_any_dim()
test_single_change_principle()
test_unverified_triple_cannot_serve()
test_stronger_model_can_be_blocked()
test_canary_degradation_rolls_back()
test_drift_alerts_without_auto_change()
test_retire_model_scan()
print("全部 7 个断言通过：三元组换一维指纹必变 / 单一变更 / 未评测组合不许上 serving / "
"更强的模型也可能被门拦 / 灰度劣化自动回滚 / 漂移只告警不自动改 / 退役一次扫出清单")
跑 python test_mini_llmops.py，全绿。7 个断言里我挑三条说透：
断言③是整条流水线的守门员。 它先 register_triple 登记一个三元组（账上有这笔），之后直接 set_serving——抛错。这一刀为什么最要紧：注册和发布是两件事，登记过不等于被评测过；只有 eval_gate 内部调了 store.lock() 的那组指纹，才进得了“已验证组合表”。想亲手验证这刀很容易：把 eval_gate 里那句 store.lock(candidate, report) 注释掉再跑测试——publish() 走到灰度那步 set_serving 会当场抛 ValueError（这组没进已验证组合表），断言③的 is_verified 也一起红。守门员一缺席，整条流水线集体炸锅，而不是只红一条断言。
断言④是取舍点②的代码证据。 它把“新模型榜分 0.83 > 旧模型 0.71”和“新模型 schema 遵守率 0.75 < 旧模型 1.00”这两句同时钉在一段代码里——榜分更高的那个上不去，旧三元组继续服务，成本还是 3 倍。这不是随机数凑出来的巧合，是 _score_table 里写死的一行 table[(m_new, p1, "case_json_schema")] = {"passed": False, "schema_ok": False, ...}：结构化输出的闸（第 7 篇）保的是契约，契约能不能守跟我用哪个模型强相关——换模型必须重测遵守率，不能只看它“更聪明”。
断言⑤说明回滚是从结构里长出来的，不是靠人记的。 canary 走到第二阶段 canary 时指标劣化，store.rollback() 生成一条 RollbackRecord，audit 表多一行（from/to/reason 齐），serving 指针指回旧三元组。留意 stages[:2] == ["shadow", "canary"]——影子阶段是稳的，灰度第二站才崩的。这正好是踩坑①那个“影子里看着挺好、放 5% 才出事”的压缩版：影子看不出问题，不代表没问题；小流量那一站才说真话。
四、踩坑记录：这三个坑，每个都真付过费
4.1 换模型一行配置上生产，schema 遵守率掉了，回滚时对不上账
症状：把网关里的 model 名换成榜分更高的新版，影子看着没问题，放 5% 之后第 7 篇的 schema 闸开始降级，有几类意图答非所问。回滚时我卡住了——模型名我知道，可“当时那一版 Prompt 配的是哪个数据版本”我答不上来，因为从来没有一个地方把这三件记在一起。
排查：根因不是“新模型不行”，是我把换模型当升级（改配置）而不是当发布（换零件要过门）。换模型同时改了四件事：Prompt 的效果、schema 遵守率、成本结构、缓存行为——我只看了榜分。榜分更高 ≠ 我的应用更好，这不是体感，是 GSM1k 那 8 个点、Goodhart 通道那条 MMLU 曲线在公开说的同一件事。
修复：模型进 registry（能力/schema 遵守率/成本/合规四件标签），换模型走完整发布流程（重跑 golden set → 影子 → 灰度 → 熔断），回滚粒度改成三元组指针。教训：换模型不是升级，是发布——榜分高和应用变好之间，隔着一次评测；省掉它，就是拿线上流量赌。
4.2 同时改了 Prompt 和数据版本，指标掉了半天不知道是谁的锅
症状：一次迭代里我又调了 Prompt 措辞、又换了知识库的新版本（第 12 篇的 V_{k+1}），指标掉了。我花了半天在 Prompt 和检索之间二选一，最后发现两个都有问题、还互相掩盖——单独看哪一个都不至于掉这么多，凑一起才炸。
排查：根因是没有“单一变更原则”。两个维度同时动，三元组变了两次，评测结论归因不到任何一维。这不是我一个人的毛病，Microsoft 那门 evaluate-optimize-agents 官方课里就写着“每个实验变体一个分支、把性能变化归因到具体那一处改动，而不是把多处改动混在一个分支里”；Google 的 ML 指南也是同一条——一次只做一个小的、迭代的改动，改优了把它当新基线。两家大厂跟我踩的是同一个坑，说明这不是我笨。
修复：一次只动一件，两维同改就拆成两次发布。demo 里 single_change_from() 直接把“两维同改”拒掉，publish() 连评测门都不让它进——测了也归因不到，白测。教训：三元组一次只许在一个维度上 +1——省掉的拆分成本，会以“出事了不知道谁的锅”连本带利还回来。
4.3 供应商公告下线旧模型，我扫不出哪些线上配置还钉着它
症状：收到“某模型 API 将在 X 个月后退役”的通知，我开始慌：哪些应用、哪些版本还钉着它？只能翻配置文件、翻历史记录，翻了一下午还漏了两处。我们用的模型不在少数，改一处漏一处。这事的紧迫感是真有的：OpenAI 对 GA 模型至少提前 6 个月、specialized 变体至少 3 个月、preview 可以短到 2 周；Anthropic 至少 60 天（claude-sonnet-4-20250514 从 2026-04-14 标记 deprecated 到 2026-06-15 退休是 62 天）；Azure GA 给 60 天、preview 只给 30 天。“公告到下线”典型 60 天到 6 个月，preview 真的可能两周就没了。
排查：根因是“模型”这一维从来没有被登记进一个可查询的地方——模型 ID 散在配置里，退役就成了考古。你没把它当零件，它退役的时候你连清单都列不出来。
修复：模型注册 + 退役扫描（按 model_id 扫所有钉着它的三元组，列出待迁移清单，serving 排最前）。更根本的一条纪律：假设任何一个模型明天都会消失，从第一天就把“可迁移”当设计目标。demo 里 retire_model() 就是一条 SELECT（triples_pinning），一次扫出 2 个待迁移三元组。教训：模型不是资产，是耗材——退役不是意外，是必然；可迁移性在第一天设计，不在最后一天抢救。
五、选型对比：三条路线、工具现状、7 问清单
5.1 大表：LLMOps 三条路线
方案发布原子覆盖哪些变更线回滚粒度组合可复现性适用团队① 没有流水线（改哪算哪）没有“发布”这个概念，配置热更 + 拍脑袋上线只盖住“改 Prompt”，模型和数据版本没进发布全靠记忆：改回去、或者重启低：说不清线上跑的是哪三件单人 Demo、内部工具、验证阶段② MLOps 直搬（模型为中心的 registry + 训练流水线）一个模型权重模型线很强，Prompt 和数据版本没进同一个发布回滚了模型，Prompt 和数据还停在新版中：模型可复现，另两维不可复现有训练团队的 AI 公司，模型迭代是主线③ 以版本三元组为中心的 LLM 应用发布流水线（本篇 demo 就是最小达成）一个版本三元组（Prompt × 数据 × 模型）三条线全覆盖，变更从同一个入口进指针级：指回旧三元组，零重建高：每次上线能说清是哪三件、能整体回滚把 LLM 应用当生产系统运营的团队
结论：路线①是“赌运气”——省下的流程成本会以线上事故的形式还回来；路线②是“用旧地图走新路”——它的发布原子是模型，Prompt 和数据版本没进发布，你回滚了模型，另两件还停在新版上；路线③是“为 LLM 应用量身定做的发布纪律”。 小团队不用一上来就建全套：先做“模型注册 + 三元组锁 + 单一变更原则”三件（第 4 篇的内容哈希身份直接复用），评测门和灰度复用第 2、3 篇，观测复用第 8、9 篇——先把“一次发布能说清是哪三件”这件事立起来，其他站随规模补。
5.2 副表：可落地的工具现状（2026 已核实）
环节工具/版本与本篇的关系模型注册/版本MLflow Model Registry 3.16.1；Hugging Face Hub（SaaS 无产品版本号）生产化落点是本篇 PartRegistry 的模型那半；MLflow 的 LoggedModel 把模型↔代码版本↔Prompt↔评测链接起来。LoRA 归档口径：HF Hub 直接支持 adapter-only + pin base revision；MLflow 3.0 起 registry 面向适配器扩展，裸 LoRA 直存还在等 mlflow.diffusers flavor 进 stable——别读成“MLflow 已完整支持 LoRA 归档”Prompt 资产管理Langfuse Prompt Management v4.38.0（v4 于 2026-08 发布）；LangSmith（SaaS/企业自托管，SDK 0.13.0）Langfuse 每个 prompt 版本自动有 version ID + 标签，deploy = 移 production 标签、rollback = 移回旧标签、零重部署；它管“内容 + 标签发布 + 观测”，第 4 篇自研 registry 管“内容哈希身份”——薄厚互补数据版本DVC 3.67.1（第 12 篇已核）本篇三元组里 data_version 的取值来源，不重复核实实验追踪W&B（Weave + SDK，SaaS 无产品版本号）；MLflowWeave 把 prompt / dataset / model 当一等版本化对象，是“最能三件记成一次 run”的开箱工具，最贴近三元组心智网关 / Trace / 评测LiteLLM（第 5 篇已核）；Langfuse / OTel（第 8 篇）；RAGAS v0.4.3+ / DeepEval 3.9.9（第 2/11 篇已核）直接沿用，本篇不重复核实。注意：第 8/10 篇核的 Langfuse v3.176.0 已被 2026-08 的 v4 取代，本篇起按 v4 口径推理引擎vLLM 0.26.0 / SGLang 0.5.16 / TensorRT-LLM 1.2.1开篇 4.2 第⑤条，本篇只给边界：那是推理成本/吞吐的问题，跟发布纪律是两件事
接到工具上，就是一句话：demo 里 PartRegistry（零件注册）+ ReleaseStore（三元组 + 审计账本，sqlite3）的生产化，就是“MLflow/HF Hub 管模型 + Langfuse 管 Prompt + DVC 管数据 + 一张发布账本”；eval_gate 就是第 2 篇 harness、canary 就是第 3 篇灰度、drift_check 就是监控告警——流水线六站的逻辑一行不改。 换的始终是接口后面那层达成，编排骨架一行不动。
5.3 可复用 checklist：LLMOps 落地 7 问
1. 你的“一次发布”是什么？是一个模型，还是一组版本？ 没有三元组 → 出事说不清是哪三件。
2. 换模型走过评测门吗，还是改一行配置就上了？ 没走 → 越强的模型可能越差，你只是在赌。
3. 一次发布里会不会同时动了两件（Prompt + 数据 / 数据 + 模型）？ 会 → 出事归因不到任何一维。
4. 回滚的粒度是什么？ 重跑流程 → 太慢；指针级回滚 → 零重建。
5. 出问题时能一眼看出“这批异常属于哪个三元组”吗？ 不能 → trace 和成本没按三元组打标。
6. 模型退役了，你能扫出哪些线上版本还钉着它吗？ 不能 → 退役那天要考古。
7. 你在微调之前，把 Prompt/数据/路由榨干了吗？ 没榨干就上微调 → 最贵、最难回滚的一步先花了。
六、总结 + 下一篇预告
回到开篇 4.2 那张上位图。前 12 篇造好的零件，在这一篇被串成了一条流水线：Prompt / 数据 / 模型三条线的变更从同一个口进来，锁成一个三元组，过评测门、过发布门，被 Trace 和成本账盯着，被反馈闭环推着，被退化监控和退役扫描兜着，转一圈回到入口。开篇 4.2 那三阶段飞轮——Build & Test → Ship & Monitor → Learn & Refine——不是一句口号，就是这条环形流水线的原话（出处是 Google Cloud Agent Quality Flywheel，2026）。
一句话收口：LLMOps 管的不是模型，是“一次可信发布的完整身份”——三件各自有身份、凑一起有指纹、回滚有指针、退役有清单。
但这条流水线是给“一个应用”用的。当一个业务需要多个 Agent 分工协作（顺序编排 / 并行 / 辩论 / 分层调度），这套发布纪律、这套“变更 → 评测 → 灰度 → 观测”的骨架要怎么扩到多 Agent 上？每个 Agent 各有各的 Prompt、有的还各有各的模型，三元组会不会变成“九元组”？下一篇《Multi-Agent 协作实战》：楼盖好了、也管起来了，接下来是楼里住几个人、怎么分工的问题。一个应用的发布纪律在本篇，多个 Agent 怎么协作，在第 14 篇。
附录 A：LLMOps 上位图完整零件映射表
正文那张零件表是给“对着代码库点一遍”用的，这版补上输出物与纪律来源。
前篇产出是什么零件挂在流水线哪一站输出物纪律来源本篇补的那句话第 4 篇 Prompt registry三元组第 1 维变更入口 → ① 注册prompt_id（content_hash）内容哈希才是身份，semver 只是标签身份从 Prompt 扩到三件，凑一起再算三元组指纹第 12 篇数据底座 ReleaseBundle三元组第 2 维变更入口 → ① 注册data_version（不可变快照）版本不能原地改，只能出新版本改数据 = 出新的发布，不是“顺手更新一下索引”第 5 篇网关（路由/fallback）三元组第 3 维变更入口 → ① 注册model_id（多模型并行）按版本 ID 路由网关管路由，管不了“这模型配这 Prompt 能不能上”第 2 篇 Golden Set评测门② 评测门逐用例 diff + 通过/回归不过门不许出 staging评测对象从“一个 Prompt”改成“一个三元组”第 3 篇灰度 + 熔断发布门③ 发布门分阶段指标 + 回滚动作任一阶段劣化即回滚回滚粒度从“服务版本”改成“三元组指针”第 8 篇全链路 Trace观测④ 观测trace（按指纹打标）PII 不落、降级要标trace 按三元组指纹打标，出事能一眼归组第 9 篇成本六维成本账④ 观测每三元组成本便宜杠杆先用尽换一维就要重算一遍成本账第 10 篇反馈闭环飞轮的推力⑤ 反馈仲裁后的 golden set 增量弱信号必须仲裁反馈挂 trace_id + 三元组指纹双标第 11 篇换版纪律不可变纪律⑥ 退化监控与退役巡检告警 → 再发布触发源不能 UPDATE 只能换版本升格为“三件都不能原地改”第 6 篇护栏 / 第 7 篇 schema 闸常驻件全程对抗回归 + 遵守率换模型/换 Prompt 要重过护栏版本化但先不塞三元组（免得组合爆炸）
附录 B：组合爆炸、微调成本、漂移阈值、退役周期（方向性口径）
正文只留了大白话和量级，这里给锚点。下面这些量级都是方向性的、来自我自己的场景，落到你的业务请自己重算。
- 组合爆炸的估算（我的场景估算，不是公开引用）：Prompt 10 版 × 数据 8 版 × 模型 5 版 = 400 组。这个 400 是我拿自己场景算的演示数字，没有公开权威出处，别当引用。公开能站住的是收敛办法：Google ML 指南说“一次只做一个小的改动”，并明确反对照搬组合（“同事那组超参有效，不代表对你这模型也有效”，要逐项试）；Microsoft 官方课说每个实验变体一个分支，理由是“把性能变化归因到具体那一处改动，而不是把多处改动混在一个分支里”。单一变更原则 + 只锁被评测过的组合 + 定期清理过期组合，这三条是能把 400 收敛下来的办法。
- 榜单 vs 应用评测不一致的公开证据：GSM1k（NeurIPS 2024，Zhang et al.）——把 GSM8k 换题重出，模型在 GSM8k 与 GSM1k 之间掉分最高 8 个百分点，“模型生成 GSM8k 样例的可能性”与“过拟合差”Spearman 相关 0.36；Goodhart 通道论文（arXiv 2609.00064）——把一个模型往代理指标上微调，代理指标逼近天花板（差 0.5%）的同时 MMLU 从 0.371 掉到 0.279；“The Leaderboard Illusion”（Singh et al., 2025）——LMArena 争议里 Meta Llama-4 前有 27 个私有变体、Elo 虚高最多 112%。三条指向同一句：榜单分高 ≠ 我的应用变好。
- schema / JSON 遵守率随模型变化的量级：SO-Bench（CVPR 2026）——同任务同 schema，GPT-4o 指令跟随 92.36% 对结构化 API 99.67%，GPT-5 指令跟随 99.67%，Claude-3.5-Sonnet 无约束约 74.5%，GPT-4-0613 硬约不到 40%；schema 深度放大差距，GPT-5/Gemini-2.5-Pro 深度 >6 仍 >95%，小模型（Intern3.5-VL 4B）掉约 40 个点。ExtractBench（arXiv 2602.12247）——复杂 PDF→JSON 抽取里换用结构化输出 API 反而让整体 validity 从 51% 掉到 37%。换模型遵守率掉“几个点到几十个点”是常见量级，demo 那个反直觉断言不是编的。
- 微调成本量级（支撑“微调是最后一颗子弹”）：LoRA/QLoRA 一次实验约 $5–$50（Azure 口径 LoRA $50–$500）；7B LoRA 单卡 RTX 4090 约 $1.36–$2.72；70B LoRA 单卡 H100 约 $24–$48；QLoRA 13B 单卡 L4 约 $3.52–$7.04；full fine-tuning $100–$10,000+；managed API 微调 GPT-4o-mini 约 $50–$100、Claude 约 $100–$300；对齐训练（RLHF/DPO）方向性 $100–$10,000+/run 且要持续人工反馈喂 reward model。ICL 那头的账：几乎零起步，但每次请求都带 example tokens，高流量下（百万请求/月量级）重复 few-shot 的 token 成本能超过一次性 $100–$1,000 的微调。共识决策树：Prompt → Few-shot → RAG → LoRA/QLoRA → Full FT，微调排倒数第二。
- 漂移检测与阈值口径（方向性，按业务校准）：定义——Data drift（covariate shift）是 P(X) 漂移；Concept drift 是 P(Y|X) 漂移；Prior/label shift 是 P(Y) 漂移；Response drift（LLM 特有）是上游静默换权重/安全过滤器/系统 prompt 导致输出格式、语气、啰嗦度、拒答率变。检测方法——表格特征用 PSI / KL / KS / 卡方，高维用 embedding centroid 余弦距离 / MMD / classifier-based，LLM 特有信号用输出 token entropy/perplexity、schema 失败率跳变、LLM-as-judge 语义漂移归类、canary prompts（固定 golden prompt 定期跑）。阈值——PSI 0.25 显著、动；KL 0.5；KS p20%；schema 失败率 0.5%→5% 的跳变是“上游静默更新”的强信号。阈值必须按业务校准、绑定可测的性能/业务损失，否则误报。demo 里的
DRIFT_THRESHOLDS就是按这套口径写的（代码取严格>，即 0.10、0.25 这些边界值不触发告警），但具体取 0.10 还是 0.2 是你的决定。 - 模型退役周期：OpenAI——GA 模型至少 6 个月、specialized 变体至少 3 个月、preview 可短至约 2 周（案例：GPT-4 一代下线日 2026-10-23、Assistants API 2026-08-26）；Anthropic——至少 60 天通知 + 给建议用替代 + 定退休日（
claude-sonnet-4-202505142026-04-14 deprecated → 2026-06-15 retired，62 天；Claude Opus 4.1 2026-08-05 retired，约 61 天）；Azure AI Foundry——GA 60 天、preview 30 天，第三方托管模型生命周期缩短到 12 个月。典型 60 天到 6 个月，preview 短至 2 周。 - LLMOps 与 MLOps 的边界：LLMOps 在 MLOps 之上额外管四件事——Prompt 版本（prompts-as-code、进版本控制）、非确定性输出质量（无固定 ground truth，靠 LLM-as-judge / eval 套件 / human-in-the-loop）、在线人类反馈（点赞点踩回流 + RLHF）、Token 成本（成本从“训练算力”变“推理 token”）。口径来源多源一致：MLOps 核心产物=训练权重、评估=accuracy/F1/RMSE、成本=训练算力、迭代单元=重训；LLMOps 核心产物=prompt/chain/agent 配置、评估=相关性/忠实度/毒性/幻觉率、成本=推理 token、迭代单元=改一句 prompt 或换一个模型。说明：开篇 4.2 那句“LLMOps 还需额外管理 Prompt 版本、非确定性输出质量、在线人类反馈、Token 成本”，四个分项各自都有公开出处，但整句没有逐字引用，是本专栏对多家口径的归纳。
- 护栏版本化的边界：公开口径（AWS Well-Architected“Agentic AI Lens”+ policy-as-code 一派）是护栏/安全配置要按 policy-as-code 进 Git、加 semver + hash + changelog + owner、走 Draft → Review → Active → Archived 生命周期 + PR 审批 + CI 跑 golden set、drift detection + 不可变审计、金丝雀 1%→5%→25%→100% + kill-switch。但没有任何权威口径说“护栏必须和 prompt/data/model 塞进同一个发布原子”——于是“版本化 + 独立发布 + 先不塞进三元组”这个边界站得住：护栏版本化是共识，塞不塞进三元组是取舍。
附录 C：真实工具对接片段（版本已核实，2026-09）
demo 的六站逻辑一行不改，把两端换成真实工具即可。
# 模型注册：MLflow Model Registry 3.16.1 / Hugging Face Hub。
# 生产里 PartRegistry 的 model 那半 -> MLflow Model Registry；数据那半 -> DVC；Prompt -> Langfuse。
# import mlflow
# with mlflow.start_run():
# mlflow.log_params({"prompt_id": pid, "data_version": dv}) # 三件记成一次 run
# mlflow.set_tag("triple_fingerprint", triple.fingerprint()) # 三元组指纹当一次发布的身份
# mlflow.register_model(model_uri, name="x200-assistant")
# LoRA/适配器归档口径：HF Hub 用 push_adapter_to_hub() 存 adapter-only（10-200MB），
# adapter_config.json 可 pin base model revision —— 这正是三元组里 model 维要钉住的东西：
# 只存适配器 + 引用 base model，而不是把 base model 复制一份进你的仓库。
# Prompt 资产管理：Langfuse Prompt Management v4.38.0（v4 于 2026-08 发布）。
# 它每个 prompt 版本自动有 version ID + 标签（production/staging/latest），
# deploy = 移 production 标签、rollback = 移回旧标签、零重部署。
# from langfuse import Langfuse
# lf = Langfuse()
# lf.create_prompt(name="x200-system", prompt=PROMPT_V1, labels=["staging"])
# # 放量前把 production 标签移到新版本；回滚就是把标签移回去
# 边界：Langfuse 不支持内置 PR 审批流、也不替你切流量（放量比例在你代码里）——
# 所以本篇的"单一变更原则 + 评测门 + 灰度"这三样，工具不会替你做，得你自己在流水线里立。
# 数据版本：DVC 3.67.1（第 12 篇已核）——三元组的第二维从这儿取值，不重复说明。
# 网关：LiteLLM（第 5 篇已核）——按三元组的 model_id 路由 + fallback。
# 网关是"模型"这一维的切换开关，也是"换模型走发布不走升级"的执法点。
# 观测 / 成本 / 评测：Langfuse + OTel（第 8 篇）、RAGAS v0.4.3+/DeepEval 3.9.9（第 2 篇）。
# eval_gate -> 第 2 篇 harness；canary -> 第 3 篇灰度；drift_check -> 监控告警系统。
# trace 按三元组指纹打标这一条，在 Langfuse 里就是一个 tag：
# lf.trace(name="chat", metadata={"triple": triple.fingerprint()})
# 推理引擎（开篇 4.2 第⑤条，本篇只给边界）：vLLM 0.26.0 / SGLang 0.5.16 / TensorRT-LLM 1.2.1。
# 这些解决"推理成本/吞吐/显存"，跟发布纪律是两件事：换推理引擎不必重跑评测门
# （除非它改了输出分布），搭好发布流水线也不会自动省 token。
落到具体工具，就这一句：PartRegistry（模型侧）→ MLflow/HF Hub，PartRegistry（Prompt 侧）→ Langfuse，PartRegistry（数据侧）→ DVC，ReleaseStore → 一张发布账本；eval_gate / canary / drift_check 换成第 2/3 篇与监控告警——六站逻辑一行不改。
附录 D：demo 确定性评分桩、漂移检测与退役扫描的最小算法说明
demo 用的是确定性近似，纯标准库就够，不是真实评测、不是真实漂移检测，也不是真实注册中心。
- 为什么 mock 评分表能造出“更强的模型反而更差”：
_score_table按(model_id, prompt_id, case_id)显式写死结果——旧模型配任意 Prompt 全过；新模型配旧 Prompt 时，case_json_schema判schema_ok=False、case_refund_policy判passed=False。于是新模型pass_rate=0.50、schema_rate=0.75，旧模型都是 1.00，eval_gate判它为 regressed。为什么索引必须带 prompt_id：只按 model_id 索引，就表达不了“旧模型配新 Prompt”和“新模型配旧 Prompt”是两种不同行为分布——而“同一版 Prompt 在新模型上不是同一件事”正是本篇那把刀。这不是随机数凑的：随机数做不到“每次都拦得住”，断言就锁不住。 - 漂移检测为什么是纯函数：
drift_check(online_metrics, thresholds)不接 store、不写任何状态，超阈值就返回告警 + 建议动作。这不是偷懒，是纪律的代码化——监控只告警不自动改，自动改 = 绕开评测门，等于把门拆了。真实生产把这个纯函数换成你的监控告警系统（PSI / KL / KS / embedding 距离 / schema 失败率跳变），动作照旧是“发告警、让人从变更入口排发布”，不是“自动切指针”。 - 退役扫描为什么是一次
SELECT：triples_pinning(model_id)一条 SQL 扫triples表，ORDER BY CASE state WHEN 'serving' THEN 0 WHEN 'retained' THEN 1 ELSE 2 END让 serving 排最前——线上正在跑的先迁移。为什么不能只扫“当前生效的”：退役那天你要迁移的包括线上 serving 的和保留可回滚的旧三元组，只扫 serving 会漏掉回滚路径。为什么必须在账本里查：模型 ID 散在配置里就是考古，进了账本才是查询。 - 为什么
verified要单独一张表：set_serving只认verified，不认triples的状态字段。若把“已验证”并进triples的一个布尔列，一次误操作就能把没评过的标成评过的；分成两张表，lock()是唯一的写入路径（只有eval_gate会调它），守门员就没法被绕过。 - 为什么回滚要写
audit：rollback()除了把 serving 指回旧三元组，还往audit落一行（from_triple/to_triple/reason/operator）。指针拨回去只是状态，账才是历史——第 4 篇“rollback 不删历史、新增 revert 版本”那条心智，在发布层就是这张表。
🎯 更多专栏系列这篇可以查看博客主页📑 👍 若这篇对你有所触动，恳请**
就写这么多吧，内容比较基础，适合入门回顾。有补充的地方欢迎留言一起完善。
评论 (0)
暂无评论