架构定位

生产工程

用评测、可观测性、部署与证据证明系统在真实约束下可靠。

如何证明系统可靠，并完成部署、作品集和面试表达？

受这些层约束

概念校准

验证与评测

它是什么: 用测试、Schema、引用、状态和验收规则检查单次结果，并用数据集衡量系统表现。
它不是什么: 不等于让模型对自己的输出再给一次意见；验证必须落到外部可检查证据。

可观测性

它是什么: 通过结构化日志、Trace、指标和版本信息解释系统在一次运行中发生了什么。
它不是什么: 不只是保存终端输出；没有关联 ID、状态转换和敏感信息边界的日志难以诊断。

Harness

它是什么: 承载模型、工具、状态、权限、日志、预算、人工确认和恢复的运行控制环境。
它不是什么: 它能限制风险并保留证据，但不能直接阻止模型产生幻觉。

评测、可观测性、部署与可靠性

构建评测集、Tracing、质量成本指标、部署、回滚和事故演练。

本章只做一件事形成可部署、可监控、可回滚的 Forge 候选版本。

开始前，Forge 已经具备已完成 Module 15 的“多平台适配与迁移”，其通过验收的 solution 是本章起点。

完成后，Forge 将能够形成可观察的线上版本，覆盖质量、延迟、成本、失败和回滚。

卡住时的最小恢复点只运行本地评测和结构化追踪，先证明失败可定位。

下载本章实验仓库Python 3.12 · pytest · Pydantic · SQLite · deterministic Mock

阶段能力链

01定义带稳定 ID、输入和验收条件的评测案例，并阻止重复案例02从逐案例结果计算准确率、引用覆盖率和工具成功率，并保留分母03记录带父子关系的 Span，并确保一个运行中的追踪 ID 保持一致04按运行聚合成本和延迟，并对超预算结果给出可发布的门禁结论05将评测、迁移和安全检查绑定到不可变发布清单，缺失门禁不得晋级06比较基线与金丝雀错误率和延迟，在明确阈值下自动建议回滚07记录检测、缓解、恢复和复盘时间点，并验证恢复目标与证据完整性

每一步只增加一种可验证能力；后一步建立在前一步已经通过的代码和测试上。

概念校准\n\n在开始操作代码前，先把本章涉及的概念放回正确的工程边界。\n\n
\n\n### 验证与评测 evaluation\n\n- 它是什么: Verification 用测试、Schema、引用、状态或验收规则检查一次结果；Evaluation 用数据集和指标衡量多次运行表现。\n- 它不是什么: 它不是让模型对自己的回答再给一次意见；通过与否必须落到外部可检查证据。\n- 与相邻概念的关系: Agent Loop 依赖 Verification 决定停止，RAG 需要分层评测，Planning 需要节点验收，Observability 提供运行记录。\n- 在 HeatStack Forge 中的位置: Forge 为单次运行保存合同测试与 evidence，再用固定场景集比较质量、延迟、成本、权限和恢复能力。\n- 典型误用与修正: 误用是把一个综合分数当成全部结论。修正方法是定义分层指标、失败样本和版本基线，并结合产品风险评审。\n\n### 可观测性 observability\n\n- 它是什么: 通过结构化日志、Trace、指标、状态转换和版本信息解释一次运行发生了什么，并支持定位失败与比较版本。\n- 它不是什么: 它不只是终端输出，也不是无限记录所有输入；缺少关联 ID、状态转换或敏感信息边界的日志既难排查又可能泄露数据。\n- 与相邻概念的关系: Harness 产生运行事件，Agent Loop 与工具调用形成 Trace，Evaluation 聚合指标，Memory 与恢复依赖版本记录。\n- 在 HeatStack Forge 中的位置: Forge 为每次任务生成 operation_id，记录阶段、工具、耗时、成本、错误类别、状态变化和证据路径，同时过滤敏感信息。\n- 典型误用与修正: 误用是等失败后再临时加日志。修正方法是在合同设计时定义事件与关联字段，并用故障注入确认能重建时间线。\n\n### Harness harness\n\n- 它是什么: 承载模型、工具、状态、权限、日志、预算、人工确认、隔离和恢复的运行控制环境。\n- 它不是什么: 它可以缩小风险、拒绝动作并保留证据，但不能保证模型输出永远正确，也不能代替领域验收。\n- 与相邻概念的关系: Agent Loop 在 Harness 中运行；Tool、Memory 和 MCP 连接受其策略约束；Observability 记录过程，Evaluation 判断行为。\n- 在 HeatStack Forge 中的位置: Forge Harness 管理工作目录、工具、确认队列、预算、Checkpoint、审计日志和回滚，把模型建议与真实副作用分开。\n- 典型误用与修正: 误用是把系统提示词当成安全边界。修正方法是把权限、路径、网络、密钥、预算和回滚落实在模型之外的控制中。\n\n

STAGE 01

定义带稳定 ID、输入和验收条件的评测案例，并阻止重复案例

先看一个具体问题

评测案例需要稳定标识，否则重复案例会破坏追踪

你正在构建一个评测数据集，用来保存多个评测案例。每个案例包含一个输入和一个期望输出，用于验证系统行为。

如果两个案例使用相同的 case_id，后续统计和追踪就会混淆：你无法区分哪个输入对应哪个结果，甚至可能意外覆盖已有案例。

当前代码中 EvalDataset 的 validate() 只检查了版本和案例列表是否为空，没有检查 case_id 是否唯一。

因此，当测试 test_duplicate_case_ids_are_rejected 构造两个 case_id 都为 “same” 的案例时，validate() 不会抛出异常，导致测试失败。

先做判断

在 EvalDataset.validate() 中，除了检查版本和案例列表非空，还需要增加什么检查才能让重复案例 ID 被拒绝？

检查所有案例的 expected 字段是否非空
检查所有 case_id 是否唯一
检查 version 是否以 ‘v’ 开头
检查 input 字段是否包含特定键

判断依据: 正确答案是检查所有 case_id 是否唯一。因为测试 test_duplicate_case_ids_are_rejected 期望当两个案例具有相同 ID 时抛出包含 “unique” 的 ValueError。其他检查虽然可能有用，但不是本测试要求的。

它是什么

评测数据集是一个带版本、由唯一案例组成的集合

EvalDataset 包含一个版本字符串和一个案例元组，每个案例有唯一的 case_id、输入和期望输出。

validate() 方法负责在数据集被使用前检查其完整性：版本和案例非空、案例 ID 唯一、每个案例的 expected 非空。

唯一 ID 是评测案例的稳定标识，确保每个案例可以被独立追踪和引用。

它不是什么

评测数据集不执行评测，也不计算指标

它只负责存储和验证案例定义，不运行模型或比较输出。

它不保证案例内容在业务上正确，只保证结构上满足约束。

它不是一个数据库，不会自动持久化或提供查询接口。

它与相邻概念的关系

EvalCase 与 EvalDataset 的关系

EvalCase 是单个评测案例，包含 case_id、input 和 expected。

EvalDataset 聚合多个 EvalCase，并添加版本信息。

validate() 方法遍历所有案例，检查 ID 唯一性和 expected 非空，确保数据集可用。

本阶段的边界决定: 当数据集包含重复 case_id 时，validate() 必须抛出 ValueError，因为重复 ID 会导致评测结果无法追踪。

实验 ID: evaluation-observability-deployment-01-versioned-evaluation-dataset
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/evaluation-observability-deployment/01-versioned-evaluation-dataset/
验收: 待验收

教师示范

目标: 演示如何创建 EvalCase 和 EvalDataset 并调用 validate() 通过有效案例测试。

首先，我们创建一个有效的 EvalCase，它有一个唯一的 case_id、一个输入字典和一个期望输出字典。

然后，我们将这个案例放入 EvalDataset 中，并指定版本号。

最后，调用 validate() 方法，它应该不抛出任何异常，表示数据集有效。

实施步骤

定义 EvalCase 实例：case = EvalCase(“case-1”, {“q”: “x”}, {“status”: “ok”})
定义 EvalDataset 实例：dataset = EvalDataset(“v1”, (case,))
调用 dataset.validate()，确认无异常抛出。

真实 solution 代码: forge/reliability/__init__.py

