在昇腾NPU上开发自定义算子,最怕的不是写不出来,而是写出来了不知道对不对。数值正确性、边界条件、性能回归,这三件事任何一件没做好,上线之后就是生产事故。monocase是CANN开源社区里专门解决这个问题的工具。

monocase在CANN里的位置

monocase是昇腾CANN开源社区里的算子验证框架,和golden-stone、op-test这些仓库并列,属于"测试与验证工具"这一类。

从CANN五层架构来看,monocase位于第2层和第3层之间——它既依赖opbase提供的基础算子接口(第2层),又对接BiSheng编译器和GE图引擎(第3层),在算子编译之后、上线之前做验证。

实际使用中,monocase的定位是:算子开发完成之后,上线之前,做最后一道验证关卡。它覆盖的验证维度包括:数值正确性(和Golden数据对比)、边界条件(特殊输入、大输入、小输入)、性能回归(和基准性能对比)、API兼容性(算子接口是否发生变化)。

为什么需要专门的验证框架

在CPU和GPU上开发算子,通常有成熟的验证工具:CPU上有Google Test + Eigen的参考实现,GPU上有CUDA官方提供的kernel示例可以作为Golden。

昇腾NPU上的情况不一样:

  1. Golden数据怎么来:NPU的算子用Ascend C写,跑在达芬奇架构上,中间计算过程(特别是Cube Unit的矩阵乘)和CPU的参考实现可能有细微差异(比如累加顺序不同导致 rounding 误差不同)。直接用CPU参考实现做Golden,可能会因为合理的数值差异导致误报。

  2. 边界条件覆盖成本高:NPU上的算子要处理各种特殊情况(输入是NaN/Inf、输入维度不是16的倍数、输入数据量超出UB大小),这些边界条件手动写测试用例成本很高。

  3. 性能回归难量化:算子改了一行代码,性能是提升了还是下降了,需要和标准实现做对比。这个对比需要控制变量(同样的输入数据、同样的硬件环境、同样的编译选项),手动做很麻烦。

monocase就是用来系统化解决这三个问题的。

monocase的验证流程

monocase的验证流程分为四个阶段:

阶段1:Golden数据生成

用CPU上的参考实现(通常是Eigen或者手动写的朴素实现)生成一组输入数据和对应的输出数据,作为Golden。这一步骤在CPU上跑,保证数值正确性的基准。

阶段2:NPU算子执行

把同样的输入数据送到NPU上,调用待验证的算子,拿到NPU上的输出数据。

阶段3:数值对比

把NPU输出和Golden output做对比,计算绝对误差和相对误差。monocase支持多种对比标准:严格相等(整数算子)、允许小误差(浮点算子)、统计学对比(大矩阵的输出,逐元素对比成本太高,用均值、方差、最大值等统计量对比)。

阶段4:边界条件和性能测试

在数值正确性通过之后,monocase会自动生成一组边界条件的测试用例(特殊输入、大输入、小输入),以及一组性能基准测试(和基准性能对比,检查是否有回归)。

代码示例:用monocase验证自定义算子

下面给一个完整的monocase验证示例,假设我们要验证一个自定义的MatMul算子:

# monocase 验证示例:自定义 MatMul 算子
import torch
import torch_npu
from monocase import OpValidator, GoldenGenerator

# 定义待验证的算子(封装成 callable)
def my_matmul(a, b):
    """自定义的 MatMul 算子,内部调用 Ascend C kernel"""
    return torch.matmul(a, b)  # 实际场景替换为自定义算子调用

# 1. Golden 数据生成
golden_gen = GoldenGenerator(op_name="MatMul")
# 生成 10 组随机输入,数据类型 FP16,形状随机
test_cases = golden_gen.generate(
    num_cases=10,
    dtype=torch.float16,
    shape_range=[(1, 128), (128, 4096)]  # M/N/K 的范围
)

# 2. 创建验证器
validator = OpValidator(
    op_under_test=my_matmul,
    golden_source="cpu_eigen",  # 用 Eigen 的 CPU 实现做 Golden
    tolerance=1e-3  # 相对误差容忍度
)

# 3. 执行验证
results = validator.validate(test_cases)

# 4. 输出验证结果
for i, result in enumerate(results):
    status = "PASS" if result.is_pass() else "FAIL"
    max_err = result.max_abs_error()
    print(f"Test case {i}: {status}, max abs error={max_err:.6f}")

# 5. 生成验证报告
validator.generate_report(output_path="./matmul_validation_report.html")

这段代码展示了monocase的核心工作流:生成Golden数据 → 执行NPU算子 → 数值对比 → 生成报告。

边界条件测试

monocase的一个强项是自动生成边界条件的测试用例。下面给一个边界条件测试的示例:

# 边界条件测试示例
from monocase import BoundaryTester

# 创建边界条件测试器
boundary_tester = BoundaryTester(op_name="MatMul")

# 添加边界条件测试用例
# 1. 输入包含 NaN
boundary_tester.add_case(
    name="input_with_nan",
    a=torch.tensor([1.0, float('nan'), 3.0]),
    b=torch.tensor([2.0, 3.0, 4.0]),
    expected_behavior="output_should_contain_nan"
)

# 2. 输入维度不是 16 的倍数(NZ 格式的对齐要求)
boundary_tester.add_case(
    name="dimension_not_aligned",
    a=torch.randn(17, 32),  # M=17 不是 16 的倍数
    b=torch.randn(32, 64),
    expected_behavior="should_pad_or_error"
)

