企业官网建设流程全解析

• 算子库
• 人工智能
• 深度学习
• Ascend

导读

本指南围绕 CANN ops-transformer 仓库中attention/recurrent_gated_delta_rule算子的 pytest 测试框架展开，它覆盖了该算子在 NPU 上功能验证的完整链路：CPU 侧复现算子逻辑生成 golden 数据、NPU 侧通过 TorchNPU 直调算子、以及两者之间的精度对比。读完本文，你将掌握 single / rdv / random / random_npu / mss 五种测试模式的用法、随机用例生成规则、双层种子复现机制，以及 mssanitizer 设备侧内存检测的接入方式，可直接用于该算子的日常调测与回归验证。

一、框架定位与整体架构

1.1 要解决什么问题

RecurrentGatedDeltaRule（循环门控 Delta 规则，RGDR）是一种应用于线性注意力机制与循环神经网络的算子，在每个时间步根据输入q_t、k_t、v_t与上一隐藏状态S_{t-1}计算输出o_t与新状态S_t（算法公式见 算子 README）。由于计算过程带循环依赖，NPU kernel 的正确性验证不能只靠"跑通"，必须逐元素与可靠参照实现比对。该测试框架正是为此设计：

• CPU 侧：复现算子功能用以生成 golden 数据；
• NPU 侧：通过 TorchNPU 进行算子直调获取实际数据；
• 精度对比：进行 CPU 与 NPU 结果的精度对比验证算子功能。

1.2 与算子工程的关系

该测试框架位于算子的tests/pytest/目录，与算子源码同仓管理：

• 算子接口声明：recurrent_gated_delta_rule.h 定义了l0op::RecurrentGatedDeltaRule的入参顺序（query、key、value、beta、stateRef、actualSeqLengths、ssmStateIndices、g、gk、numAcceptedTokens、scaleValue）；
• 测试中通过torch_npu.npu_recurrent_gated_delta_rule(...)以 eager 方式直调该接口（见下文 golden 脚本）；
• 算子功能、参数表与产品支持情况见 算子 README 及 aclnnRecurrentGatedDeltaRule 文档。

二、当前实现范围：参数约束速查

测试框架当前支持的参数范围如下，这也是所有用例（含随机用例）生成时必须遵守的边界：

三、环境配置与前置条件

3.1 前置要求

1. 安装 TorchNPU 安装包（需及时更换为最新版本），官方安装教程见 Ascend/pytorch；
2. 完成环境安装和环境变量配置，具体操作参考仓库根目录的 README。

3.2 custom 包调用

框架支持 custom 包调用方式，即在编译安装好的 CANN 环境上以 custom 算子包形式加载本算子后运行测试。

四、pytest 文件结构与职责

tests/pytest/ ├── test_run.sh # 执行脚本（五种模式入口） ├── conftest.py # pytest 钩子：逐用例记录参数/结果/精度/种子，会话结束落 CSV ├── recurrent_gated_delta_rule_golden.py # CPU 侧 golden 实现 + CPU/NPU 精度对比 ├── pytest.ini # pytest 基础配置 ├── test_recurrent_gated_delta_rule_single.py # 单用例运行主程序（参数化入口） ├── recurrent_gated_delta_rule_operator_single.py# CPU 侧算子逻辑 + NPU 算子直调封装 ├── test_recurrent_gated_delta_rule_paramset.py # 单用例入参配置 └── test_recurrent_gated_delta_rule_paramset_rdv.py # RDV 测试入参配置

各文件职责如下：

• test_run.sh：shell 执行入口，根据第一个参数分发到 single / rdv / random / random_npu / mss 分支，自动创建output/目录、生成时间戳日志与 CSV 路径，并通过tee保存屏显；
• conftest.py：两条核心钩子。_set_tensor_seed夹具在每条用例前固定torch与torch.npu的随机种子（TORCH_SEED指定则用指定值，否则自动生成并回写）；pytest_runtest_makereport逐用例采集结果、入参、精度 PctRlt、错误信息，pytest_sessionfinish在会话结束写入 CSV；
• recurrent_gated_delta_rule_golden.py：包含 CPU golden 实现cpu_recurrent_gated_delta_rule、数据生成器rand_range、精度对比函数check_result以及 eager 直调封装run_recurrent_gated_delta_rule_eager；
• test_recurrent_gated_delta_rule_single.py：定义param_names与随机参数生成器_generate_random_param_dict，用@pytest.mark.ci+@pytest.mark.parametrize将用例参数化执行；
• pytest.ini：基础 pytest 配置（log_format = %(message)s），标记ci由主程序文件中的@pytest.mark.ci使用。

4.1 主流程代码骨架

执行链路为：test_run.sh设置TEST_MODE与CSV_FILE→test_recurrent_gated_delta_rule_single.py根据模式构建参数组合 →recurrent_gated_delta_rule_operator_single.output_operator解包参数 →run_recurrent_gated_delta_rule_eager依次完成参数校验、数据生成、CPU golden、NPU 直调与精度断言。核心直调代码如下：