"""Evaluation and production reliability primitives."""

真实 solution 代码: forge/reliability/dataset.py

from dataclasses import dataclass

@dataclass(frozen=True)
class EvalCase:
    case_id: str
    input: dict[str, object]
    expected: dict[str, object]

@dataclass(frozen=True)
class EvalDataset:
    version: str
    cases: tuple[EvalCase, ...]

    def validate(self) -> None:
        ids = [case.case_id for case in self.cases]
        if not self.version.strip() or not self.cases:
            raise ValueError("dataset version and cases are required")
        if len(ids) != len(set(ids)):
            raise ValueError("evaluation case IDs must be unique")
        if any(not case.expected for case in self.cases):
            raise ValueError("every case needs acceptance data")

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

EvalCase 使用 dataclass 定义，自动生成 init 和 eq 等方法。

EvalDataset 的 validate() 方法目前只检查版本和案例非空，因此有效案例可以通过。

这个测试验证了基本的数据集创建和验证流程。

带提示练习

补全 validate() 中的重复 ID 检查，使 test_duplicate_case_ids_are_rejected 通过。

打开 forge/reliability/dataset.py 文件。
在 validate() 方法中，在检查版本和案例非空之后，添加检查 case_id 唯一性的代码。
提示：可以使用列表推导式提取所有 case_id，然后比较列表长度和集合长度。
如果发现重复，抛出 ValueError，消息中包含 “unique” 字样。
运行测试确认通过。

逐级提示

使用 [case.case_id for case in self.cases] 获取所有 ID。
使用 len(ids) != len(set(ids)) 判断是否有重复。
抛出 ValueError(“evaluation case IDs must be unique”)。

完成标准

test_duplicate_case_ids_are_rejected 测试通过
validate() 在遇到重复 ID 时抛出包含 “unique” 的 ValueError

故障注入与排查

当前代码中 validate() 没有检查 case_id 唯一性。运行测试 test_duplicate_case_ids_are_rejected 时，两个具有相同 ID 的案例被接受，导致测试失败并显示 AssertionError: duplicate evaluation cases were accepted。

本阶段的目标就是实现重复 ID 检查。你已经学习了如何提取 ID 并比较集合大小，现在可以定位到 validate() 方法中缺少的检查，并添加修复代码。

已验证的故障合同

目标测试: test_duplicate_case_ids_are_rejected
故障文件: forge/reliability/dataset.py

诊断顺序

运行测试命令 python -m pytest -q ../tests/test_stage.py::test_duplicate_case_ids_are_rejected，观察失败输出。
查看测试代码，发现它构造了两个 case_id 都为 “same” 的案例，并期望 validate() 抛出 ValueError。
检查 forge/reliability/dataset.py 中的 validate() 方法，发现它只检查了版本和案例非空，没有检查 ID 唯一性。
因此，当调用 validate() 时没有异常，测试中的 else 分支执行，抛出 AssertionError。

修复原则: 在 validate() 方法中添加 ID 唯一性检查：提取所有 case_id 到列表 ids，然后比较 len(ids) 和 len(set(ids))，如果不相等则抛出 ValueError(“evaluation case IDs must be unique”)。完成修复后运行 python -m pytest -q ../tests/test_stage.py::test_duplicate_case_ids_are_rejected，并确认 test_duplicate_case_ids_are_rejected 从失败变为通过。

python -m pytest -q ../tests/test_stage.py::test_duplicate_case_ids_are_rejected

迁移挑战

使用办公场景数据创建包含三个不同案例的 EvalDataset，确保版本为 ‘office-v1’ 且每个案例有唯一 ID 和非空 expected。

改变的条件: 将案例数据从通用示例改为办公场景，并指定版本为 ‘office-v1’，同时保持三个案例且 ID 唯一。

编写一个 Python 脚本或交互式代码，创建三个 EvalCase 实例，分别使用办公场景的输入和期望输出，例如：案例1：输入 {‘query’: ‘总结会议纪要’}，期望 {‘summary’: ‘…’}；案例2：输入 {‘query’: ‘提取行动项’}，期望 {‘actions’: […]}；案例3：输入 {‘query’: ‘生成报告’}，期望 {‘report’: ‘…’}。确保每个 case_id 唯一，例如 ‘office-1’, ‘office-2’, ‘office-3’。然后将它们放入 EvalDataset，版本设为 ‘office-v1’，并调用 validate() 确认无异常。

完成标准

创建的 EvalDataset 版本为 ‘office-v1’，包含三个案例，每个案例的 case_id 唯一且 expected 非空。
调用 validate() 不抛出异常，并且打印或返回成功信息。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么评测案例需要稳定 ID？如果两个案例有相同 ID 会发生什么？

一个有力回答应包含

稳定 ID 确保每个评测案例可以被唯一标识和追踪，避免结果混淆。
如果两个案例有相同 ID，评测结果无法区分是哪个输入产生的，可能导致数据覆盖或统计错误。

掌握证据

通过 test_valid_dataset_has_versioned_cases 和 test_duplicate_case_ids_are_rejected 两个测试。
能够解释为什么需要唯一 ID：防止评测结果混淆和意外覆盖。
在 validate() 中实现了 ID 唯一性检查，使用集合比较长度。

STAGE 02

从逐案例结果计算准确率、引用覆盖率和工具成功率，并保留分母

先看一个具体问题

如何从多个案例结果中汇总出有意义的整体质量指标？

你已经有了 CaseResult 数据类，它记录单个案例的通过状态、引用要求和工具调用情况，但还没有一个函数能把这些案例结果汇总成整体质量指标。

如果直接手工计算准确率和引用覆盖率，不仅容易出错，而且当某个指标没有适用案例时（例如没有引用要求），分母为零会导致除零错误或错误的零分。

本阶段需要实现 summarize() 函数，它接收一个 CaseResult 元组，返回包含案例数、准确率、引用覆盖率和工具成功率的字典，并且正确处理分母为零的情况。

先做判断

在实现 summarize() 之前，请预测：当某个指标的分母为零（例如没有任何引用要求）时，该指标应该返回什么值？

返回 0.0，因为没有成功案例
返回 1.0，表示该指标不适用，视为成功
抛出异常，因为无法计算
返回 None，表示缺失

判断依据: 正确答案是返回 1.0。分母为零意味着该指标没有适用的案例，因此不应该惩罚系统；这种“空真”语义在质量指标中很常见，例如没有引用要求时引用覆盖率应视为完全满足。

它是什么

质量指标汇总是什么

质量指标汇总是把多个案例的离散结果（通过/失败、引用有效数、工具成功数）聚合成整体比率的过程，它保留了每个指标的分母（适用案例总数），从而反映系统在相关场景下的表现。

summarize() 函数接收一个 CaseResult 元组，返回一个字典，其中 accuracy 是正确案例的比例，citation_coverage 是有效引用占所需引用的比例，tool_success 是成功工具调用占尝试调用的比例。

它不是什么

质量指标汇总不是什么

它不是对单个案例的详细审查，也不是对模型输出的主观评价；它只基于 CaseResult 中已经记录的结构化字段进行算术聚合。

它不涉及追踪、预算或任何运行时上下文，也不负责判断案例是否应该被包含；这些决策由上游的评测流程完成。

它与相邻概念的关系

与其他概念的关系

CaseResult 是输入数据类，summarize() 消费它并产生指标字典；ratio() 是辅助函数，用于安全地计算比率，处理分母为零的情况。

这些指标为后续的可观测性面板和评测报告提供数据，但它们本身不产生日志或追踪信息。

本阶段的边界决定: 当分母为零时，指标应返回 1.0（空真），而不是 0.0 或抛出异常，因为零适用案例意味着没有失败，不应惩罚系统。

实验 ID: evaluation-observability-deployment-02-quality-metrics
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/evaluation-observability-deployment/02-quality-metrics/
验收: 待验收

教师示范

目标: 演示 summarize() 如何从两个 CaseResult 计算准确率和引用覆盖率。

假设你有两个案例：第一个案例通过，需要 2 个引用且都有效，尝试 1 个工具且成功；第二个案例失败，需要 1 个引用且有效，尝试 1 个工具但失败。

我们调用 summarize() 并观察返回的字典，其中准确率应为 0.5（一个通过一个失败），引用覆盖率应为 2/3（总共 3 个所需引用中 2 个有效）。