# 3. 输入数据量超出 UB 大小
boundary_tester.add_case(
    name="input_exceeds_ub",
    a=torch.randn(4096, 4096),  # 大矩阵
    b=torch.randn(4096, 4096),
    expected_behavior="should_split_and_compute"
)

# 4. 输入是零矩阵
boundary_tester.add_case(
    name="zero_input",
    a=torch.zeros(128, 128),
    b=torch.randn(128, 128),
    expected_behavior="output_should_be_zero"
)

# 执行边界条件测试
boundary_results = boundary_tester.run_all(my_matmul)

# 输出结果
for result in boundary_results:
    print(f"{result.name}: {result.status}")

这段代码会自动测试各种边界条件,检查算子在这些特殊情况下的行为是否符合预期。

性能回归测试

monocase还支持性能回归测试——算子改了之后,性能不能比基准实现差。下面给一个性能回归测试的示例:

# 性能回归测试示例
from monocase import PerformanceRegTester

# 创建性能回归测试器
perf_tester = PerformanceRegTester(
    op_name="MatMul",
    baseline_source="ops_blas_gemm",  # 基准实现:ops-blas 的 GEMM
    regression_threshold=0.1  # 允许 10% 的性能回退
)

# 添加性能测试用例
perf_test_cases = [
    {"M": 1, "N": 11008, "K": 4096},   # Decode 阶段
    {"M": 128, "N": 4096, "K": 4096},   # Prefill 阶段
    {"M": 32, "N": 4096, "K": 128},     # Q 投影
]

for case in perf_test_cases:
    perf_tester.add_case(**case)

# 执行性能回归测试
perf_results = perf_tester.run_all(my_matmul)

# 输出结果
for result in perf_results:
    status = "PASS" if result.is_pass() else "FAIL"
    speedup = result.speedup_vs_baseline()
    print(f"Case {result.case_name}: {status}, speedup={speedup:.2f}x")

这段代码会把自定义算子和ops-blas的通用GEMM做性能对比,检查是否有回归。

monocase和op-test的关系

monocase和op-test这两个仓库容易搞混。从功能定位上看:

  • monocase:系统化的算子验证框架,覆盖数值正确性、边界条件、性能回归
  • op-test:快速的算子冒烟测试,主要检查算子能不能跑通、API是否正确

实际使用中,op-test适合在算子开发过程中快速迭代(改一点代码,跑一下op-test,看有没有crash);monocase适合在算子开发完成之后做系统化的验证(跑完整的一套测试用例,生成验证报告)。

两者的依赖关系是:monocase → op-test → opbase(基础算子接口)。

和CI/CD的集成

monocase支持和CI/CD流水线集成。下面给一个和GitLab CI集成的示例:

# .gitlab-ci.yml 示例:集成 monocase 验证
stages:
  - build
  - test
  - validate  # 新增验证阶段

# 算子验证任务
op_validate:
  stage: validate
  script:
    - source /usr/local/Ascend/ascend-toolkit/setenv.bashrc
    - python -m pip install monocase
    - cd my_op_project
    # 跑 monocase 验证
    - python run_monocase.py --op_name=my_matmul --output=report.html
    # 检查验证结果
    - python check_monocase_result.py --report=report.html --fail_on_error
  artifacts:
    paths:
      - my_op_project/report.html
    expire_in: 1 week
  only:
    - merge_requests
    - main

这个配置会在每次MR和main分支提交的时候,自动跑monocase验证,生成验证报告,并且如果验证失败就阻止合并。

踩过的几个坑

第一个坑是Golden数据的数值差异。一开始用CPU的Eigen实现做Golden,发现和NPU算子的输出对比,相对误差有1e-3左右,超过了预设的1e-5阈值。排查之后发现,Eigen的矩阵乘累加顺序和NPU的Cube Unit不一样,导致rounding误差不同。解法是把容忍度放宽到1e-3,或者用NPU上已经验证过的算子(比如ops-blas的GEMM)做Golden。

第二个坑是边界条件测试的用例生成策略。monocase支持随机生成边界条件测试用例,但随机生成可能会漏掉一些关键的边界情况。解法是手动添加关键边界情况(比如维度不是16的倍数、输入是零矩阵),和随机生成结合使用。

第三个坑是性能回归测试的硬件环境一致性。性能测试需要在完全一样的硬件环境上跑,如果有其他进程在跑,会干扰性能测试结果。解法是把性能测试放在专用的测试机器上跑,并且每次测试之前清掉所有干扰进程。

总结

monocase是昇腾CANN里保证算子质量的关键工具。它系统化了算子验证的流程:Golden数据生成、NPU算子执行、数值对比、边界条件测试、性能回归测试。

在昇腾NPU上做自定义算子开发,monocase是上线之前必须过的最后一道关卡。CANN开源之后,monocase的验证规则完全透明,也可以根据项目需求自定义验证规则。

如果你正在做昇腾上的算子开发,建议把monocase集成到开发流程里,早点发现数值正确性和性能回归问题,避免上线之后踩坑。

仓库地址:https://atomgit.com/cann/monocase

Logo

作为“人工智能6S店”的官方数字引擎,为AI开发者与企业提供一个覆盖软硬件全栈、一站式门户。

更多推荐