npu_out = torch_npu.npu_recurrent_gated_delta_rule( query, key, value, init_state, beta=beta, scale=scale_value, actual_seq_lengths=act_seq_len, ssm_state_indices=ssm_state_indices, num_accepted_tokens=num_accepted_tokens, g=g, gk=gk, )

五、五种测试模式与使用方法

在tests/pytest/目录下执行test_run.sh，不同参数对应不同模式。

5.1 单用例调测（single）

1. 手动配置 test_recurrent_gated_delta_rule_paramset.py 中的参数（文件内置Testcase0、Testcase1两组，通过ENABLED_PARAMS启用）；
2. 执行：

bash test_run.sh single

5.2 RDV 测试（rdv）

RDV 模式使用 test_recurrent_gated_delta_rule_paramset_rdv.py 中预定义的参数池ENABLED_PARAMS_RDV，覆盖 batch_size × mtp × dk × dv 组合、nk/nv 比例矩阵、大 shape（如 dk=dv=512）以及 state 为 fp32 / 非连续（non-contiguous）的变体，适用于批量回归：

bash test_run.sh rdv

从源码看，rdv 参数池由GROUP_1~GROUP_6与L0_L1_CASES拼接而成，并通过_gen_fp32_state_cases、_gen_non_contiguous_state_cases派生 state 数据类型与内存布局的交叉组合，最终用例规模为_ALL_BF16_CASES × 4。

5.3 随机用例测试（random）

随机生成 N 条用例并执行（含 CPU golden 精度对比），可用RANDOM_SEED环境变量固定随机种子复现（不指定则自动生成并记录到 CSV）：

bash test_run.sh random 100

5.4 随机用例测试（仅 NPU，random_npu）

随机生成 N 条用例，设置SKIP_GOLDEN=1跳过 CPU golden 计算与精度对比，仅执行 NPU 算子。由于输入张量直接在 NPU 上生成，host 内存占用低，执行速度更快：

bash test_run.sh random_npu 100

5.5 mssanitizer 检测（mss）

随机生成 N 条用例，仅 NPU 执行（不跑 golden），并在 mssanitizer 下运行，检测设备侧内存越界、非法地址访问等问题。屏显含====== ERROR即判定 FAIL（脚本退出码为 1）。默认只检测本算子 kernel（--kernel-name=RecurrentGatedDeltaRule），跳过 ZerosLike/ViewCopy/TensorMove/rand 等旁路 kernel 的检测开销：

bash test_run.sh mss 10

mssanitizer 安装（前置条件）：

• 官方快速入门文档：mssanitizer 快速入门；
• CANN 工具包已自带时无需额外安装（如/home/developer/Ascend/cann-9.2.0/bin/mssanitizer，确保其所在目录在 PATH 中）；
• 未自带时源码编译安装：

git clone https://gitcode.com/Ascend/mssanitizer.git cd mssanitizer python3 build.py # 编译完成后安装 run 包 artifacts/mindstudio-sanitizer_26.0.0_x86_64.run --run --install-path=/home/developer/Ascend/

mss 模式环境变量定制：

MSS_TOOL=racecheck bash test_run.sh mss 10 # 换检测工具(memcheck/racecheck/initcheck/synccheck) MSS_KERNEL='' bash test_run.sh mss 10 # 关闭kernel过滤，检测全部kernel MSS_EXTRA_OPTS='--leak-check=yes --full-backtrace=yes' bash test_run.sh mss 10 # 追加mssanitizer参数 MSSANITIZER_BIN=/path/to/mssanitizer bash test_run.sh mss 10 # 指定mssanitizer路径

从 test_run.sh 源码看，mss 模式还有两个值得关注的实现细节：_mss_preflight前置校验会先跑 1 条用例确认 mssanitizer 真正拦截到 kernel（日志中出现[mssanitizer] Start...sanitizer on kernel），否则直接中止，避免"检测无效但显示通过"的假象；MSS_BATCH分批模式下每批使用seed+i独立种子重启 mssanitizer，避免 host 内存累积，CSV/log 追加写入单一文件。

六、随机用例生成规则

random / random_npu / mss 三种模式共用同一套随机参数生成器_generate_random_param_dict（见 test_recurrent_gated_delta_rule_single.py），在算子约束内从 0 随机生成，不依赖 single/rdv 参数池。每条用例的入参生成规则如下（按算子接口入参顺序）：

其他固定项：data_type 固定 bfloat16；query/key_datarange 固定 [-1,1]；beta_datarange 固定 [0,1]；state_datarange 固定 [-10,10]。

shape 约束：0<Nk≤256、0<Nv≤256 且 Nv≥Nk 且 Nv%Nk==0、0<Dk≤512、0<Dv≤512、mtp≤8、BlockNum≥T。