实施步骤

创建两个 CaseResult 实例，分别传入相应的参数。
调用 summarize((case1, case2))。
检查返回字典中的 accuracy 和 citation_coverage 值。

真实 solution 代码: forge/reliability/metrics.py

from dataclasses import dataclass

@dataclass(frozen=True)
class CaseResult:
    passed: bool
    citations_required: int = 0
    citations_valid: int = 0
    tools_attempted: int = 0
    tools_succeeded: int = 0

def ratio(numerator: int, denominator: int) -> float:
    return 1.0 if denominator == 0 else numerator / denominator

def summarize(results: tuple[CaseResult, ...]) -> dict[str, float | int]:
    if not results:
        raise ValueError("results are required")
    return {
        "cases": len(results),
        "accuracy": ratio(sum(item.passed for item in results), len(results)),
        "citation_coverage": ratio(sum(item.citations_valid for item in results), sum(item.citations_required for item in results)),
        "tool_success": ratio(sum(item.tools_succeeded for item in results), sum(item.tools_attempted for item in results)),
    }

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

accuracy 计算为通过案例数除以总案例数：1/2 = 0.5。

citation_coverage 计算为有效引用总数除以所需引用总数：(1+1)/(2+1) = 2/3 ≈ 0.6667。

tool_success 计算为成功工具调用总数除以尝试工具调用总数：(1+0)/(1+1) = 0.5。

带提示练习

修正 ratio() 函数，使分母为零时返回 1.0，让 test_not_applicable_metric_is_not_failure 通过。

打开 forge/reliability/metrics.py，找到 ratio() 函数。
当前实现是 return numerator / denominator if denominator else 0.0，请将其改为当 denominator == 0 时返回 1.0。
运行测试 python -m pytest -q tests/test_stage.py::test_not_applicable_metric_is_not_failure 验证修复。

逐级提示

提示 1：使用条件表达式或 if 语句检查分母是否为零。
提示 2：返回 1.0 而不是 0.0，因为零适用案例应视为成功。
提示 3：确保 summarize() 使用修正后的 ratio()。

完成标准

ratio(0, 0) 返回 1.0。
test_not_applicable_metric_is_not_failure 测试通过。

故障注入与排查

在故障实验室中，我们提供了一个有缺陷的 forge/reliability/metrics.py 版本，其中 ratio() 函数在分母为零时返回 0.0。这导致 test_not_applicable_metric_is_not_failure 测试失败，因为一个没有引用要求的案例被错误地报告为引用覆盖率为 0。

本阶段你已经理解了空真语义，并知道如何修改 ratio() 来处理分母为零的情况。通过诊断这个故障，你将巩固这一概念，并学会如何定位和修复类似的指标计算错误。

已验证的故障合同

目标测试: test_not_applicable_metric_is_not_failure
故障文件: forge/reliability/metrics.py

诊断顺序

运行故障命令 python -m pytest -q ../tests/test_stage.py::test_not_applicable_metric_is_not_failure，观察断言失败：assert 0.0 == 1.0。
检查 forge/reliability/metrics.py 中的 ratio() 函数，发现当分母为零时返回 0.0。
理解测试期望：对于没有引用要求的案例，引用覆盖率应为 1.0，因为该指标不适用，不应惩罚系统。
修复 ratio() 函数，使其在分母为零时返回 1.0。

修复原则: 将 ratio() 函数中的 return numerator / denominator if denominator else 0.0 改为 return 1.0 if denominator == 0 else numerator / denominator。这样，当分母为零时返回 1.0，表示空真成功。

python -m pytest -q ../tests/test_stage.py::test_not_applicable_metric_is_not_failure

迁移挑战

现在使用音乐场景数据来测试你的理解。创建三个 CaseResult 实例，其中一个没有引用要求（即 citations_required=0），另外两个有引用要求且有效引用数等于所需数。

改变的条件: 将案例数据从软件交付场景改为音乐创作场景，并调整引用要求以包含一个不适用案例。

在 Python 脚本中创建三个 CaseResult 实例：第一个案例通过且没有引用要求（CaseResult(True)），第二个案例通过且需要 2 个引用、2 个有效引用，第三个案例通过且需要 1 个引用、1 个有效引用。然后调用 summarize() 并打印结果。

完成标准

打印的字典中 citation_coverage 应为 1.0，因为所有适用案例的引用都有效，且不适用案例不影响覆盖率。
cases 应为 3，accuracy 应为 1.0（所有案例都通过）。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么分母为零时指标应视为成功？请解释这种空真语义在质量指标中的合理性，并说明它与返回 0.0 的区别。

一个有力回答应包含

分母为零意味着没有适用案例，因此该指标没有失败的机会，不应惩罚系统。
返回 1.0 表示完全满足要求，而返回 0.0 会错误地暗示系统在该指标上表现不佳。
空真语义在逻辑和数学中常见，例如空集合上的全称命题为真。

掌握证据

通过 test_metrics_use_case_level_counts 和 test_not_applicable_metric_is_not_failure 两个测试。
能够解释为什么分母为零时指标应返回 1.0（空真成功），而不是 0.0。
在 forge/reliability/metrics.py 中正确实现了 ratio() 和 summarize() 函数，并保留了案例计数。

STAGE 03

记录带父子关系的 Span，并确保一个运行中的追踪 ID 保持一致

先看一个具体问题

一次运行中的 Span 必须共享同一个 Trace ID

当前 forge/reliability/ 目录中只有 metrics.py，还没有 tracing.py，因此测试文件 tests/test_stage.py 在导入 Span 和 validate_trace 时会直接失败。

你需要从零创建 forge/reliability/tracing.py，定义 Span 数据类和 validate_trace 函数，使两个测试通过。

test_valid_parent_child_trace 要求接受两个共享 trace_id="t1" 且父子关系正确的 Span；test_cross_trace_child_is_rejected 要求拒绝一个来自不同 trace 的子 Span。

如果只检查 Span ID 唯一性而忽略 trace ID 一致性，跨 trace 的子 Span 会被错误接受，导致追踪数据被污染。

先做判断

在实现 validate_trace 时，除了检查 Span ID 唯一性，为什么还必须检查所有 Span 的 trace_id 是否一致？

因为 trace ID 是追踪的唯一标识，不一致会导致无法关联同一运行的所有 Span。
因为 trace ID 必须与 Span ID 相同。
因为 trace ID 只用于排序，不影响追踪正确性。
因为 trace ID 可以随意设置，不需要检查。

判断依据: 正确选项是第一个。Trace ID 用于标识一次完整运行，所有属于该运行的 Span 必须共享同一个 trace ID，否则无法重建调用链。其他选项误解了 trace ID 的作用。

它是什么

Trace 是一棵由 Span 组成的树

一次运行对应一个 Trace，由唯一的 trace_id 标识；Trace 内的每个操作记录为一个 Span，Span 之间通过 parent_id 形成父子关系。

validate_trace 负责检查 Trace 的结构完整性：Span ID 必须唯一、所有 Span 的 trace_id 必须一致、每个 parent_id 必须指向已存在的 Span、持续时间不能为负。

它不是什么

Trace 验证不涉及指标或预算

validate_trace 只检查结构和 ID 一致性，不计算任何指标（如总时长、平均延迟），也不检查预算或资源消耗。

它不负责生成 Span，也不负责将 Span 发送到任何存储或可视化系统；它只是对已生成的 Span 集合进行静态校验。

它与相邻概念的关系

Span 与 Trace 的关系

一个 Trace 包含多个 Span，每个 Span 记录一个操作（如 run、tool.call）及其持续时间。

根 Span 的 parent_id 为 None，其他 Span 的 parent_id 必须指向同一 Trace 内的某个 Span ID。

validate_trace 通过检查 trace_id 集合大小是否为 1 来强制所有 Span 属于同一 Trace，通过检查 parent_id 是否在 Span ID 集合中来保证父子关系有效。

本阶段的边界决定: 当发现跨 Trace 的子 Span 时，validate_trace 必须抛出 ValueError，因为接受它会导致追踪数据无法正确关联，后续分析会出错。

