op-test

算子鲁棒性测试

---

流程概览

1. 信息收集 — 读取 task_code / kernel_code，提取 tensor 签名、维度语义、kernel 分块参数
2. 准备环境与测试目录 — 根据用户提供的 akg_agents 路径和环境名激活环境，创建独立测试目录
3. 编写测试脚本并运行 — 复制通用运行器 (robustness_test_runner.py)，编写算子特定脚本（CONFIG + TEST_CASES + make_inputs），导入 Model / ModelNew，构造变异 shape/dtype 输入，精度验证（子进程 + 超时保护）+ 性能测试（方法与 akg_agents 仓库一致），DevicePool 支持多设备并行
4. 分析结果与生成报告 — 输出 JSON 原始数据 + Markdown 洞察分析（失败模式聚类、性能剖面、修复建议）

---

Shape 变异策略

总元素数分级

根据单个 tensor flatten 后的元素数分为三级：

同一级别内 1 个代表值即可。

选值原则

1. 每级 1 个代表：小、中、大各一个 shape
2. 对齐边界：根据 kernel 的 BLOCK_SIZE / VECTOR_SIZE 取 BLOCK-1 和 BLOCK+1
3. 最小边界：所有维度为 1
4. 极端纵横比：某个维度为 1，其他维度大（如单 batch）
5. 非 2 的幂：至少 1 个含素数或非对齐维度的 shape

典型 case 组成（8-12 个）

维度约束

• 自由维度（如 batch_size）：直接变异输入 shape
• 参数绑定维度（如 Linear 的 in_features）：需重新创建模型实例

---

dtype 变异策略

精度容忍度：

---

执行流程

阶段 0：信息收集

读取 task_code 和 kernel_code，提取：

1. tensor 签名：从 get_inputs() 和 Model.forward() 确定每个输入的 shape 和 dtype
2. 维度语义：标注每个维度（batch、reduction_axis、spatial 等）
3. 参数约束：从 get_init_inputs() 确定哪些维度受模型参数绑定
4. kernel 分块参数：查看 kernel 中的 BLOCK_SIZE、VECTOR_SIZE 等常量
5. 当前验证通过的配置：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 出错概率：

通用运行器见 @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

# 1. 配置
CONFIG = {
    "verify_dir": "<case 目录路径>",       # task_code 和 kernel_code 所在目录
    "task_module": "<task_code 模块名>",    # 包含 Model 和 get_inputs
    "kernel_module": "<kernel_code 模块名>", # 包含 ModelNew
    "op_name": "...",
    "framework": "torch",
    "dsl": "...",           # "triton_ascend" / "triton_cuda" / "cpp" / ...
    "backend": "...",       # "ascend" / "cuda" / "cpu"
    "arch": "...",          # "Ascend910B" / "sm_80" / "x86_64"
    "device_ids": [0],      # 多设备: [0, 1, 2, 3]
    "seed": 42,
    "warmup_times": 5,
    "run_times": 50,
    "verify_timeout": 300,
}

# 2. 测试 case（算子特定的 shape/dtype 变异）
TEST_CASES = [
    # (tag, params_dict, dtype, description)
    ("original", {"M": 1024, "K": 4096}, torch.float32, "原始 shape"),
    ("small",    {"M": 4,    "K": 16},   torch.float32, "小 shape"),
    ...
]

# 3. 输入构造函数（按算子 tensor 签名编写）
def make_inputs(params, dtype, device):
    M, K = params["M"], params["K"]
    return [torch.randn(M, K, dtype=dtype, device=device)]

# 4. 入口（固定写法，无需修改）
if __name__ == "__main__":
    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

性能测试：

精度验证和性能测试均以子进程执行：

• 精度验证：子进程 + 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。

你需要亲自完成以下工作（不是由脚本完成）：

1. 读取 JSON 数据
2. 结合阶段 0 收集的算子信息（tensor 签名、kernel 分块参数等），对测试结果进行人工分析
3. 撰写 {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": 5,
    "perf_warn": 3,
    "perf_fail": 2,
    "perf_skip": 2
  },
  "cases": [
    {
      "tag": "original",
      "shape": "(16, 16384)",
      "dtype": "torch.float32",
      "description": "原始通过 shape (中等，~262K 元素)",
      "device_id": 0,
      "accuracy": "PASS",
      "base_time_us": 85.07,
      "gen_time_us": 108.42,
      "speedup": 0.785,
      "perf_status": "FAIL",
      "error": null
    },
    {
      "tag": "large",
      "shape": "(16384, 16384)",
      "dtype": "torch.float32",
      "description": "大 shape (编译超时)",
      "device_id": 0,
      "accuracy": "TIMEOUT",
      "base_time_us": null,
      "gen_time_us": null,
      "speedup": null,
      "perf_status": "SKIP",
      "error": "精度验证超时 (300s)，已终止子进程"
    }
  ]
}

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 波动属正常

反模式