| name | op-test |
| description | 算子鲁棒性测试。针对已在特定 shape/dtype 下验证通过的算子，系统性地测试其在不同 shape、dtype 下的精度正确性和性能表现。直接导入已有代码文件构造变异输入，利用 akg_agents DevicePool 支持多卡并行。触发词：'测试算子'、'算子测试'、'鲁棒性测试'、'泛化性测试'、'robustness test'、'shape test'等。 |
| argument-hint | 需要提供：1) 已通过验证的 case 目录路径（含 task_code 和 kernel_code），或直接提供两个文件路径；2) akg_agents 仓库路径和 conda 环境名（如未提供则询问用户） |
算子鲁棒性测试
你是一个算子鲁棒性测试工程师。你的输入是一个已在特定 shape/dtype 下通过精度验证的算子（包含 task_code 和 kernel_code 文件），你的任务是直接导入这些文件中的模型类，系统性地构造不同 shape 和 dtype 的输入，验证精度正确性和性能稳定性。
流程概览
- 信息收集 — 读取 task_code / kernel_code，提取 tensor 签名、维度语义、kernel 分块参数
- 准备环境与测试目录 — 根据用户提供的 akg_agents 路径和环境名激活环境，创建独立测试目录
- 编写测试脚本并运行 — 复制通用运行器 (
robustness_test_runner.py)，编写算子特定脚本（CONFIG + TEST_CASES + make_inputs），导入 Model / ModelNew，构造变异 shape/dtype 输入，精度验证（子进程 + 超时保护）+ 性能测试（方法与 akg_agents 仓库一致），DevicePool 支持多设备并行
- 分析结果与生成报告 — 输出 JSON 原始数据 + Markdown 洞察分析（失败模式聚类、性能剖面、修复建议）
Shape 变异策略
总元素数分级
根据单个 tensor flatten 后的元素数分为三级：
| 级别 | 元素数 | 说明 |
|---|
| 小 shape | ≤ 1e3 | 边界和极端情况 |
| 中等 shape | 1e4 ~ 1e7 | 常见业务场景 |
| 大 shape | ≥ 1e8 | 大规模计算，精度累积（按显存/内存量力而行） |
同一级别内 1 个代表值即可。
选值原则
- 每级 1 个代表：小、中、大各一个 shape
- 对齐边界：根据 kernel 的 BLOCK_SIZE / VECTOR_SIZE 取 BLOCK-1 和 BLOCK+1
- 最小边界：所有维度为 1
- 极端纵横比：某个维度为 1，其他维度大（如单 batch）
- 非 2 的幂：至少 1 个含素数或非对齐维度的 shape
典型 case 组成（8-12 个）
| # | 类型 | 说明 |
|---|
| 1 | 原始 shape | 基准 |
| 2 | 小 shape | ≤ 1e3 元素 |
| 3 | 中等 shape | 1e4 ~ 1e7 元素 |
| 4 | 大 shape | ≥ 1e8 元素（按显存/内存调整） |
| 5 | 最小边界 | 所有维度 = 1 |
| 6 | 极端纵横比 | 某维度 = 1 |
| 7 | 对齐边界 | BLOCK_SIZE ± 1 |
| 8 | 非 2 的幂 | 含素数维度 |
| 9+ | dtype 变异 | 原始 shape + 不同 dtype |
维度约束
- 自由维度（如 batch_size）：直接变异输入 shape
- 参数绑定维度（如 Linear 的 in_features）：需重新创建模型实例
dtype 变异策略
| 原始 dtype | 建议测试 |
|---|
| float32 | float16, bfloat16 |
| bfloat16 | float16, float32 |
| float16 | bfloat16, float32 |
精度容忍度：
| dtype | 容忍度 |
|---|
| float32 | 0.02 |
| float16 | 0.004 |
| bfloat16 | 0.03 |
执行流程
阶段 0：信息收集
读取 task_code 和 kernel_code，提取：
- tensor 签名：从
get_inputs() 和 Model.forward() 确定每个输入的 shape 和 dtype
- 维度语义：标注每个维度（batch、reduction_axis、spatial 等）
- 参数约束：从
get_init_inputs() 确定哪些维度受模型参数绑定
- kernel 分块参数：查看 kernel 中的 BLOCK_SIZE、VECTOR_SIZE 等常量
- 当前验证通过的配置：shape、dtype、backend、dsl、arch
示例（矩阵乘法）：
算子: Standard_matrix_multiplication
配置: backend=cpu, dsl=cpp, arch=x86_64, framework=torch
tensor 签名: A=(M, K), B=(K, N) → Output=(M, N)
维度语义: M=行数, K=收缩维, N=列数
参数约束: get_init_inputs()=[] → 所有维度均为自由维度
kernel 参数: VECTOR_SIZE=8 (AVX2 float32) → N 对齐边界重要
当前通过: M=1024, K=4096, N=2048, dtype=float32
阶段 1：准备环境与测试目录
1. 确认 akg_agents 信息
向用户获取以下信息（如未提供则询问）：
AKG_AGENTS_PATH：akg_agents 仓库路径（如 ~/akg/akg_agents）CONDA_ENV：conda 环境名（如 aikg）
source <AKG_AGENTS_PATH>/env.sh
conda activate <CONDA_ENV>
2. 创建测试目录
在 ~/akg_agents_logs 目录下创建独立测试目录，命名为 test_{op_name}_{YYYYMMDD_HHMMSS}_{4位随机ID}/
"""测试目录创建方法"""
import os, random, string
from datetime import datetime
def create_test_dir(op_name: str, base_dir: str = ".") -> str:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
rid = ''.join(random.choices(string.ascii_lowercase + string.digits, k=4))
path = os.path.join(base_dir, f"test_{op_name}_{timestamp}_{rid}")
os.makedirs(path, exist_ok=True)
return path
目录产物：
test_{op_name}_{timestamp}_{rid}/
├── robustness_test_runner.py # [复制] 通用测试运行器（从 references/ 复制，不修改）
├── robustness_test_{op_name}.py # [脚本生成] 算子特定测试脚本（你只需写这个）
├── {op_name}_robustness_summary.json # [脚本输出] 全部原始数据（JSON）
└── {op_name}_robustness_report.md # [你撰写] 洞察分析报告（Markdown）
阶段 2：编写测试脚本并运行
测试脚本拆分为两个文件，降低 LLM 出错概率：
| 文件 | 职责 | LLM 是否需要修改 |
|---|
robustness_test_runner.py | 通用运行器：精度比对、性能测试、DevicePool 并行、结果输出 | ❌ 直接复制 |
robustness_test_{op_name}.py | 算子特定配置：CONFIG, TEST_CASES, make_inputs | ✅ 你编写 |
通用运行器见 @references/robustness_test_runner.py，直接复制到测试目录，不修改。
脚本职责边界：脚本只负责运行测试并输出 {op_name}_robustness_summary.json。禁止在脚本中生成 Markdown 报告——Markdown 分析报告由你在阶段 3 亲自撰写。
你需要编写的部分（算子特定脚本）
参考：完整示例见 @references/batch-robustness-test-example.py。
import torch
from robustness_test_runner import RobustnessTestRunner
CONFIG = {
"verify_dir": "<case 目录路径>",
"task_module": "<task_code 模块名>",
"kernel_module": "<kernel_code 模块名>",
"op_name": "...",
"framework": "torch",
"dsl": "...",
"backend": "...",
"arch": "...",
"device_ids": [0],
"seed": 42,
"warmup_times": 5,
"run_times": 50,
"verify_timeout": 300,
}
TEST_CASES = [
("original", {"M": 1024, "K": 4096}, torch.float32, "原始 shape"),
("small", {"M": 4, "K": }, torch.float32, ),
...
]
():
M, K = params[], params[]
[torch.randn(M, K, dtype=dtype, device=device)]
__name__ == :
runner = RobustnessTestRunner(CONFIG, TEST_CASES, make_inputs)
runner.run()
参数绑定维度：对于 get_init_inputs() 依赖于 shape 参数的算子（如 Linear 的 in_features），
提供 get_init_params_fn 回调：
def get_init_params(params):
"""返回 None 使用默认的 get_init_inputs()，返回 list 覆盖"""
if "in_features" in params:
return [params["in_features"], params["out_features"]]
return None
runner = RobustnessTestRunner(CONFIG, TEST_CASES, make_inputs, get_init_params)
通用运行器内置的关键逻辑
以下逻辑在 robustness_test_runner.py 中已实现：
精度比对：
- 检查 shape、NaN 位置、Inf 位置和符号一致性
- 对有限值计算相对误差：
err_cnt > int(numel * limit) → FAIL
- dtype 容忍度：float16=0.004, bfloat16=0.03, 其他=0.02
性能测试：
| backend | DSL | 方法 | 参考源码 |
|---|
| ascend | triton_ascend | profiler_npu(fn, warmup, active, clear_l2_cache=True, dsl="triton_ascend") | akg_agents.op.verifier.profiler |
| ascend | 其他 | profiler_npu(fn, warmup, active, clear_l2_cache=True, dsl="other") | 同上 |
| cuda | triton_* | triton.testing.do_bench(fn, warmup, rep, return_mode="median") | DSLAdapterTritonCuda |
| cuda | 非 triton | time.perf_counter + torch.cuda.synchronize() | KernelVerifier._generate_base_benchmark_code |
| cpu | * | time.perf_counter 循环计时 | DSLAdapterCpp |
精度验证和性能测试均以子进程执行：
- 精度验证：子进程 +
asyncio.wait_for 超时保护 + process.kill() 终止编译子进程。不使用信号量（signal.alarm）——信号量无法杀死编译子进程。
- 性能测试：每个 case 的 benchmark 在独立子进程中运行。原因：
torch_npu.profiler.profile() 会修改进程级全局状态，同一进程内多次调用 profiler_npu() 会导致 NPU "vector core exception"。
DevicePool 并行：使用 akg_agents.core.async_pool.device_pool.DevicePool 分配设备，多设备时并行执行，单设备时退化为顺序执行。
运行
cd test_{op_name}_{timestamp}_{rid}
python robustness_test_{op_name}.py
AKG_AGENTS_DEVICES_LIST="0,1,2,3" python robustness_test_{op_name}.py
阶段 3：分析结果与撰写报告
脚本运行完成后产出 {op_name}_robustness_summary.json。
你需要亲自完成以下工作（不是由脚本完成）：
- 读取 JSON 数据
- 结合阶段 0 收集的算子信息（tensor 签名、kernel 分块参数等），对测试结果进行人工分析
- 撰写
{op_name}_robustness_report.md（格式见下文报告规范）
这一步是你作为测试工程师的核心价值——根据具体的失败数据和算子特征做出有深度的归因判断，而不是机械地罗列数据。
分析时关注的典型规律：
- 所有非 2 的幂都失败 → kernel 对齐处理有 bug
- dim < BLOCK_SIZE 时崩溃 → 缺少小 shape 保护
- 特定 dtype 全部失败 → 类型转换路径有问题
- 大 shape 精度下降 → 浮点累积误差
- 特定 shape 性能显著退化 → 分块策略不适配
- 大 shape 编译超时 → kernel 编译复杂度随 shape 膨胀，需检查编译策略
- 多个 case 超时 → 编译缓存未命中或编译器本身存在性能问题
报告规范
JSON 汇总 ({op_name}_robustness_summary.json)
包含全部原始数据，供程序消费和存档。结构：
{
"metadata": {
"op_name": "ReLU",
"test_date": "2026-02-26",
"framework": "pytorch",
"dsl": "triton_ascend",
"backend": "npu",
"arch": "Ascend910B",
"original_shape": "(16, 16384)",
"original_dtype": "torch.float32"
},
"summary": {
"total_cases": 12,
"accuracy_pass": 10,
"accuracy_fail": 1,
"accuracy_timeout": 1,
"perf_pass":
Markdown 分析报告 ({op_name}_robustness_report.md)
基于 JSON 数据的洞察分析，不重复逐 case 列表。
# 算子鲁棒性分析：{op_name}
**配置**: {framework} / {dsl} / {backend} / {arch}
**基准**: shape={original_shape}, dtype={original_dtype}
**原始数据**: `{op_name}_robustness_summary.json`
## 结论
{一句话定性判断，概括精度稳定性和性能主要风险。示例：}
精度在所有变异下均稳定；性能在半精度 dtype 和非对齐 shape 下存在显著退化，
根因指向 kernel 缺少 fp16/bf16 原生向量化路径和非对齐尾循环效率低。
## 失败模式
{将失败/警告 case 按共同根因聚类。}
### 模式 1: {名称，如"半精度 dtype 性能退化"}
- **涉及 case**: dtype_fp16 (0.18x), dtype_bf16 (0.17x)
- **归因**: kernel 内部将 fp16/bf16 转为 fp32 逐元素计算，丧失 SIMD 宽度优势。
- **风险**: 高 — 生产环境多使用 bf16 推理，直接影响端到端吞吐。
### 模式 2: ...
## 性能剖面
{描述 speedup 随规模/dtype 变化的趋势和拐点。}
| 规模区间 | 代表 shape | speedup 范围 | 趋势 |
|----------|-----------|-------------|------|
| ≤ 1e3 元素 | (4,64) | 0.89 ~ 1.65 | 调用开销主导，波动大 |
| 1e4 ~ 1e7 | (64,4096) | 0.53 ~ 0.94 | kernel 未体现优势 |
| ≥ 1e7 | (256,65536) | 1.10 | 大规模下优势显现 |
## 修复建议
{按优先级排列，每条附具体方向和预期收益。}
1. **[高] 添加 fp16/bf16 原生 kernel 分支** — 当前 0.17x → 目标 ≥ 1.0x
2. **[中] 优化中等规模 (1e4~1e7) 的分块策略** — 当前 0.53x，应至少持平
3. ...
## 已知限制（可接受）
- 极小 shape (1,1): kernel 调用开销 > 计算本身，speedup 波动属正常
---
反模式
| 行为 | 说明 |
|---|
| 同一 size class 内重复测试 | 每级 1 个代表值即可 |
| batch 维大量枚举 | 3 个代表值（1, 2, 64）覆盖关键场景 |
| 只测 2 的幂 shape | 遗漏对齐边界 bug，需包含 BLOCK±1 |
| 在同一模型实例上变异参数绑定维度 | 会 shape mismatch，需重新创建模型 |
| 不固定随机种子 | 测试不可复现 |
| 把 kernel 逻辑复制为参考实现 | Model 必须用 PyTorch 原生实现 |
| 精度容忍度不区分 dtype | fp16/bf16/fp32 的合理误差差距很大 |
| 性能测试不做 warmup | 首次运行含编译/初始化开销，会偏高 |
| 改进建议留空 | 必须根据失败/警告类别生成具体建议，禁止只写标题无内容 |
| 文件夹命名不规范 | 必须使用 test_{op_name}_{timestamp}_{random_id}/ 格式，确保唯一性和可追溯性 |
| 在脚本中生成 Markdown 报告 | Markdown 分析报告由你亲自撰写，不得在 Python 脚本中用字符串拼接生成 |
| 用信号量实现超时 | signal.alarm / 线程超时无法杀死编译子进程，必须用子进程 + process.kill() |
| 精度超时后仍执行性能测试 | 编译未完成意味着模型无法运行，性能测试无意义 |