实验 ID: evaluation-observability-deployment-03-correlated-tracing
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/evaluation-observability-deployment/03-correlated-tracing/
验收: 待验收

教师示范

目标: 演示创建两个 Span 并调用 validate_trace() 通过有效父子追踪测试。

首先定义 Span 数据类，包含 trace_id、span_id、parent_id、name 和 duration_ms 五个字段。

然后实现 validate_trace 的基础检查：空 Trace 报错、Span ID 唯一性检查。

最后创建两个 Span：根 Span Span("t1", "root", None, "run", 20) 和子 Span Span("t1", "tool", "root", "tool.call", 8)，调用 validate_trace 应无异常。

实施步骤

创建 forge/reliability/tracing.py 文件。
定义 Span 数据类，使用 @dataclass(frozen=True) 使其不可变。
实现 validate_trace 函数，先检查空 Trace，再检查 Span ID 唯一性。
在测试中构造两个共享 trace_id="t1" 的 Span，调用 validate_trace 验证通过。

真实 solution 代码: forge/reliability/tracing.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Span:
    trace_id: str
    span_id: str
    parent_id: str | None
    name: str
    duration_ms: int

def validate_trace(spans: tuple[Span, ...]) -> None:
    if not spans:
        raise ValueError("trace is empty")
    if len({span.trace_id for span in spans}) != 1:
        raise ValueError("trace IDs must match")
    ids = {span.span_id for span in spans}
    if len(ids) != len(spans):
        raise ValueError("span IDs must be unique")
    for span in spans:
        if span.parent_id is not None and span.parent_id not in ids:
            raise ValueError("span parent is missing")
        if span.duration_ms < 0:
            raise ValueError("duration cannot be negative")

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

Span 数据类使用 frozen=True 确保 Span 创建后不可修改，避免意外更改追踪数据。

validate_trace 首先检查 Span ID 唯一性，防止重复 ID 导致父子关系混乱。

在有效父子追踪中，两个 Span 的 trace_id 相同，且子 Span 的 parent_id 指向根 Span 的 span_id，因此验证通过。

带提示练习

在 validate_trace() 中添加 trace ID 一致性检查，使 test_cross_trace_child_is_rejected 通过。

在 validate_trace 中，在检查 Span ID 唯一性之前或之后，添加检查所有 Span 的 trace_id 是否一致的代码。
使用集合推导式 {span.trace_id for span in spans} 获取所有 trace ID，如果集合大小不为 1，则抛出 ValueError("trace IDs must match")。
运行测试 python -m pytest -q tests/test_stage.py::test_cross_trace_child_is_rejected，确认测试通过。

逐级提示

提示：集合大小不为 1 意味着存在多个不同的 trace ID。
提示：错误消息必须包含 “trace IDs”，因为测试断言了这一点。
提示：检查顺序不影响正确性，但建议先检查 trace ID 一致性再检查 Span ID 唯一性。

完成标准

validate_trace 在遇到不同 trace ID 的 Span 时抛出 ValueError。
test_cross_trace_child_is_rejected 测试通过。

故障注入与排查

故障夹具 fixtures/failure.json 中注入了一个缺陷：validate_trace 只检查了 Span ID 唯一性，没有检查 trace ID 一致性。当测试 test_cross_trace_child_is_rejected 运行时，传入两个 Span：Span("t1","root",None,"run",20) 和 Span("t2","tool","root","tool.call",8)，它们的 trace ID 不同，但函数没有抛出异常，导致测试失败。

本阶段的核心是理解并实现 trace ID 一致性检查。通过分析故障夹具中的代码，学习者可以看到缺少该检查的后果，并能够根据测试要求添加正确的检查逻辑。

已验证的故障合同

目标测试: test_cross_trace_child_is_rejected
故障文件: forge/reliability/tracing.py

诊断顺序

运行故障命令 python -m pytest -q ../tests/test_stage.py::test_cross_trace_child_is_rejected，观察测试失败，错误为 AssertionError: cross-trace span was accepted。
查看 forge/reliability/tracing.py 中的 validate_trace 实现，发现只有 Span ID 唯一性检查，没有 trace ID 一致性检查。
理解测试意图：当 Span 的 trace ID 不同时，应该抛出 ValueError，且错误消息包含 “trace IDs”。
在 validate_trace 中添加检查：if len({span.trace_id for span in spans}) != 1: raise ValueError("trace IDs must match")。

修复原则: 在 validate_trace 函数中，在检查 Span ID 唯一性之前添加 trace ID 一致性检查：使用集合推导式获取所有 trace ID，如果集合大小不为 1，则抛出 ValueError("trace IDs must match")。这样，当传入不同 trace ID 的 Span 时，函数会正确拒绝。

python -m pytest -q ../tests/test_stage.py::test_cross_trace_child_is_rejected

迁移挑战

办公研究场景：你正在处理一份文档分析任务，需要记录三个 Span：根 Span 表示整个运行，工具 Span 表示文档解析，LLM Span 表示摘要生成。所有 Span 共享同一个 trace_id ‘office-trace-1’，并形成正确的父子关系。

改变的条件: 从两个 Span 的父子追踪扩展到三个 Span 的办公场景，并引入新的 Span 名称（doc.parse, llm.summarize）和不同的持续时间，同时保持 trace_id 一致性和父子关系正确性。

在 Python 交互式环境或单独脚本（如 office_trace.py）中，创建三个 Span 对象：根 Span Span("office-trace-1", "root", None, "run", 30)，工具 Span Span("office-trace-1", "tool", "root", "doc.parse", 12)，LLM Span Span("office-trace-1", "llm", "tool", "llm.summarize", 18)。然后调用 validate_trace 验证这三个 Span 是否构成有效追踪。

完成标准

运行 python -c "from forge.reliability.tracing import Span, validate_trace; validate_trace((Span('office-trace-1','root',None,'run',30), Span('office-trace-1','tool','root','doc.parse',12), Span('office-trace-1','llm','tool','llm.summarize',18)))" 不抛出异常。
运行后无异常，且三个 Span 的 trace_id 均为 office-trace-1，父子关系为 root->tool->llm。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么 validate_trace 必须检查所有 Span 的 trace_id 是否一致？如果省略这个检查，会对追踪系统造成什么具体后果？

一个有力回答应包含

Trace ID 用于标识一次完整运行，所有属于该运行的 Span 必须共享同一个 trace ID，否则无法将分散的 Span 关联起来。
如果省略检查，来自不同运行的 Span 可能被错误地合并到同一个 Trace 中，导致调用链混乱，后续分析（如延迟统计、错误定位）会得出错误结论。
在分布式系统中，跨 Trace 的 Span 混入会破坏追踪的完整性，使得无法准确重建请求路径。

掌握证据

通过 test_valid_parent_child_trace 和 test_cross_trace_child_is_rejected 两个测试，证明 validate_trace 能正确处理有效追踪和跨 Trace 子 Span。
能够解释 Trace ID 一致性的必要性：所有 Span 必须共享同一个 trace ID，否则追踪数据无法关联。
在办公场景的转移挑战中，成功创建三个 Span 并验证其有效性，展示了在不同场景下应用追踪验证的能力。

STAGE 04

按运行聚合成本和延迟，并对超预算结果给出可发布的门禁结论

先看一个具体问题

如何判断一次运行是否在成本与延迟预算内？

在发布前，我们需要一个明确的预算检查函数，它接收实际成本、实际延迟和预算对象，返回超预算的维度名称。

当前代码库中还没有 forge/reliability/budgets.py 文件，因此测试在导入时就会失败。

测试 test_over_budget_dimensions_are_named 期望当成本 1.5 超过预算 1.0 时返回 ('cost',)，而 test_exact_budget_is_allowed 期望当成本 1.0 和延迟 1000 恰好等于预算时返回空元组。

你需要创建该文件，定义 Budget 数据类和 check_budget 函数，并确保比较逻辑正确。

先做判断

如果实际成本恰好等于预算上限，check_budget 应该返回什么？

返回包含 ‘cost’ 的元组，因为达到上限就是超预算
返回空元组，因为等于上限仍在允许范围内
返回包含 ‘latency’ 的元组，因为延迟也可能等于上限
抛出异常，因为等于上限是边界情况

判断依据: 预算上限是最大允许值，等于上限意味着没有超过，因此不应被视为违规。返回空元组表示所有维度都在预算内。