内存约束：Dk×Dv 受 state 元素数上限_STATE_ELEM_CAP=2.0B约束（budget = STATE_ELEM_CAP // (BlockNum × Nv)），防止单进程 host OOM。源码注释给出了估算依据：单 case host 峰值 ≈ baseline(1.6GB) + 10×state 元素字节数（bf16 state 最坏情形下原张量 2E +.to(fp32)4E +.clone()4E = 10E），取 2.0B 时 host 峰值约 21.6GB，不超出 24GB 上限；NPU HBM 远不构成瓶颈。

随机种子机制：

• RANDOM_SEED控制 shape/参数序列（一个 seed 对应一组确定的 N 条用例参数）；
• TORCH_SEED控制张量数值（每条用例独立，conftest 自动生成并记 CSV）；
• 不设RANDOM_SEED时自动生成并回写os.environ，conftest 落 CSV；
• 复现：RANDOM_SEED=<seed> bash test_run.sh random N（整批复现）或TORCH_SEED=<tensor_seed>配合 CSV 入参（单条数值级复现）。

重复概率：参数空间约 5×10¹⁰（500 亿），10000 条撞车概率 <0.1%，实际不会重复。

七、结果输出与复现

所有模式执行后均输出到output/目录（已 gitignore）：

• run_<时间戳>.log/mss_<时间戳>.log：完整执行日志（tee 屏显）；
• result_<时间戳>.csv：逐用例结果表，每行一条用例，列含义如下：

关于mss_check回填的实现：由于 mssanitizer 报错时 pytest 仍可能 PASSED，test_run.sh 中的_backfill_mss_check函数会解析日志中====== ERROR块的个数并回填到 CSV 对应列；分批模式下_merge_batch_csvs负责将各批次 CSV 按顺序合并为单一文件。

失败用例复现（两层种子配合 CSV 入参）：

# 整批复现（同 shape 序列）：CSV 取 random_seed RANDOM_SEED=<random_seed> bash test_run.sh random N # 单条数值级复现（同 shape + 同张量数值）：CSV 取入参与 tensor_seed TORCH_SEED=<tensor_seed> bash test_run.sh random 1

八、环境变量汇总

九、日志与 CSV 自动保存

所有模式均通过tee自动保存屏显到output/目录（已 gitignore）：

十、精度对比实现要点

golden 侧精度对比的核心是 recurrent_gated_delta_rule_golden.py 中的check_result函数，其判定策略可归纳为三层：

1. 达标率门槛：使用np.isclose(rtol, atol, equal_nan=True)逐元素比对，统计PctRlt = 通过元素数 / 总数 × 100%，默认pct_thd=0.005（即允许 0.5% 比例不达标）；
2. 误差上限保护：即使达标率满足，若归一化相对误差max_error >= max_diff_hd(10.0)仍判失败，防止个别元素严重溢出；
3. 溢出监控：golden 侧出现 inf/nan 时单独上报并计数。

对 bf16 使用rtol=0.0078125, atol=0.0001，其他类型使用rtol=0.005, atol=0.000025；大张量按1<<22分块比对以控制 host 峰值内存。run_recurrent_gated_delta_rule_eager在最后对输出与 state 两个结果分别断言Pass，任一失败即用例失败。

十一、与算子实现的衔接：从测试回看算子语义

测试框架中 golden 的计算流程与 算子 README 中的公式严格对应：每个时间步先对状态施加门控衰减S = S * alpha * alphak，再按 delta 规则更新S = S + (v - S·k) * beta * k^T，最后o = S·q / sqrt(Dk)（cpu_recurrent_gated_delta_rule中q = q * scale_value即对应1/sqrt(d_k)缩放）。state 在测试中以stateRef语义原地更新（npu_state_out = init_state），ssm_state_indices将每个时间步映射到对应状态块，num_accepted_tokens用于从 batch 中已接受 token 位置初始化状态——这些行为均可从测试入参与 golden 实现直接观察到。若需进一步阅读算子侧实现，可参考 op_host 目录 下的 def/infershape/tiling 源文件与 op_kernel 目录 的 kernel 实现。

十二、快速上手指南

1. 在仓库根目录完成 README 所述的环境安装与配置，并安装 TorchNPU；
2. 首次验证推荐先跑单用例：配置 test_recurrent_gated_delta_rule_paramset.py 后执行bash test_run.sh single；
3. 需要批量回归时执行bash test_run.sh rdv；
4. 泛化验证用RANDOM_SEED=<seed> bash test_run.sh random 100，需要快速执行（不做精度对比）用bash test_run.sh random_npu 100；
5. 内存安全检测在安装 mssanitizer 后执行bash test_run.sh mss 10，并可用MSS_TOOL/MSS_BATCH等环境变量定制；
6. 所有结果均自动落盘output/result_<timestamp>.csv，失败用例按 CSV 中的random_seed与tensor_seed两层种子即可精确复现。

• 算子库
• 人工智能
• 深度学习
• Ascend