11｜机制实验与完整源码
用固定输入做一次可定位的改动，看哪些输出该变、哪些不该变。本页把“跑得通”推进到独立计算路径和故障注入。
图 1｜只修改未来两项，前三项输出保持不变（原创教学示意）
阅读路线：先确认输入、断言与结果记录的关系，再依次检查监督位置、attention 与缓存，最后检查偏好目标、组内优势和概率守恒。只需使用源码时可跳到第五章，阅读机制时按 1→4 顺序进行。
本页目录：1. 入口、环境与证据边界 · 2. 损失与监督边界 · 3. Attention 与缓存状态 · 4. 训练目标与概率守恒 · 5. 结果与完整源码
前面各篇把模型训练拆成了数据、概率、梯度与推理状态。这一篇让它们落到同一套可以独立运行的实验中：不下载权重，不调用模型服务，只用 NumPy 检验八个容易说错的机制。完整源码在文末，保存为 mechanism_labs.py 即可运行。实验实际执行于 2026 年 10 月 8 日，运行环境为 Python 3.12.14、NumPy 2.3.5、Windows 11，数组采用 float64。
实验对象是机制，不是模型能力。 随机初始化的微型 decoder 可以证明同一计算的两种写法等价，不能证明它会写代码、推理或回答问题。本篇没有完成预训练、RLHF 或 GRPO 训练，也没有测量商业模型的吞吐。前文中的教学代码片段应分别理解；本篇实际运行的是这里给出的 NumPy 源码。
1. 入口、环境与证据边界
1.1 从输入、断言到可复查结果
代码有三层：稳定的 softmax、交叉熵、attention 等基础函数；调用这些函数的八个实验；汇总环境与结果的 JSON 报告。每个实验先构造可解释的输入，再比较数学上应相等的结果，最后返回实际误差。断言失败会中断程序，只有八组全部通过才会输出 status: passed。
图 2｜检查失败会停止程序；通过记录同时保留环境、输入和误差。
在已有 Python 的环境中安装依赖并运行：
python -m pip install numpy==2.3.5
python mechanism_labs.py --output mechanism-results.json --samples 200000
此次执行使用已有的 NumPy，没有临时安装包。--output 指定报告位置，--samples 只控制 speculative sampling 的蒙特卡洛次数。默认结果写入脚本上一级目录的 qa/mechanism-results.json，复制单文件运行时建议显式传入输出位置。代码不访问网络，也没有需要准备的训练集。
1.2 原留存运行记录与数值容差
以下环境、终端输出、误差和抽样频率保留自 2026-10-08 的原实验记录。2026-10-19 又在 Python 3.12.14、NumPy 2.3.5、Windows 11、float64 下运行了同一份源码，八组实验全部通过。此次两层 KV cache 的最大 logit 误差为 ，原记录为 ，均远低于 的断言容差；其余原文显示精度下的数值与抽样频率一致。下面保留原运行日志，便于区分两次记录。
实际终端输出中的关键内容为：
PASS: 8 mechanism labs
CE finite-difference error: 6.189e-12
Cache logit max error: 6.661e-16
Speculative proportions: [0.599475, 0.300205, 0.10032]
每个数值断言都有容差。浮点数经过不同矩阵运算路径，最后几位可能不同，因此“相同”通常表示数值等价，不要求二进制逐位相同。NumPy 的 assert_allclose 用绝对与相对容差共同判断，具体语义见其官方文档。固定种子有助于重现输入，但不同 NumPy 版本、平台或底层线性代数库仍可能改变最后几位。
2. 损失与监督边界
2.1 交叉熵梯度与有限差分
图 3｜解析梯度与损失扰动的独立检查路径
图中右侧曲线改变目标 logit，绿色点对应原输入；局部斜率为 −0.75。数值检查分别构造正负扰动输入，再由两次损失差估计导数。
第一个实验的词表只有三个 token，logits 是 [ln 2, 0, 0]，目标 token 的下标是 2。softmax 后概率为 [0.5, 0.25, 0.25]，所以交叉熵为 −ln 0.25 = ln 4，对三个 logits 的梯度分别为 [0.5, 0.25, −0.75]。这也是第 03 篇的手算例。
检查梯度时，程序没有再写一遍解析公式，而是对每个 logit 做微小扰动，通过损失差分估计导数：
本篇数值检查的符号、形状与单位
| 符号 | 含义 | 形状或单位 |
|---|---|---|
| 预测 logits | ，无量纲 | |
| 有效目标位置的平均负对数似然 | 标量，nat/token | |
| 仅第 个元素为 1 的扰动方向 | 与 同形状 | |
| 中心差分步长，本实验为 | 与 logit 同单位 | |
| 数值梯度 | 与 同形状 | |
| attention 头数、序列长度、每头宽度 | 正整数 | |
| 目标分布、草稿或参考分布 | 长度 ，概率和为 1 | |
| 第 个回答的组内优势 | 标量，无量纲 | |
| 草稿 token 的接受概率 | 标量，范围 |
中心差分在函数足够光滑时的截断误差随 缩小，但过小的步长又会使两个接近的浮点数相减，放大舍入误差。因此不应把 无止境缩小。这次采用 float64 与 ，解析梯度和数值梯度的最大绝对误差为 ，低于断言使用的绝对容差 。
这项检查能发现常见错误：忘记减去目标 one-hot、平均分母错误、softmax 轴错误。它检查的是对 logits 的导数；并未检查一个完整网络每层的参数梯度。脚本没有自动微分引擎，也没有实现 decoder 的反向传播。
2.2 三种 mask 管三个不同的边界
第二组实验同时检查 loss mask、文档 attention mask 与 SFT 的目标位置偏移。它们都常被写成布尔数组，但决定的事情不同。
2.2.1 Loss mask：监督哪些预测位置
图 4｜被忽略预测行的直接 logit 梯度为零
图中第二行变灰，表示这一行的预测不进入损失；分母只统计另外两行。
先看 loss mask。三行 logits 中，第二行不参与监督，损失只平均第一行与第三行。程序把第二行的 logits 改成 [100, −100, 0, 0]，损失仍然相同；这一行的直接 logit 梯度严格为零。有效位置的梯度则被有效目标数 2 归一化，而不是被全部位置数 3 归一化。
用户 token 不计入损失，不等于用户内容没有影响梯度。 在真实 decoder 中，assistant 的预测依赖用户上下文，损失仍会沿 attention 与残差路径传回用户位置的隐藏状态。这里证明的仅是被忽略预测位置的直接 logit 梯度为零。
2.2.2 Attention mask：独立文档之间不读取
图 5｜两个独立文档的因果块对角可见关系
图中两个蓝色三角属于两个文档；位置 4 只保留自身蓝格，不能穿过文档边界读位置 0—3。
再把两个四 token 文档放在同一行，文档编号为 [0,0,0,0,1,1,1,1]。如果它们是独立样本，那么 attention 必须同时满足“key 位置不晚于 query”和“两个位置来自同一文档”。第二个文档的第一个 token 只能看见自己，不能读取前面四个 token。边界处第一个文档末尾预测第二个文档开头的目标也被忽略，七个预测位置的有效 mask 是：
[1, 1, 1, 0, 1, 1, 1]
这是本实验选择的独立文档 packing 语义。有些实际预训练流程允许拼接文档互相可见，两种方式不能仅凭存在 EOS 来互相推断。EOS 是一个 token；硬隔离由 attention mask 决定。
2.2.3 SFT 目标 mask：监督位置前移一格
图 6｜assistant 目标与预测它的 logits 相差一格
图中的箭头从下排预测位置指向上排目标 token；绿色目标 4、5、6，对应蓝色预测位置 3、4、5。
SFT 的检查则针对最容易出现的偏移：原序列位置 4、5、6 属于 assistant 目标，对应预测它们的 logits 位于 3、4、5。程序对原目标 mask 取 [1:]，结果确实为 3、4、5。如果把原 mask 直接贴到已经移位的 logits 上，就会监督错位置。
3. Attention 与缓存状态
3.1 缩放点积的数值路径
图 7｜两组 key/value 的权重与加权输出
图中 score 经 softmax 变成两个权重；每个权重只与对应 value 相乘，然后相加。
第三组实验先复现 attention 的手算：query 是 [1,0]，两个 key 是 [1,0] 与 [0,1]，value 是 [2,0] 与 [0,4]。缩放点积除以 后，权重约为 [0.669762,0.330238]，输出约为 [1.339523,1.320954]。
3.2 因果方向与未来扰动
首屏示意图固定前三项，只修改未来两项；虚线是观察边界，图只画出部分允许的因果连接，实际矩阵方向由本实验检查。
随后程序构造五位置 attention，把未来位置 3、4 的输入加上 100，前面三个位置的输出变化为零；上三角未来位置权重也严格为零。这个测试验证 causal mask 的方向：如果误把下三角写成上三角，扰动未来输入就会泄漏到前面的输出。
3.3 两层 decoder 的三条等价前向路径
图 8｜整段、增长前缀与缓存路径比较同一组词表 logits
图中三行都对应七个固定位置，比较的是每个位置的 17 维输出，不是三次独立随机生成的答案。
第四组把检查推进到完整的两层微型 decoder。它包含独立的层参数、RMSNorm、两头 MHA、逐对 RoPE、SwiGLU、残差连接与词表输出投影。参数随机初始化，词表 17、宽度 8、每头宽度 4、FFN 中间宽度 12。它是教学结构，不代表任何未公开商业型号的实际架构。
同一串七个 token 分别走三条计算路径：一次输入全部 token；对每个增长中的前缀重新计算；每次只输入一个新 token 并读取每层 KV cache。比较的是所有位置的词表 logits：
图 9｜两层各维护自己的 cache；三条路径使用相同位置与参数（原有教学图）。
实测整段与 cache 的最大误差为 ；逐前缀重新计算与 cache 的误差同样如此。两层最终的 K 与 V 各自形状都是 [2,7,4]，含义是 [头数,已有 token 数,每头宽度]。程序另外确认两层的 K 不相同，避免“缓存一个数组给所有层共用”这样的伪实现通过简单检查。
缓存路径中新 token 的 RoPE 位置必须使用全局位置，而不是每次都从零开始。当前 query 是已有序列的最后一个位置，因而可以读全部缓存 key；它的 attention mask 是 [1,T] 的全真矩阵。若机械地对这个长方形矩阵使用从左上角开始的下三角，最后一个 query 反而只能读第一个 key。
cache 的核心性质是复用过去已经算出的 K/V，并保持下一步 logits 数值等价。 这里没有 dropout，没有量化，没有训练时更新权重，也没有修改历史上下文。测试没有实现 paged attention、GQA 或跨请求 cache 管理，更没有展示速度提升；这些工程性质需要另外的实验。
3.4 故障注入：形状全对，为什么logits仍然错误
图 10｜三种语义错误可以保留同样的七位置缓存（原创教学示意）
先比较图中每一排的长度：它们都保留七个位置。第二排重置的是位置编号，历史仍允许被读取；第三排把第一层的K、V分别错复制给第二层；最后一排才是把长方形mask误写成左上角下三角，只让当前query读第一项。它们破坏了三个不同的约定，不能只靠 shape == [2,7,4] 判断正确。
为检验这些区别，补充实验直接导入本页原样的 mechanism_labs.py，使用相同七个token、同一随机参数和float64。基准输出先由正确的 model.full(ids) 保存；再在逐token路径中一次只注入一种错误。原始完整脚本不改，实验之后也恢复被临时替换的函数。
| 逐token实现 | 两层缓存形状 | 最大logit绝对误差 |
|---|---|---|
| 正确实现 | 均为 [2,7,4] | |
| 每步RoPE位置归零 | 均为 [2,7,4] | 0.247728 |
| 把Cache1复制给Cache2 | 均为 [2,7,4] | 0.944921 |
对 [1,T] 用左上角下三角 | 均为 [2,7,4] | 1.551164 |
这不是把“错误程序崩溃”当作检查。三条错误路径都能完成前向，且都通过形状检查；它们却全部不能通过 atol=rtol=1e-12 的同输入logits比较。逐项观察错误，就能知道该检查哪条不变量：
- 位置归零改变旋转关系。 cache长度仍随
position增长，拼接也正常；但新Q/K的RoPE角度没有使用真实全局位置。没有上三角泄露，也不代表位置语义正确。 - 串层复制改变值的来源。 第二层使用了第一层隐藏状态投影出的历史。它们形状相同，不代表含义相同。原脚本的“两层K不能全部相同”检查能捕获这次特定故障；一般情况下，跨路径logits比较仍是必要的独立证据。
- 长矩形mask改变可见集合。
np.tril(np.ones((1,T)))只有第一个位置为真。当前query明明位于真实序列末尾，却被当成矩阵第一行的时间起点。源代码的全真[1,T]在“只有一个最新query、无padding”的条件下才是正确选择。
补充实验于2026-10-19在Python 3.12.14、NumPy 2.3.5、Windows11、float64下实际执行；这些误差只属于这套固定微型模型。它们没有测量真实模型质量，也不否定第10篇那种明确定义了共享结构的CSA2：故障是违反当前独立层配置，而不是“任何共享都错误”。
保留三个检查层次很有用：shape检查对象能否接上；位置、mask、所有权检查它表示什么；同输入下的独立计算路径检查组合后是否仍等价。单个通过信号不能替代另外两个。
4. 训练目标与概率守恒
4.1 DPO 的局部梯度与相对关系
图 11｜两个回答都降概率，差值与 DPO 目标仍可能改善
图中绿点和红点都向更小的 log probability 移动，红点移动更远，所以两点间距增大。这个反例针对目标函数的局部输入，不表示真实网络参数会按两条独立旋钮更新。
第五组实验把 DPO 的四个序列 log probability 当作输入：当前策略胜答与败答为 −2、−3，参考策略二者均为 −2.5，。因此 preference margin 为 ，损失为 。解析梯度对胜答 log probability 为 −0.0900332，对败答为 +0.0900332，与中心差分的最大误差为 。
沿这两个输入的负梯度走一步，损失下降到 0.596519。再构造一个容易忽略的反例：把胜答 log probability 减少 1，败答减少 2，两者概率都下降，但相对 margin 增大，DPO 损失仍下降到 0.513015。它说明偏好对目标主要约束相对关系。
这里把两个 log probability 看成局部独立变量，方便检查第 04 篇推导的表达式。真实训练中的 log probability 来自共享网络、归一化词表概率与整段序列求和，不能把这个两变量下降步骤视为网络参数更新。DPO 的推导与训练目标见原论文。
4.2 GRPO 的组内优势
图 12｜同一奖励组在两种标准差约定下的优势
先观察奖励到组均值的差，再看分母的标准差；更改 ddof 就改变优势幅度，不能在复现时省略这个约定。
第六组检查 GRPO 的组内归一化。对同一问题采样四个回答，奖励为 [0,0,1,1]，均值为 0.5，总体标准差为 0.5，于是：
本实验采用总体标准差 ddof=0、，优势约为 [−1,−1,1,1]。若改成样本标准差 ddof=1，同一组优势约为 [−0.866025,−0.866025,0.866025,0.866025]。两种标准差都存在于软件接口与算法实现中，复现时必须写明选择，NumPy 的定义见std 文档。
所有回答都得 1 或都得 0 时，组内差值全为零，归一化优势也全为零。这说明单靠这一项信号，整组全对与整组全错都没有相对排序梯度；有 KL、熵或其他辅助项的训练仍可能产生更新。本实验还验证了打乱回答顺序只会同样打乱优势，不改变其数值。
4.2.1 Clipped 项的正负优势
图 13｜四个固定样本逐项比较未裁剪与裁剪目标
图中最后一行分别从上面两行取较小值；第二列和第三列显示为什么“只裁剪 ratio”不是同一个操作。
PPO 风格 clipped 项也必须考虑优势正负。取 ratio [1.5,1.5,0.5,0.5]、优势 [2,−2,2,−2]、clip 范围 [0.8,1.2]，逐元素计算 min(ratio*A, clip(ratio)*A)，得到 [2.4,−3,1,−1.6]。不能只把 ratio 裁剪后乘优势，这会改变保守目标在负优势时的行为。GRPO 的源头可查DeepSeekMath，clipped surrogate 可查PPO 原论文。代码只检查冻结的数值优势与目标，没有 rollout 或 actor 训练循环。
4.3 完整离散分布上的 KL
图 14｜相同两组分布，KL 的两个方向不同
图中蓝条与橙条具有不同的概率质量；交换方向时，参与平均的权重也交换，结果不会由距离对称性保证相同。
第七组对完整的小词表分布直接求 KL。当 、 时， nat，反向为 nat，二者不同。相同分布的 KL 为零；若 某项为正而 同项为零，前向 KL 为无穷大。真实大词表训练常使用采样估计，本实验没有把单次 log ratio 误称为总是非负的 KL。
4.4 Speculative sampling 的概率守恒
4.4.1 接受与残差的精确质量分解
图 15｜接受质量与拒绝补偿恢复目标分布
绿色部分是直接接受的质量，红色部分是拒绝概率；补偿只流向 target 相对 draft 缺失的第一个 token。
最后一组回答 speculative sampling 最关键的问题：从草稿分布 提议 token，如何保持最终 token 服从目标分布 ？对提议的 token ，接受概率为：
当 时，草稿根本不会提议这一项，代码无需执行这个除法。直接接受路径分配给 token 的概率质量为 。剩下的质量必须补给目标分布相对草稿不足的部分：
当 时，拒绝后从 采样；当两个分布完全相同时，拒绝概率为零，不需要定义可采样的残差分布。最终分布为：
代码使用 、。接受概率为 [1,0.75,0.5]，直接接受质量为 [0.4,0.3,0.1]，拒绝概率为 0.2，残差分布为 [1,0,0]。拒绝后全部补到第一项，最终正好回到 [0.6,0.3,0.1]，精确枚举的浮点误差为 。
4.4.2 有限样本频率与标准误差
图 16｜原留存频率偏差在标准误差尺度上的位置
图中圆点接近零线，横轴是各 token 的标准误差单位。图上界限检查的是有限样本频率，不能代替前面的精确质量证明。
再用种子 73 实际采样 200000 次，观察到 [0.599475,0.300205,0.100320]，实际接受率为 0.801205。每个 token 边际频率的标准误差按 估计，最大偏差只有 0.4793 个标准误差，低于代码预先声明的六倍标准误差边界。频率检查是有限样本验证；精确质量分解才解释为什么分布守恒。
拒绝后直接从原目标分布采样，一般会改变输出分布。 在这个例子中，它会得到 [0.52,0.36,0.12]，因为直接接受路径已经覆盖了目标的部分质量，剩余部分需要按正残差补偿。这一机制可查Fast Inference from Transformers via Speculative Decoding与Accelerating Large Language Model Decoding with Speculative Sampling。
本实验只验证固定前缀下的一步接受律。完整多 token 算法还需要按前缀逐个接受、首次拒绝后停止草稿分支，以及处理全部接受后的额外 token。若使用 temperature、top-k、top-p，公式里的 与 必须对应实际使用的归一化采样分布；不能一个用截断后分布、另一个用原始概率。这里未实现整套多 token decoder，也未测量其加速收益。
4.4.3 错误补偿：为什么重新从target采样仍会偏离target
图 17｜上排只填缺口，下排把概率又分给已经覆盖的项（原创教学示意）
看上排：青色已接受质量先占据三个容器的一部分，只有第一项尚未到达目标线，琥珀色正残差只补第一项。下排则把拒绝后的质量重新按完整target分给三个容器，第二、三项重复得到本来已经覆盖的质量。容器刻度只作缺口关系示意，精确概率用下面的数值计算，不从插画量高度。
仍取 、。直接接受质量已经是 ，总拒绝率为0.2。正确残差 ，因此补偿项为 。如果错误地让拒绝分支重新从 采样，新增项会变成 ：
第一项少了0.08，第二项多0.06，第三项多0.02。分布仍然非负、总和仍然为1，所以“概率和等于1”也检查不出这个错误。逐项比较得到 距离 ，总变差距离为0.08；这是系统性偏差，不是200000次采样的随机抖动。
V2补充实验实际枚举了这条错误分支，并断言正确补偿恢复 、错误补偿得到上述分布。它还检查 和相同分布的原有分支，没有改写或删减原脚本的断言。由此可把调试顺序分清：先检验精确概率质量，再做有限样本频率检查；若精确质量已经错了，增加抽样次数只会更精确地看见错误分布。
5. 结果与完整源码
5.1 原留存结果与各自验证范围
2026-10-08原留存八项结果摘要
| 检查 | 实际结果 | 验证范围 |
|---|---|---|
| CE 梯度有限差分 | 最大误差 | 对 logits 的导数 |
| Mask 后梯度 | 被忽略行全为零；有效行误差 | loss mask、packing 边界、SFT 偏移 |
| 因果 attention | 未来扰动引起的前缀变化为 0 | 因果方向与归一化 |
| 两层 KV cache | 最大 logit 误差 | 固定权重的前向等价 |
| DPO 局部梯度 | 最大误差 | 四个 log probability 的目标表达式 |
| GRPO 组内优势 | 混合组约为 ，同奖组为 0 | 标准差约定、排列等价、clip 符号 |
| KL | 前向 0.192745、反向 0.223144 nat | 完整离散分布与零支持 |
| Speculative sampling | 精确误差 ，200000 次抽样通过 | 单前缀接受律与残差分布 |
5.2 调用路线、接口与完整文件
完整代码使用固定种子，不需要前文 PyTorch 片段。展开后复制保存即可；后续修改时，应保留独立比较路径和断言，避免把“实际输出”写死成“正确答案”。若想继续扩展，优先验证一个明确问题，例如 GQA 的 cache 头数、跨文档 mask 或量化后的容差，再根据结果判断是否需要扩大实验。
完整 NumPy 实验源码：mechanism_labs.py
"""Deterministic NumPy mechanism labs, not a production LLM trainer.
Run: python mechanism_labs.py --output ../qa/mechanism-results.json
All numerical equivalence checks use fixed inputs, float64 and explicit tolerances.
"""
from __future__ import annotations
import argparse
from datetime import datetime, timezone, timedelta
import json
from pathlib import Path
import platform
import sys
import time
import numpy as np
def softmax(x: np.ndarray, axis: int = -1) -> np.ndarray:
shifted = x - np.max(x, axis=axis, keepdims=True)
exp = np.exp(shifted)
return exp / exp.sum(axis=axis, keepdims=True)
def log_softmax(x: np.ndarray) -> np.ndarray:
shifted = x - np.max(x, axis=-1, keepdims=True)
return shifted - np.log(np.exp(shifted).sum(axis=-1, keepdims=True))
def finite_diff(fn, x: np.ndarray, eps: float = 1e-5) -> np.ndarray:
grad = np.empty_like(x, dtype=np.float64)
for index in np.ndindex(x.shape):
plus, minus = x.copy(), x.copy()
plus[index] += eps
minus[index] -= eps
grad[index] = (fn(plus) - fn(minus)) / (2 * eps)
return grad
def masked_ce(logits, targets, valid):
"""[N,V] logits, [N] targets and Boolean valid mask; returns mean and gradient."""
count = int(valid.sum())
if count == 0:
raise ValueError("No supervised targets")
lp = log_softmax(logits)
loss = -float(lp[np.arange(len(targets)), targets][valid].mean())
grad = np.exp(lp)
grad[np.arange(len(targets)), targets] -= 1
grad *= valid[:, None] / count
return loss, grad
def lab_ce_gradient():
logits = np.array([[np.log(2.0), 0.0, 0.0]], dtype=np.float64)
loss, gradient = masked_ce(logits, np.array([2]), np.array([True]))
numeric = finite_diff(lambda z: masked_ce(z, np.array([2]),
np.array([True]))[0], logits)
error = float(np.max(np.abs(gradient - numeric)))
np.testing.assert_allclose(gradient, [[0.5, 0.25, -0.75]], atol=1e-14)
np.testing.assert_allclose(gradient, numeric, atol=1e-9, rtol=1e-8)
assert np.isclose(loss, np.log(4.0))
return {"loss_nat": loss, "gradient": gradient.tolist(),
"finite_difference_step": 1e-5, "max_abs_gradient_error": error,
"abs_tolerance": 1e-9, "relative_tolerance": 1e-8}
def lab_masks():
rng = np.random.default_rng(3)
logits = rng.normal(size=(3, 4))
targets = np.array([2, 1, 3])
valid = np.array([True, False, True])
loss, grad = masked_ce(logits, targets, valid)
numeric = finite_diff(lambda z: masked_ce(z, targets, valid)[0], logits)
np.testing.assert_allclose(grad, numeric, atol=1e-9, rtol=1e-8)
changed = logits.copy()
changed[1] = [100, -100, 0, 0]
assert masked_ce(changed, targets, valid)[0] == loss
np.testing.assert_array_equal(grad[1], np.zeros(4))
# Two independent four-token documents: A EOS -> B BOS is not supervised.
document_ids = np.array([0, 0, 0, 0, 1, 1, 1, 1])
prediction_valid = document_ids[:-1] == document_ids[1:]
positions = np.arange(8)
allowed = ((positions[None, :] <= positions[:, None]) &
(document_ids[:, None] == document_ids[None, :]))
assert not prediction_valid[3]
assert allowed[4].sum() == 1
assert allowed[7].sum() == 4
assert not allowed[4:, :4].any()
# Original token positions 4,5,6 are assistant targets; prediction positions 3,4,5.
assistant_target = np.array([False, False, False, False, True, True, True])
sft_prediction_mask = assistant_target[1:]
np.testing.assert_array_equal(np.flatnonzero(sft_prediction_mask), [3, 4, 5])
return {"masked_loss_nat": loss, "masked_row_gradient": grad[1].tolist(),
"max_abs_gradient_error": float(np.max(np.abs(grad - numeric))),
"packing_prediction_mask": prediction_valid.astype(int).tolist(),
"packing_attention_allowed": allowed.astype(int).tolist(),
"sft_prediction_positions": np.flatnonzero(sft_prediction_mask).tolist(),
"masked_logits_perturbation_delta": 0.0}
def attention(q, k, value, allowed):
"""q:[H,Tq,Dh], k/value:[H,Tk,Dh], allowed:[Tq,Tk]."""
score = q @ k.transpose(0, 2, 1) / np.sqrt(q.shape[-1])
score = np.where(allowed[None], score, -np.inf)
if not allowed.any(axis=-1).all():
raise ValueError("Every query must have an allowed key")
weights = softmax(score)
return weights @ value, weights
def lab_attention():
q = np.array([[[1.0, 0.0]]])
k = np.array([[[1.0, 0.0], [0.0, 1.0]]])
value = np.array([[[2.0, 0.0], [0.0, 4.0]]])
out, weight = attention(q, k, value, np.ones((1, 2), dtype=bool))
rng = np.random.default_rng(4)
x = rng.normal(size=(1, 5, 4))
triangular = np.tril(np.ones((5, 5), dtype=bool))
full, full_weights = attention(x, x, x, triangular)
changed = x.copy()
changed[:, 3:] += 100.0
perturbed, _ = attention(changed, changed, changed, triangular)
prefix_error = float(np.max(np.abs(full[:, :3] - perturbed[:, :3])))
assert prefix_error == 0.0
assert np.max(np.abs(full_weights[:, np.triu_indices(5, 1)[0],
np.triu_indices(5, 1)[1]])) == 0.0
np.testing.assert_allclose(full_weights.sum(axis=-1), 1.0, atol=1e-14)
return {"hand_example_weights": weight[0, 0].tolist(),
"hand_example_output": out[0, 0].tolist(),
"future_perturbation_prefix_max_error": prefix_error,
"upper_triangle_max_weight": 0.0}
def rms_norm(x):
return x / np.sqrt(np.mean(x * x, axis=-1, keepdims=True) + 1e-6)
def rope(x, positions):
"""x:[H,T,Dh]; positions:[T], Dh even; pairwise RoPE with base 10000."""
frequency = 10000.0 ** (-np.arange(0, x.shape[-1], 2) / x.shape[-1])
angles = positions[:, None] * frequency[None, :]
c, s = np.cos(angles)[None], np.sin(angles)[None]
result = np.empty_like(x)
even, odd = x[..., 0::2], x[..., 1::2]
result[..., 0::2] = even * c - odd * s
result[..., 1::2] = even * s + odd * c
return result
class TinyDecoder:
"""Two independent pre-norm layers, MHA, RoPE, SwiGLU; no trained checkpoint."""
def __init__(self):
rng = np.random.default_rng(11)
self.width, self.heads, self.head_dim, self.vocab = 8, 2, 4, 17
self.embedding = rng.normal(scale=0.3, size=(self.vocab, self.width))
self.output = rng.normal(scale=0.2, size=(self.width, self.vocab))
self.layers = []
for _ in range(2):
layer = {name: rng.normal(scale=0.2, size=(8, 8))
for name in ("q", "k", "v", "o")}
layer.update({"gate": rng.normal(scale=0.2, size=(8, 12)),
"up": rng.normal(scale=0.2, size=(8, 12)),
"down": rng.normal(scale=0.2, size=(12, 8))})
self.layers.append(layer)
def split_heads(self, x):
return x.reshape(-1, self.heads, self.head_dim).transpose(1, 0, 2)
def feed_forward(self, x, layer):
z = rms_norm(x)
gate = z @ layer["gate"]
silu = gate / (1 + np.exp(-gate))
return x + (silu * (z @ layer["up"])) @ layer["down"]
def full(self, ids):
x = self.embedding[ids].copy()
t = len(ids)
allowed = np.tril(np.ones((t, t), dtype=bool))
for layer in self.layers:
z = rms_norm(x)
q = rope(self.split_heads(z @ layer["q"]), np.arange(t))
k = rope(self.split_heads(z @ layer["k"]), np.arange(t))
v = self.split_heads(z @ layer["v"])
result, _ = attention(q, k, v, allowed)
x = x + result.transpose(1, 0, 2).reshape(t, self.width) @ layer["o"]
x = self.feed_forward(x, layer)
return rms_norm(x) @ self.output
def step(self, token, position, caches):
x = self.embedding[[token]].copy()
for layer_index, layer in enumerate(self.layers):
z = rms_norm(x)
q = rope(self.split_heads(z @ layer["q"]), np.array([position]))
k_new = rope(self.split_heads(z @ layer["k"]), np.array([position]))
v_new = self.split_heads(z @ layer["v"])
cache = caches[layer_index]
if cache is None:
k, v = k_new, v_new
else:
assert cache[0].shape[1] == position
k = np.concatenate([cache[0], k_new], axis=1)
v = np.concatenate([cache[1], v_new], axis=1)
caches[layer_index] = (k, v)
# Current query is the LAST position: every existing cached key is allowed.
result, _ = attention(q, k, v, np.ones((1, position + 1), dtype=bool))
x = x + result.transpose(1, 0, 2).reshape(1, self.width) @ layer["o"]
x = self.feed_forward(x, layer)
return (rms_norm(x) @ self.output)[0]
def lab_cache_equivalence():
model = TinyDecoder()
ids = np.array([1, 5, 3, 8, 2, 9, 4])
full = model.full(ids)
caches = [None] * len(model.layers)
cached = np.stack([model.step(int(token), i, caches)
for i, token in enumerate(ids)])
recomputed = np.stack([model.full(ids[:i + 1])[-1] for i in range(len(ids))])
error = float(np.max(np.abs(full - cached)))
np.testing.assert_allclose(full, cached, atol=1e-12, rtol=1e-12)
np.testing.assert_allclose(recomputed, cached, atol=1e-12, rtol=1e-12)
assert not np.allclose(caches[0][0], caches[1][0])
changed = ids.copy()
changed[4:] = [10, 11, 12]
future_error = float(np.max(np.abs(full[:4] - model.full(changed)[:4])))
assert future_error == 0.0
return {"configuration": {"layers": 2, "heads": 2, "head_dim": 4,
"width": 8, "ffn_width": 12, "vocab": 17,
"tokens": 7, "dtype": "float64",
"position": "pairwise RoPE", "ffn": "SwiGLU"},
"full_vs_cached_max_abs_logit_error": error,
"prefix_vs_cached_max_abs_logit_error": float(np.max(np.abs(recomputed - cached))),
"future_perturbation_prefix_max_error": future_error,
"per_layer_cache_shapes": [list(c[0].shape) for c in caches],
"abs_tolerance": 1e-12, "relative_tolerance": 1e-12}
def lab_dpo():
reference = np.array([-2.5, -2.5])
policy = np.array([-2.0, -3.0])
beta = 0.2
def fn(x):
margin = beta * ((x[0] - reference[0]) - (x[1] - reference[1]))
return float(np.logaddexp(0, -margin))
margin = beta * ((policy[0] - reference[0]) - (policy[1] - reference[1]))
scale = -beta / (1 + np.exp(margin))
analytic = np.array([scale, -scale])
numeric = finite_diff(fn, policy)
np.testing.assert_allclose(analytic, numeric, atol=1e-9, rtol=1e-8)
improved = fn(policy - 0.1 * analytic)
assert improved < fn(policy)
# Counterexample: both answers become less probable, but rejected falls more.
shifted = policy + [-1.0, -2.0]
assert fn(shifted) < fn(policy)
return {"policy_log_probs": policy.tolist(), "reference_log_probs": reference.tolist(),
"beta": beta, "margin": float(margin), "loss_nat": fn(policy),
"analytic_gradient": analytic.tolist(),
"max_abs_gradient_error": float(np.max(np.abs(analytic - numeric))),
"loss_after_gradient_step": improved,
"both_probabilities_lower_counterexample_loss": fn(shifted)}
def group_advantage(rewards, ddof=0, eps=1e-8):
return ((rewards - rewards.mean(axis=-1, keepdims=True)) /
(rewards.std(axis=-1, keepdims=True, ddof=ddof) + eps))
def lab_grpo():
rewards = np.array([[0.0, 0.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0],
[0.0, 0.0, 0.0, 0.0]])
population = group_advantage(rewards)
sample = group_advantage(rewards, ddof=1)
np.testing.assert_allclose(population[0], [-1, -1, 1, 1], atol=3e-8)
np.testing.assert_array_equal(population[1:], np.zeros((2, 4)))
order = [3, 0, 2, 1]
np.testing.assert_allclose(group_advantage(rewards[:, order]), population[:, order])
advantage = np.array([2.0, -2.0, 2.0, -2.0])
ratio = np.array([1.5, 1.5, 0.5, 0.5])
objective = np.minimum(ratio * advantage, np.clip(ratio, 0.8, 1.2) * advantage)
np.testing.assert_allclose(objective, [2.4, -3.0, 1.0, -1.6])
return {"rewards": rewards.tolist(), "population_std": rewards.std(axis=-1).tolist(),
"population_advantage": population.tolist(), "sample_advantage": sample.tolist(),
"epsilon": 1e-8, "ppo_clip_objectives": objective.tolist(),
"note": "advantages are frozen numeric values; this is not an actor training loop"}
def kl_divergence(p, q):
if np.any(p < 0) or np.any(q < 0) or not np.isclose(p.sum(), 1) or not np.isclose(q.sum(), 1):
raise ValueError("Inputs must be probability distributions")
positive = p > 0
if np.any(q[positive] == 0):
return float("inf")
return float(np.sum(p[positive] * np.log(p[positive] / q[positive])))
def lab_kl():
p, q = np.array([0.8, 0.2]), np.array([0.5, 0.5])
forward, reverse = kl_divergence(p, q), kl_divergence(q, p)
assert forward >= 0 and reverse >= 0 and not np.isclose(forward, reverse)
assert kl_divergence(p, p) == 0
assert np.isinf(kl_divergence(np.array([0.5, 0.5]), np.array([1.0, 0.0])))
return {"p": p.tolist(), "q": q.tolist(), "forward_kl_nat": forward,
"reverse_kl_nat": reverse, "identical_kl_nat": 0.0,
"zero_reference_support": "positive p on zero q gives infinity"}
def speculative_components(p, q):
# Never divide by q=0: such a token cannot be drawn as a draft candidate.
alpha = np.ones_like(q)
np.divide(p, q, out=alpha, where=q > 0)
alpha = np.minimum(alpha, 1.0)
accepted_mass = q * alpha
rejection_mass = float(1 - accepted_mass.sum())
positive = np.maximum(p - q, 0.0)
residual = positive / positive.sum() if positive.sum() > 0 else np.zeros_like(p)
final = accepted_mass + rejection_mass * residual
return alpha, accepted_mass, rejection_mass, residual, final
def lab_speculative(samples):
p, q = np.array([0.6, 0.3, 0.1]), np.array([0.4, 0.4, 0.2])
alpha, accepted, rejection, residual, final = speculative_components(p, q)
np.testing.assert_allclose(final, p, atol=1e-14)
# Support differences and equality exercise the zero-denominator branches.
for target, draft in [(p, p), (np.array([0.0, 0.7, 0.3]), np.array([0.3, 0.7, 0.0]))]:
np.testing.assert_allclose(speculative_components(target, draft)[-1], target, atol=1e-14)
rng = np.random.default_rng(73)
candidates = rng.choice(len(q), size=samples, p=q)
accept = rng.random(samples) < alpha[candidates]
result = candidates.copy()
result[~accept] = rng.choice(len(p), size=int((~accept).sum()), p=residual)
observed = np.bincount(result, minlength=len(p)) / samples
sigma = np.sqrt(p * (1 - p) / samples)
max_z = float(np.max(np.abs(observed - p) / sigma))
assert max_z < 6.0, "Empirical proportions exceeded an explicit six-sigma bound"
return {"target": p.tolist(), "draft": q.tolist(), "accept_probability": alpha.tolist(),
"accepted_output_mass": accepted.tolist(), "reject_probability": rejection,
"residual_distribution": residual.tolist(), "exact_output_distribution": final.tolist(),
"exact_max_abs_error": float(np.max(np.abs(final - p))),
"monte_carlo_seed": 73, "samples": samples, "observed_distribution": observed.tolist(),
"observed_acceptance_rate": float(accept.mean()), "max_standard_error_units": max_z,
"standard_error_bound_units": 6.0,
"note": "single-prefix acceptance law; not a full multi-token decoder or speed benchmark"}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output", type=Path,
default=Path(__file__).resolve().parents[1] / "qa" / "mechanism-results.json")
parser.add_argument("--samples", type=int, default=200_000)
args = parser.parse_args()
if args.samples < 10_000:
parser.error("--samples must be >= 10000 for the statistical check")
start = time.perf_counter()
labs = {"ce_gradient": lab_ce_gradient(), "masks": lab_masks(),
"causal_attention": lab_attention(), "cache_equivalence": lab_cache_equivalence(),
"dpo": lab_dpo(), "grpo": lab_grpo(), "kl": lab_kl(),
"speculative_sampling": lab_speculative(args.samples)}
report = {"status": "passed", "scope": "8 NumPy mechanism labs; no model training",
"timestamp_shanghai": datetime.now(timezone(timedelta(hours=8))).isoformat(),
"runtime": {"python": platform.python_version(), "numpy": np.__version__,
"platform": platform.platform(), "dtype": "float64"},
"elapsed_seconds": time.perf_counter() - start, "labs": labs}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"PASS: {len(labs)} mechanism labs")
print(f"CE finite-difference error: {labs['ce_gradient']['max_abs_gradient_error']:.3e}")
print(f"Cache logit max error: {labs['cache_equivalence']['full_vs_cached_max_abs_logit_error']:.3e}")
print(f"Speculative proportions: {labs['speculative_sampling']['observed_distribution']}")
print(f"Result: {args.output.resolve()}")
return 0
if __name__ == "__main__":
sys.exit(main())
5.3 V2补充实验：从算术对照到故障注入
2026-10-19实际运行的补充实验复用上方完整脚本，增加第08篇的带宽预算、第09篇的标量梯度对照，以及本篇三类cache故障与错误残差枚举。原脚本逐字保留。先把两份代码保存在同一目录，再运行：
python fault_injection.py --output v2-results.json
结果文件分别记录环境、各步中间数值和断言。这里的标量更新只演示归一化目标；cache故障是固定参数的前向检查，没有训练LLM、运行多卡或测量GPU吞吐。
完整V2补充源码：fault_injection.py
"""Supplemental mechanism checks; original mechanism_labs.py remains unchanged.
Run beside the original script: python fault_injection.py --output v2-results.json
Only scalar arithmetic and tiny fixed NumPy forward passes; no model training.
"""
from __future__ import annotations
import argparse
from datetime import datetime, timedelta, timezone
import json
from pathlib import Path
import platform
import numpy as np
import mechanism_labs as labs
def bandwidth_budget():
weight_bytes = 7_000_000_000 * 2
kv_bytes = 2 * 32 * 8192 * 8 * 128 * 2
assumed_bandwidth = 1_000_000_000_000
result = []
for batch in (1, 8):
read_bytes = weight_bytes + batch * kv_bytes
lower_seconds = read_bytes / assumed_bandwidth
result.append({"batch": batch, "read_bytes": read_bytes,
"lower_bound_ms": lower_seconds * 1000,
"aggregate_token_rate_ceiling": batch / lower_seconds,
"per_request_token_rate_ceiling": 1 / lower_seconds})
assert kv_bytes == 2**30
return {"scope": "self-defined streaming-read lower bound, not measured throughput",
"weight_bytes": weight_bytes, "per_request_kv_bytes": kv_bytes,
"assumed_bandwidth_bytes_per_second": assumed_bandwidth, "cases": result}
def unequal_rank_gradients():
# All features equal 1. Two targets are 1; the other six targets are 0.
targets = [np.ones(2), np.zeros(6)]
w, rate = 0.0, 0.4
p = 1 / (1 + np.exp(-w))
sums = np.array([(p - y).sum() for y in targets])
counts = np.array([len(y) for y in targets])
naive = float((sums / counts).mean())
correct = float(sums.sum() / counts.sum())
compensated_local = sums * len(targets) / counts.sum()
ddp_result = float(compensated_local.mean())
assert naive == 0 and correct == 0.25 and ddp_result == correct
new_w = w - rate * correct
new_p = 1 / (1 + np.exp(-new_w))
old_loss = float(np.log(2))
new_loss = float(-(2*np.log(new_p)+6*np.log1p(-new_p))/8)
assert new_loss < old_loss
return {"rank_token_counts": counts.tolist(), "rank_gradient_sums": sums.tolist(),
"rank_mean_gradients": (sums/counts).tolist(),
"naive_average": naive, "global_token_average": correct,
"compensated_local_gradients": compensated_local.tolist(),
"ddp_average_after_compensation": ddp_result,
"learning_rate": rate, "old_w": w, "correct_new_w": new_w,
"new_probability": float(new_p), "old_global_loss": old_loss,
"correct_new_global_loss": new_loss,
"naive_new_global_loss": old_loss,
"scope": "one scalar logistic-regression step, not a multigpu execution"}
def cache_faults():
ids = np.array([1, 5, 3, 8, 2, 9, 4])
model = labs.TinyDecoder()
expected = model.full(ids)
def run(mode):
caches = [None, None]
outputs = []
original_rope, original_attention = labs.rope, labs.attention
if mode == "reset_rope_position":
labs.rope = lambda x, positions: original_rope(x, np.zeros_like(positions))
elif mode == "top_left_rectangular_mask":
labs.attention = lambda q,k,v,allowed: original_attention(
q,k,v,np.tril(np.ones((q.shape[1],k.shape[1]),dtype=bool)))
try:
for pos, token in enumerate(ids):
outputs.append(model.step(int(token), pos, caches))
if mode == "copy_layer1_cache_to_layer2":
caches[1] = (caches[0][0].copy(), caches[0][1].copy())
finally:
labs.rope, labs.attention = original_rope, original_attention
actual = np.stack(outputs)
shape_ok = all(c[0].shape == (2,7,4) and c[1].shape == (2,7,4) for c in caches)
error = float(np.max(np.abs(expected-actual)))
ownership_ok = not np.allclose(caches[0][0], caches[1][0])
return {"mode": mode, "shape_check_passed": shape_ok,
"per_layer_cache_shapes": [list(c[0].shape) for c in caches],
"max_absolute_logit_error": error,
"logit_allclose_passed": bool(np.allclose(expected,actual,atol=1e-12,rtol=1e-12)),
"distinct_layer_keys_passed": ownership_ok}
cases = [run(m) for m in ("correct", "reset_rope_position",
"copy_layer1_cache_to_layer2", "top_left_rectangular_mask")]
assert cases[0]["logit_allclose_passed"]
assert all(c["shape_check_passed"] for c in cases)
assert all(not c["logit_allclose_passed"] for c in cases[1:])
assert not cases[2]["distinct_layer_keys_passed"]
return {"tokens": ids.tolist(), "atol":1e-12,"rtol":1e-12,"cases":cases}
def wrong_rejection_fallback():
p,q = np.array([0.6,0.3,0.1]),np.array([0.4,0.4,0.2])
alpha,accepted,reject,residual,correct = labs.speculative_components(p,q)
wrong = accepted + reject*p
difference = wrong-p
assert np.allclose(correct,p,atol=1e-14)
assert np.allclose(wrong,[0.52,0.36,0.12])
for target,draft in ((p,p),(np.array([0.0,0.7,0.3]),np.array([0.3,0.7,0.0]))):
np.testing.assert_allclose(labs.speculative_components(target,draft)[-1],target,atol=1e-14)
return {"target":p.tolist(),"draft":q.tolist(),"accepted_mass":accepted.tolist(),
"reject_mass":reject,"correct_residual":residual.tolist(),
"correct_output":correct.tolist(),"wrong_output":wrong.tolist(),
"wrong_minus_target":difference.tolist(),
"wrong_l1_distance":float(np.abs(difference).sum()),
"wrong_total_variation_distance":float(np.abs(difference).sum()/2),
"scope":"exact single-prefix mass enumeration, not Monte Carlo"}
def main():
parser=argparse.ArgumentParser(description=__doc__)
parser.add_argument('--output',type=Path,default=Path('v2-results.json'))
args=parser.parse_args()
report={"status":"passed","timestamp_shanghai":datetime.now(timezone(timedelta(hours=8))).isoformat(),
"runtime":{"python":platform.python_version(),"numpy":np.__version__,
"platform":platform.platform(),"dtype":"float64"},
"bandwidth_budget":bandwidth_budget(),"unequal_rank_gradients":unequal_rank_gradients(),
"cache_faults":cache_faults(),"wrong_rejection_fallback":wrong_rejection_fallback()}
args.output.parent.mkdir(parents=True,exist_ok=True)
args.output.write_text(json.dumps(report,ensure_ascii=False,indent=2),encoding='utf-8')
print(json.dumps(report,ensure_ascii=False,indent=2))
if __name__=='__main__':
main()
正在连接 GitHub 评论…