它是什么

预算检查是什么

预算检查是一个纯函数，它比较实际测量值与预算上限，并返回所有超预算维度的名称。

它只负责数值比较，不涉及发布门禁或回滚决策。

当实际值严格大于上限时才视为违规，等于上限是允许的。

它不是什么

预算检查不是什么

它不是发布门禁系统，不会自动阻止部署或触发回滚。

它不处理负数输入，负数测量值应被视为无效并抛出异常。

它不修改预算对象或测量值，只返回违规维度。

它与相邻概念的关系

与其他组件的关系

Budget 数据类定义预算上限，check_budget 使用这些上限进行比较。

测试文件 tests/test_stage.py 导入 Budget 和 check_budget，验证其行为。

该模块独立于 tracing 模块，但未来可能被发布流程调用。

本阶段的边界决定: 预算检查只比较数值，不涉及发布门禁或回滚；负数输入应抛出异常，等于上限视为允许。

实验 ID: evaluation-observability-deployment-04-cost-latency-budgets
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/evaluation-observability-deployment/04-cost-latency-budgets/
验收: 待验收

教师示范

目标: 演示 check_budget() 如何识别超预算维度。

假设我们有一个预算 Budget(max_cost=1.0, max_latency_ms=1000)，实际成本为 1.5，延迟为 900。

成本 1.5 大于上限 1.0，因此应返回 ('cost',)；延迟 900 小于上限 1000，不违规。

我们实现 check_budget 函数，使用严格大于比较，并返回违规维度元组。

实施步骤

创建 forge/reliability/budgets.py 文件。
定义 Budget 数据类，包含 max_cost 和 max_latency_ms 字段。
实现 check_budget 函数，检查成本是否大于 budget.max_cost，若是则添加 ‘cost’ 到违规列表。
检查延迟是否大于 budget.max_latency_ms，若是则添加 ‘latency’。
返回违规元组。

真实 solution 代码: forge/reliability/budgets.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Budget:
    max_cost: float
    max_latency_ms: int

def check_budget(cost: float, latency_ms: int, budget: Budget) -> tuple[str, ...]:
    violations: list[str] = []
    if cost < 0 or latency_ms < 0:
        raise ValueError("measurements cannot be negative")
    if cost > budget.max_cost:
        violations.append("cost")
    if latency_ms > budget.max_latency_ms:
        violations.append("latency")
    return tuple(violations)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

当成本 1.5 大于预算 1.0 时，违规列表包含 ‘cost’，返回 ('cost',)。

延迟 900 小于 1000，不添加 ‘latency’，因此结果正确。

带提示练习

修正比较运算符，使等于预算时返回空元组，让 test_exact_budget_is_allowed 通过。

打开 forge/reliability/budgets.py。
检查 check_budget 函数中的比较运算符。
将 >= 改为 >，确保等于上限时不违规。
运行测试 python -m pytest -q tests/test_stage.py::test_exact_budget_is_allowed。

逐级提示

注意当前代码可能使用了 >=，这会导致等于上限时也被视为违规。
预算上限是最大允许值，等于上限应该被允许。
修改后，当成本等于 budget.max_cost 时，不应添加 ‘cost’。

完成标准

test_exact_budget_is_allowed 测试通过。
check_budget(1.0, 1000, Budget(1.0, 1000)) 返回空元组。

故障注入与排查

在 fixtures/failure.json 中，test_exact_budget_is_allowed 测试失败，因为 check_budget 使用了 >= 比较，导致恰好等于预算的成本和延迟被错误地标记为违规。你需要检查 forge/reliability/budgets.py 中的比较运算符并修复该缺陷。

本阶段已经学习了预算检查的边界条件：等于上限应视为允许。通过检查 forge/reliability/budgets.py 中的比较运算符，可以定位并修复这个缺陷。

已验证的故障合同

目标测试: test_exact_budget_is_allowed
故障文件: forge/reliability/budgets.py

诊断顺序

运行测试 python -m pytest -q ../tests/test_stage.py::test_exact_budget_is_allowed，观察失败信息。
查看 forge/reliability/budgets.py 中的 check_budget 函数。
发现比较运算符使用了 >=，导致等于上限时也被视为违规。
将 >= 改为 >，使等于上限时不再违规。

修复原则: 将 check_budget 函数中的 cost >= budget.max_cost 改为 cost > budget.max_cost，将 latency_ms >= budget.max_latency_ms 改为 latency_ms > budget.max_latency_ms。

python -m pytest -q ../tests/test_stage.py::test_exact_budget_is_allowed

迁移挑战

办公研究场景中，一次运行的成本为 2.0 美元，延迟为 1500 毫秒，预算上限分别为 2.0 和 1500。

改变的条件: 将预算上限从 1.0 和 1000 改为 2.0 和 1500，同时保持成本等于预算上限的边界情况。

定义 Budget(max_cost=2.0, max_latency_ms=1500)，调用 check_budget(2.0, 1500, budget)，并验证返回空元组。

完成标准

check_budget(2.0, 1500, Budget(2.0, 1500)) 返回空元组。
check_budget(2.0, 1500, Budget(2.0, 1500)) 不抛出异常。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么等于预算应被允许？如果成本或延迟为负数，check_budget 应该做什么？

一个有力回答应包含

等于预算意味着没有超过最大允许值，因此不应被视为违规。
负数测量值无效，应抛出 ValueError 异常。

掌握证据

通过 test_over_budget_dimensions_are_named 和 test_exact_budget_is_allowed 两个测试。
能够解释为什么等于预算上限是允许的，而不是违规。
在 forge/reliability/budgets.py 中实现了 Budget 数据类和 check_budget 函数，使用严格大于比较。

STAGE 05

将评测、迁移和安全检查绑定到不可变发布清单，缺失门禁不得晋级

先看一个具体问题

发布候选版本缺少安全门禁却被晋级

当前 forge/reliability/deployment.py 文件尚未创建，因此 tests/test_stage.py 中的两个测试在导入阶段就会失败，报错 ModuleNotFoundError: No module named 'forge.reliability.deployment'。

即使我们创建了文件并实现一个简单的 promote 函数，只要它只检查版本号存在就返回晋级字符串，那么 test_missing_security_gate_blocks_promotion 仍然会失败，因为缺少安全门禁的清单会被错误地标记为已晋级。

你需要实现一个不可变的 ReleaseManifest 数据类和一个 promote 函数，该函数必须验证发布身份（版本号非空且 artifact SHA256 长度为 64），并检查所有必需门禁（tests、evaluation、security、migration）都已通过，否则抛出包含缺失门禁名称的 ValueError。

先做判断

在实现 promote 函数时，如果只检查 manifest.version 非空就返回晋级字符串，会发生什么？

两个测试都会通过，因为版本号存在就足够晋级。
test_complete_release_is_promoted 会通过，但 test_missing_security_gate_blocks_promotion 会失败，因为缺少安全门禁的清单被错误晋级。
两个测试都会失败，因为缺少安全门禁时应该抛出异常。
test_missing_security_gate_blocks_promotion 会通过，但 test_complete_release_is_promoted 会失败。

判断依据: 正确答案是第二个选项。只检查版本号存在无法阻止缺少安全门禁的清单晋级，因此第二个测试会失败。你需要比较 passed_gates 与必需门禁集合，并在缺失时抛出 ValueError。

它是什么

发布晋级门禁是什么

发布晋级门禁是一个不可变的发布清单（ReleaseManifest）与一组必需门禁（REQUIRED_GATES）之间的集合比较过程。

promote 函数首先验证发布身份（版本号非空且 artifact SHA256 长度为 64），然后计算缺失门禁 = 必需门禁 - 已通过门禁，如果缺失集合非空则抛出包含缺失门禁名称的 ValueError，否则返回以 promoted: 开头的晋级字符串。

它不是什么

发布晋级门禁不是什么

发布晋级门禁不执行实际部署、回滚或任何基础设施操作；它只检查门禁集合是否完整。

它也不是对发布质量的全面评估，而是仅验证必需门禁是否全部通过，不关心门禁内部的具体指标或分数。

它与相邻概念的关系

与其他概念的关系

发布晋级门禁依赖于 Stage 04 的预算模块所建立的可靠性检查思想，但专注于发布前的门禁集合验证。

它与可观测性相关，因为晋级决策需要基于可观测的运行记录和测试结果，而不是模型的主观判断。

它与 Harness 的隔离和权限控制互补：Harness 负责运行时安全，而发布晋级门禁负责发布前的静态门禁检查。

本阶段的边界决定: 发布晋级只检查门禁集合，不执行实际部署或回滚；如果缺少任何必需门禁，必须抛出异常阻止晋级。

实验 ID: evaluation-observability-deployment-05-release-promotion-gates
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/evaluation-observability-deployment/05-release-promotion-gates/
验收: 待验收

教师示范

目标: 演示创建包含所有必需门禁的 ReleaseManifest 并调用 promote() 成功晋级。

我们首先定义 REQUIRED_GATES 常量，包含 tests、evaluation、security、migration 四个门禁。

然后创建 ReleaseManifest 数据类，使用 frozen=True 使其不可变，包含 version、artifact_sha256 和 passed_gates 字段。

在 promote 函数中，先验证版本号非空且 artifact SHA256 长度为 64，否则抛出 ValueError。

接着计算缺失门禁，如果缺失集合非空则抛出包含缺失门禁名称的 ValueError，否则返回晋级字符串。

实施步骤

创建 forge/reliability/deployment.py 文件。
定义 REQUIRED_GATES = frozenset({"tests", "evaluation", "security", "migration"})。
定义 ReleaseManifest 数据类，使用 @dataclass(frozen=True) 装饰器。
实现 promote 函数，包含身份验证和门禁检查逻辑。
运行测试命令验证两个测试均通过。

真实 solution 代码: forge/reliability/deployment.py

from dataclasses import dataclass

REQUIRED_GATES = frozenset({"tests", "evaluation", "security", "migration"})

@dataclass(frozen=True)
class ReleaseManifest:
    version: str
    artifact_sha256: str
    passed_gates: frozenset[str]

def promote(manifest: ReleaseManifest) -> str:
    if not manifest.version or len(manifest.artifact_sha256) != 64:
        raise ValueError("release identity is invalid")
    missing = REQUIRED_GATES - manifest.passed_gates
    if missing:
        raise ValueError("release gates missing: " + ", ".join(sorted(missing)))
    return f"promoted:{manifest.version}:{manifest.artifact_sha256[:12]}"

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

REQUIRED_GATES 定义了发布晋级必须通过的所有门禁，使用 frozenset 确保不可变。

ReleaseManifest 使用 frozen=True 使其不可变，防止在晋级过程中被意外修改。

promote 函数先验证发布身份，再检查门禁完整性，确保只有满足所有条件的发布才能晋级。

带提示练习

在 promote() 中添加缺失门禁检查，使 test_missing_security_gate_blocks_promotion 通过。

打开 forge/reliability/deployment.py 文件。
定义 REQUIRED_GATES 常量，包含所有必需门禁。
在 promote 函数中，计算 missing = REQUIRED_GATES - manifest.passed_gates。
如果 missing 非空，抛出 ValueError，消息包含缺失门禁名称。
运行测试命令验证两个测试均通过。

逐级提示

使用集合差集操作计算缺失门禁。
错误消息应包含缺失门禁的名称，例如 security。
确保 promote 函数在门禁缺失时不会返回晋级字符串。

完成标准

test_complete_release_is_promoted 通过。
test_missing_security_gate_blocks_promotion 通过。

故障注入与排查

故障夹具 fixtures/failure.json 描述了一个发布晋级绕过漏洞：promote 函数只检查版本号存在就返回晋级字符串，导致缺少安全门禁的清单被错误晋级。运行故障测试 test_missing_security_gate_blocks_promotion 会失败，因为测试期望抛出包含 security 的 ValueError，但实际没有抛出异常。

本阶段的核心机制是门禁集合比较，因此能够诊断并修复此故障。通过检查 promote 函数是否比较了 passed_gates 与必需门禁集合，可以定位缺陷并添加缺失门禁检查。

已验证的故障合同

目标测试: test_missing_security_gate_blocks_promotion
故障文件: forge/reliability/deployment.py

诊断顺序

运行故障测试命令 python -m pytest -q ../tests/test_stage.py::test_missing_security_gate_blocks_promotion，观察失败输出。
查看 forge/reliability/deployment.py 中的 promote 函数，发现它只检查 manifest.version 非空就返回晋级字符串。
确认缺少对 passed_gates 与必需门禁集合的比较，导致缺少安全门禁的清单被晋级。
修复方法：添加 REQUIRED_GATES 常量，并在 promote 中计算缺失门禁，如果缺失则抛出 ValueError。

修复原则: 在 forge/reliability/deployment.py 中定义 REQUIRED_GATES = frozenset({"tests", "evaluation", "security", "migration"})，并在 promote 函数中添加缺失门禁检查：missing = REQUIRED_GATES - manifest.passed_gates，如果 missing 非空则抛出 ValueError("release gates missing: " + ", ".join(sorted(missing)))。

python -m pytest -q ../tests/test_stage.py::test_missing_security_gate_blocks_promotion

迁移挑战

音乐创作场景：你正在准备一个音乐发行版本，需要确保所有门禁通过后才能晋级。

改变的条件: 使用音乐场景数据创建 ReleaseManifest(version='2.0.0', artifact_sha256='b'*64, passed_gates={'tests','evaluation','security','migration'})，调用 promote() 并验证返回字符串以 promoted:2.0.0 开头。

在 forge/reliability/deployment.py 中，使用给定的音乐场景数据创建一个 ReleaseManifest 实例，调用 promote() 函数，并验证返回的字符串以 promoted:2.0.0 开头。

完成标准

创建的 ReleaseManifest 实例包含版本号 2.0.0、artifact SHA256 为 64 个 b 字符、已通过门禁集合包含 tests、evaluation、security、migration。
调用 promote() 返回的字符串以 promoted:2.0.0 开头。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么安全门禁是必需的？如果缺少多个门禁，错误消息应包含什么？

一个有力回答应包含

安全门禁确保发布版本经过了安全审查，防止未通过安全检查的代码进入生产环境。
如果缺少多个门禁，错误消息应包含所有缺失门禁的名称，以便开发者一次性了解所有问题并修复。

掌握证据

通过 test_complete_release_is_promoted 和 test_missing_security_gate_blocks_promotion 两个测试。
能够解释必需门禁集合 REQUIRED_GATES 的作用，并说明为什么安全门禁是必需的。
能够在 promote 函数中正确实现缺失门禁检查，抛出包含缺失门禁名称的 ValueError。

STAGE 06

比较基线与金丝雀错误率和延迟，在明确阈值下自动建议回滚

先看一个具体问题

何时应自动建议回滚金丝雀版本？

在部署新版本时，我们通常先让一小部分流量进入金丝雀实例，并观察其健康指标。

如果金丝雀的错误率或延迟显著高于基线，系统应自动建议回滚，而不是等待人工判断。

但小样本下的统计波动很大，仅凭少量请求就触发回滚会导致误报，浪费部署资源。

因此，我们需要实现一个函数，在满足最小样本量的前提下，比较基线与金丝雀的错误率和延迟，并给出回滚建议。

先做判断

在实现回滚决策函数时，以下哪种做法最合理？

只要金丝雀错误率高于基线就立即回滚，不考虑样本量。
先检查金丝雀请求数是否达到最小样本量，未达到则返回样本不足，达到后再比较错误率和延迟。
只比较延迟，忽略错误率，因为延迟更能反映用户体验。
无论指标如何，都让模型自行判断是否回滚。

判断依据: 正确做法是先检查最小样本量，避免小样本统计波动导致误回滚；然后再比较错误率和延迟是否超过阈值。

它是什么

回滚决策函数是什么

回滚决策函数是一个纯函数，输入基线健康指标、金丝雀健康指标和最小请求数，输出是否需要回滚以及原因元组。

它基于明确的阈值规则：错误率超过基线两倍且至少为2%，或延迟超过基线1.5倍时，建议回滚。

该函数不执行实际回滚，只提供决策依据，便于上层系统采取行动。

它不是什么

回滚决策函数不是什么

它不是实际执行回滚的操作，不会修改部署状态或发送通知。

它也不是一个统计检验工具，不会计算置信区间或进行假设检验，只是简单的阈值比较。

它不负责收集指标数据，输入的健康指标由外部系统提供。

它与相邻概念的关系

与其他组件的关系

该函数依赖于 Health 数据类，后者封装了请求数、错误数和 p95 延迟。

它被测试文件 tests/test_stage.py 中的两个测试用例直接调用，验证其行为。

在部署流程中，该函数位于金丝雀分析阶段，为后续的回滚操作提供决策。

本阶段的边界决定: 回滚决策只基于健康指标，不执行实际回滚或记录事故；当样本量不足时，返回 insufficient-sample 状态，避免误判。

实验 ID: evaluation-observability-deployment-06-canary-rollback-decision
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/evaluation-observability-deployment/06-canary-rollback-decision/
验收: 待验收

教师示范

目标: 演示 rollback_required() 如何识别错误率显著上升的金丝雀。

假设基线健康指标为 Health(requests=1000, errors=5, p95_ms=200)，即错误率0.5%，延迟200ms。

金丝雀健康指标为 Health(requests=200, errors=20, p95_ms=400)，错误率10%，延迟400ms。

金丝雀请求数200大于最小样本量100，因此进入指标比较。

错误率10%大于基线错误率的两倍（1%）且大于2%，所以添加 ‘error-rate’ 原因；延迟400ms大于基线延迟的1.5倍（300ms），所以添加 ‘latency’ 原因。

最终返回 (True, (‘error-rate’, ‘latency’))，建议回滚。

实施步骤

创建基线 Health 对象：Health(1000, 5, 200)。
创建金丝雀 Health 对象：Health(200, 20, 400)。
调用 rollback_required(baseline, canary)。
检查返回的 needed 为 True，且 reasons 包含 ‘error-rate’ 和 ‘latency’。

真实 solution 代码: forge/reliability/rollback.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Health:
    requests: int
    errors: int
    p95_ms: int

    @property
    def error_rate(self) -> float:
        return self.errors / self.requests if self.requests else 0.0

def rollback_required(baseline: Health, canary: Health, min_requests: int = 100) -> tuple[bool, tuple[str, ...]]:
    if canary.requests < min_requests:
        return False, ("insufficient-sample",)
    reasons=[]
    if canary.error_rate > max(0.02, baseline.error_rate * 2): reasons.append("error-rate")
    if canary.p95_ms > baseline.p95_ms * 1.5: reasons.append("latency")
    return bool(reasons), tuple(reasons)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

该示例展示了当金丝雀错误率和延迟都显著恶化时，函数会返回 True 并给出两个原因。

阈值规则：错误率超过基线两倍且至少2%，延迟超过基线1.5倍。

注意：示例中金丝雀请求数200满足最小样本量100，因此不会触发样本不足检查。

带提示练习

在 rollback_required() 中添加最小样本检查，使 test_small_sample_does_not_trigger_automatic_rollback 通过。

打开 forge/reliability/rollback.py 文件。
在 rollback_required 函数开头添加：if canary.requests < min_requests: return False, (“insufficient-sample”,)。
确保后续的指标比较逻辑不变。
运行测试：python -m pytest -q tests/test_stage.py::test_small_sample_does_not_trigger_automatic_rollback。

逐级提示

提示1：检查 canary.requests 是否小于 min_requests。
提示2：返回的 reasons 元组应包含 ‘insufficient-sample’。
提示3：不要忘记返回 needed 为 False。

完成标准

test_small_sample_does_not_trigger_automatic_rollback 测试通过。
test_bad_canary_requests_rollback 测试仍然通过。

故障注入与排查

在故障夹具中，rollback.py 缺少最小样本量检查，导致只有3个请求的金丝雀样本触发了自动回滚。测试 test_small_sample_does_not_trigger_automatic_rollback 失败，因为函数返回了 True 而不是 False。可执行故障夹具修改 forge/reliability/rollback.py，并由 test_small_sample_does_not_trigger_automatic_rollback 稳定复现。

本阶段的核心就是实现带最小样本量检查的回滚决策函数。通过对比故障代码和正确实现，学习者能够理解样本量检查的重要性，并修复该缺陷。

已验证的故障合同

目标测试: test_small_sample_does_not_trigger_automatic_rollback
故障文件: forge/reliability/rollback.py

诊断顺序

运行故障测试：python -m pytest -q ../tests/test_stage.py::test_small_sample_does_not_trigger_automatic_rollback，观察断言失败。
检查 rollback_required 函数，发现没有检查 canary.requests 是否小于 min_requests。
对比正确实现，在函数开头添加样本量检查并返回 insufficient-sample。

修复原则: 在 rollback_required 函数开头添加：if canary.requests < min_requests: return False, (“insufficient-sample”,)。这样当样本量不足时，函数返回 False 和原因 ‘insufficient-sample’，避免误回滚。

python -m pytest -q ../tests/test_stage.py::test_small_sample_does_not_trigger_automatic_rollback

迁移挑战

办公研究场景中，你需要评估一个金丝雀版本的性能。基线健康指标为 Health(requests=500, errors=10, p95_ms=200)，金丝雀健康指标为 Health(requests=400, errors=40, p95_ms=250)。

改变的条件: 使用不同的基线/金丝雀数据集（办公研究场景），但保持 rollback_required() 的阈值逻辑不变。

使用给定的办公场景数据创建基线和金丝雀 Health 对象，调用 rollback_required()，并验证返回值为 (True, (‘error-rate’,))。

完成标准

调用 rollback_required(Health(500,10,200), Health(400,40,250)) 返回 needed=True。
返回的 reasons 元组为 (‘error-rate’,)，不包含 ‘latency’，因为延迟250ms未超过基线延迟的1.5倍（300ms）。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么小样本不应触发自动回滚？请解释样本量检查在回滚决策中的作用及其对系统稳定性的影响。

一个有力回答应包含

小样本下错误率和延迟的估计方差很大，容易因随机波动而误判。
设置最小样本量可以确保指标具有统计意义，减少误回滚，提高部署稳定性。

掌握证据

通过 test_bad_canary_requests_rollback 和 test_small_sample_does_not_trigger_automatic_rollback 两个测试。
能够解释为什么小样本不应触发自动回滚，以及样本量检查如何防止误报。
在办公场景的转移挑战中，正确使用 rollback_required() 并验证返回 (True, (‘error-rate’,))。

STAGE 07

记录检测、缓解、恢复和复盘时间点，并验证恢复目标与证据完整性

先看一个具体问题

为什么有序时间线是恢复指标可信的前提？

在事故演练中，检测、缓解、恢复和复盘四个时间点必须按先后顺序记录，否则计算出的恢复时间毫无意义。

当前 forge/reliability/incident_drill.py 文件尚不存在，测试导入 DrillRecord 和 verify_drill 时直接失败，因此需要从零创建该模块。

如果只计算 recovered_at - detected_at 而不检查时间戳顺序，像 detected_at=100, mitigated_at=80 这样的不可能序列也会被接受，导致恢复目标误判。

本阶段要求实现 verify_drill()，它必须验证时间线单调递增且证据非空，然后才计算恢复时间并判断是否达到目标。

先做判断

在实现 verify_drill() 时，如果时间戳顺序错误，函数应该怎么做？

仍然计算恢复时间，但返回 objectiveMet=False
抛出 ValueError，拒绝处理该记录
自动调整时间戳顺序后再计算
忽略顺序问题，只检查证据是否为空

判断依据: 正确做法是抛出 ValueError，因为时间线顺序是恢复指标有效性的前提，不能静默修正或忽略。

它是什么

事故演练验证是什么

事故演练验证是一个纯函数，接收 DrillRecord 和恢复目标秒数，返回包含恢复秒数和目标达成情况的字典。

它首先检查四个时间戳是否单调递增，然后检查证据元组非空，最后才计算恢复时间并比较目标。

该验证只依赖输入数据，不涉及实际部署或回滚操作，因此可以安全地用于演练复盘。

它不是什么

事故演练验证不是什么

它不是实际的事故响应系统，不会触发告警或执行回滚，只对演练记录进行静态检查。

它不负责生成时间戳或证据，这些数据必须由外部系统提供，验证只判断其合理性。

它不保证恢复目标一定达成，而是根据输入数据客观计算并返回布尔值。

它与相邻概念的关系

与其他组件的关系

DrillRecord 是数据载体，使用 dataclass(frozen=True) 确保不可变，避免验证过程中被意外修改。

verify_drill() 依赖 DrillRecord 的字段，但独立于任何存储或网络层，便于单元测试。

本模块与 rollback.py 等可靠性组件并列，共同构成事故处理工具集，但本阶段只关注验证逻辑。

本阶段的边界决定: 验证函数必须在计算任何指标之前拒绝无效输入，否则后续指标将建立在不可信的数据上。

实验 ID: evaluation-observability-deployment-07-incident-drill-evidence
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/evaluation-observability-deployment/07-incident-drill-evidence/
验收: 待验收

教师示范

目标: 演示 verify_drill() 如何从有序时间线计算恢复时间并判断目标。

假设一次演练记录为 DrillRecord(0, 30, 90, 120, ("trace.json",))，恢复目标为 120 秒。

时间戳 0→30→90→120 单调递增，证据元组包含 "trace.json"，因此验证通过。

恢复时间计算为 90 - 0 = 90 秒，小于等于目标 120 秒，所以 objectiveMet 为 True。

实施步骤

创建 forge/reliability/incident_drill.py 文件。
定义 DrillRecord 数据类，包含五个字段。
实现 verify_drill()，先检查时间顺序和证据，再计算恢复时间。
运行测试 test_ordered_drill_measures_recovery 验证输出。

真实 solution 代码: forge/reliability/incident_drill.py

from dataclasses import dataclass

@dataclass(frozen=True)
class DrillRecord:
    detected_at: int
    mitigated_at: int
    recovered_at: int
    reviewed_at: int
    evidence: tuple[str, ...]

def verify_drill(record: DrillRecord, recovery_objective_seconds: int) -> dict[str, int | bool]:
    times = (record.detected_at, record.mitigated_at, record.recovered_at, record.reviewed_at)
    if list(times) != sorted(times):
        raise ValueError("incident timestamps are out of order")
    if not record.evidence:
        raise ValueError("drill evidence is required")
    recovery_time = record.recovered_at - record.detected_at
    return {"recoverySeconds": recovery_time, "objectiveMet": recovery_time <= recovery_objective_seconds}

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

时间戳检查使用 list(times) != sorted(times) 判断是否有序，若不等则抛出 ValueError。

证据检查确保 record.evidence 非空，避免无证据的演练记录被接受。

恢复时间只使用 detected_at 和 recovered_at，与缓解和复盘时间无关，但后两者必须存在且有序。

带提示练习

在 verify_drill() 中添加时间顺序检查，使 test_out_of_order_timeline_is_rejected 通过。

打开 forge/reliability/incident_drill.py，查看当前实现。
在计算恢复时间之前，添加时间戳顺序检查。
如果时间戳无序，抛出 ValueError，错误信息包含 "order"。
运行测试 test_out_of_order_timeline_is_rejected 确认通过。

逐级提示

可以使用 sorted() 比较时间戳元组是否有序。
错误信息必须包含 "order"，否则测试断言失败。
检查应放在证据检查之前还是之后？考虑哪个错误更基础。

完成标准

test_out_of_order_timeline_is_rejected 测试通过。
test_ordered_drill_measures_recovery 仍然通过。

故障注入与排查

当前 forge/reliability/incident_drill.py 中的 verify_drill() 只计算恢复时间，没有验证时间戳顺序。运行 test_out_of_order_timeline_is_rejected 时，传入 DrillRecord(100, 80, 140, 160, ("trace.json",))，其中缓解时间 80 早于检测时间 100，但函数仍然返回 objectiveMet=True，测试因此失败。

本阶段已经学习了时间线验证的必要性，并且 verify_drill() 的最终实现包含顺序检查。通过定位这个故障，可以理解为什么必须拒绝无序时间线，以及如何修复。

已验证的故障合同

目标测试: test_out_of_order_timeline_is_rejected
故障文件: forge/reliability/incident_drill.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_out_of_order_timeline_is_rejected，观察失败信息。
查看 verify_drill() 源码，发现缺少时间戳顺序检查。
在计算恢复时间之前添加顺序检查，并抛出 ValueError。
重新运行测试，确认通过。

修复原则: 在 verify_drill() 中添加时间戳顺序检查：times = (record.detected_at, record.mitigated_at, record.recovered_at, record.reviewed_at)，然后 if list(times) != sorted(times): raise ValueError("incident timestamps are out of order")。

python -m pytest -q ../tests/test_stage.py::test_out_of_order_timeline_is_rejected

迁移挑战

办公研究场景中，一次文档处理事故的演练记录为：检测时间 10 秒，缓解时间 40 秒，恢复时间 100 秒，复盘时间 150 秒，证据文件为 office-trace.json。

改变的条件: 将演练场景从软件交付改为办公研究，并使用不同的时间戳和证据文件名。

创建 DrillRecord(detected_at=10, mitigated_at=40, recovered_at=100, reviewed_at=150, evidence=('office-trace.json',))，调用 verify_drill(record, 120)，并验证返回 {'recoverySeconds':90, 'objectiveMet':True}。

完成标准

调用 verify_drill 后返回的字典中 recoverySeconds 等于 90。
返回的字典中 objectiveMet 为 True，因为 90 秒小于目标 120 秒。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么时间戳必须单调递增？如果允许无序时间戳，恢复时间计算会有什么后果？

一个有力回答应包含

时间戳单调递增是事故时间线的基本约束，否则无法确定事件先后顺序。
如果允许无序，恢复时间可能为负数或错误值，导致恢复目标判断失真。
验证函数必须拒绝无效输入，以保证后续指标的可信度。

掌握证据

通过 test_ordered_drill_measures_recovery 和 test_out_of_order_timeline_is_rejected 两个测试。
能够解释为什么时间戳顺序检查是恢复指标计算的前提。
在办公场景中正确创建 DrillRecord 并验证恢复目标。

完成本章

形成可观察的线上版本，覆盖质量、延迟、成本、失败和回滚。

FORGE / LOCAL CHECK

本地实验自检

未开始
2阅读中
3实验已下载
4测试结果已读取
5本地自检通过

verification.json 只在当前浏览器中解析，不会上传。这里验证的是实验合同，不是服务器认证或第三方背书。

选择 verification.json

主要层

依赖能力

为后续提供

受这些层约束

概念校准

验证与评测

可观测性

Harness

评测、可观测性、部署与可靠性

先看一个具体问题

评测案例需要稳定标识，否则重复案例会破坏追踪

先做判断

它是什么

评测数据集是一个带版本、由唯一案例组成的集合

它不是什么

评测数据集不执行评测，也不计算指标

它与相邻概念的关系

EvalCase 与 EvalDataset 的关系

先看一个具体问题

如何从多个案例结果中汇总出有意义的整体质量指标？

先做判断

它是什么

质量指标汇总是什么

它不是什么

质量指标汇总不是什么

它与相邻概念的关系

与其他概念的关系

先看一个具体问题

一次运行中的 Span 必须共享同一个 Trace ID

先做判断

它是什么

Trace 是一棵由 Span 组成的树

它不是什么

Trace 验证不涉及指标或预算

它与相邻概念的关系

Span 与 Trace 的关系

先看一个具体问题

如何判断一次运行是否在成本与延迟预算内？

先做判断

它是什么

预算检查是什么

它不是什么

预算检查不是什么

它与相邻概念的关系

与其他组件的关系

先看一个具体问题

发布候选版本缺少安全门禁却被晋级

先做判断

它是什么

发布晋级门禁是什么

它不是什么

发布晋级门禁不是什么

它与相邻概念的关系

与其他概念的关系

先看一个具体问题

何时应自动建议回滚金丝雀版本？

先做判断

它是什么

回滚决策函数是什么

它不是什么

回滚决策函数不是什么

它与相邻概念的关系

与其他组件的关系

先看一个具体问题

为什么有序时间线是恢复指标可信的前提？

先做判断

它是什么

事故演练验证是什么

它不是什么

事故演练验证不是什么

它与相邻概念的关系

与其他组件的关系