diff --git a/skills/triton/kernel-triton-verifier/SKILL.md b/skills/triton/kernel-triton-verifier/SKILL.md
new file mode 100644
index 00000000..9a2729d0
--- /dev/null
+++ b/skills/triton/kernel-triton-verifier/SKILL.md
@@ -0,0 +1,313 @@
+---
+name: kernel-triton-verifier
+description: >
+ 算子代码验证 Skill — 按照标准验证流程验证生成的内核代码。
+ 创建验证项目文件，调用 scripts/verify.py 运行验证，验证通过后
+ 调用 scripts/benchmark.py 进行性能测试并收集结果。
+argument-hint: >
+ 输入：
+ - op-name: 算子名称（必须）
+ - generated-code-path: 生成的 Triton 代码文件路径（必须）
+ - pt-file-path: 包含输入数据和预期输出的 .pt 文件路径（必须）
+ - csv-file-path: 性能参考数据 vllm_gpu_perf.csv 文件路径（必须）
+ - device-id: 指定运行的 NPU 设备 ID（可选，默认 0）
+ - warmup: 性能测试 warmup 次数（可选，默认 5）
+ - repeats: 性能测试重复次数（可选，默认 50）
+ 输出：验证结果（成功/失败）、错误信息、性能数据。
+ 固定参数：framework=torch、backend=ascend、dsl=triton_ascend。
+---
+
+# Kernel Verifier Skill
+
+
+你是一个内核代码验证专家。你的任务是按照标准验证流程，创建验证项目并运行，检查生成的算子代码是否能正确编译运行且与参考实现的输出一致。验证通过后，执行性能测试并收集性能数据。
+
+
+## 验证流程
+
+```
+输入：generated_code.py + task_file.py
+ ↓
+[0. Triton 退化预检查] → scripts/validate_triton_impl.py (AST 静态分析)
+ ↓ (通过)
+[1. 创建验证项目] → 三个文件
+ ↓
+[2. 执行验证脚本] → scripts/verify.py --op_name ...
+ ↓
+[3. 收集验证结果]
+ ↓
+[验证通过] → [4. 执行性能测试] → scripts/benchmark.py --op_name ...
+ ↓
+[5. 收集性能结果]
+ ↓
+输出：验证结果 + 性能数据
+```
+
+---
+
+## Step 0: Triton 退化预检查（AST 静态分析）
+
+在创建验证项目之前，先使用 `validate_triton_impl.py` 对生成代码进行退化检测。此检查为纯 AST 静态分析，无需 NPU/torch 运行时，毫秒级完成。
+
+**命令模板**：
+
+```bash
+python3 <本skill所在目录的绝对路径>/scripts/validate_triton_impl.py \
+ <生成代码文件路径> --json
+```
+
+**检测三种退化类型**：
+
+| 类型 | 含义 | 检测方式 |
+|------|------|---------|
+| Type 1 | 完全无 `@triton.jit` kernel | AST 中无 `triton.jit` 装饰的函数定义 |
+| Type 2 | 部分计算使用 PyTorch | 代码中存在禁止的 `torch.*` / `F.*` 计算操作（精确到行号） |
+
+**结果判断**：
+- exit code == 0 → 通过，继续 Step 1
+- exit code != 0 → 退化检测到，解析 JSON 中的 `regression_type` 和 `suggestion`，直接返回失败
+
+**JSON 输出格式**：
+
+```json
+{
+ "valid": false,
+ "regression_type": 3,
+ "checks": {
+ "triton_kernel_exists": {"passed": true, "kernels": [...]},
+ "kernel_called_from_forward": {"passed": true, "called": [...]},
+ "no_forbidden_torch_ops": {"passed": false, "violations": [{"line": 45, "call": "F.softmax", "reason": "..."}]}
+ },
+ "suggestion": "..."
+}
+```
+
+---
+
+## Step 1: 创建验证项目
+
+在当前迭代的验证目录（如 `{output-path}/iter_{iteration}/verify/`）下创建三个必需文件：
+
+### 文件 1: `{op_name}.py`
+
+从入参 `generated-code-path` 指定的路径复制生成代码的完整内容到验证目录。
+
+### 文件 2: `{op_name}.pt`
+
+从入参 `pt-file-path` 指定的路径复制 `.pt` 文件到验证目录。
+
+**.pt 文件格式要求**：
+
+`.pt` 文件必须是一个通过 `torch.save()` 保存的字典，包含以下三个必需字段：
+
+```python
+{
+ "input_data": dict, # kernel 的输入参数字典
+ "grid": tuple/list, # kernel 的网格配置 (grid_x, grid_y, grid_z)
+ "gpu_output": dict # 预期的输出字典，键与 input_data 中的输出参数对应
+}
+```
+
+**加载说明**：
+
+在验证脚本中，`.pt` 文件应使用以下方式加载：
+
+```python
+torch.load(f"{verify_dir}/{op_name}.pt", map_location=torch.device('cpu'), weights_only=False)
+```
+
+
+### 文件 3: `vllm_gpu_perf.csv`
+
+从入参 `csv-file-path` 指定的路径复制 `vllm_gpu_perf.csv` 文件到验证目录。
+
+
+
+---
+
+## Step 2: 执行验证（⚠️ 必须使用本脚本，禁止自创测试方法）
+
+**必须使用** `bash` 工具调用本 skill 自带的 `scripts/verify.py` 脚本。
+
+**命令模板**：
+
+```bash
+python3 <本skill所在目录的绝对路径>/scripts/verify.py \
+ --op_name <算子名> \
+ --verify_dir <验证目录> \
+ --timeout 900 \
+ --device_id <设备ID>
+```
+
+**实际调用示例**（假设验证目录为 `/tmp/workspace/softmax/verify`，算子名为 `softmax`）：
+
+```bash
+python3 /path/to/kernel-verifier/scripts/verify.py \
+ --op_name softmax \
+ --verify_dir /tmp/workspace/softmax/verify \
+ --timeout 900 \
+ --device_id 0
+```
+
+**参数说明**：
+
+| 参数 | 必填 | 说明 |
+|------|------|------|
+| `--op_name` | 是 | 算子名称，与文件名前缀对应 |
+| `--verify_dir` | 否 | 验证目录路径，默认当前目录 |
+| `--timeout` | 否 | 超时秒数，默认 900 |
+| `--device_id` | 否 | 指定 NPU 设备 ID，默认 0 |
+
+**超时设置**：默认 900 秒，复杂算子可适当增加。
+
+**⛔ 禁止事项**：
+- 禁止自己编写 Python 代码来测试算子（如手动 import 并 forward 比较）
+- 禁止使用 `torch.allclose` 或其他自创方法替代 `scripts/verify.py`
+- 禁止跳过此步骤直接报告验证结果
+
+---
+
+## Step 3: 收集验证结果
+
+根据脚本的退出码和输出判断验证结果：
+
+### 验证通过
+
+脚本 stdout 输出 `"验证成功"` 且退出码为 0。
+
+返回：
+- `verifier_result = true`
+- `verifier_error = ""`
+
+### 验证失败
+
+脚本 stderr 包含错误信息且退出码非 0。
+
+返回：
+- `verifier_result = false`
+- `verifier_error` = stderr 中的完整错误输出（包括 AssertionError 信息和 traceback）
+
+### 超时
+
+脚本输出 `"验证超时"` 且退出码为 1。
+
+返回：
+- `verifier_result = false`
+- `verifier_error = "验证超时（300秒）"`
+
+---
+
+## Step 4: 执行性能测试（验证通过后执行）
+
+**仅在验证通过后执行**，使用 `bash` 工具调用本 skill 自带的 `scripts/benchmark.py` 脚本。
+
+**命令模板**：
+
+```bash
+python3 <本skill所在目录的绝对路径>/scripts/benchmark.py \
+ --op_name <算子名> \
+ --verify_dir <验证目录> \
+ --warmup \
+ --repeats <测试次数> \
+ --output <输出文件路径> \
+ --device_id <设备ID>
+```
+
+**实际调用示例**：
+
+```bash
+python3 /path/to/kernel-verifier/scripts/benchmark.py \
+ --op_name softmax \
+ --verify_dir /tmp/workspace/softmax/verify \
+ --warmup 5 \
+ --repeats 50 \
+ --output /tmp/workspace/softmax/iter_0/perf_result.json \
+ --device_id 0
+```
+
+> **注意**：`--output` 路径由调用方指定，性能报告将写入该路径。通常由 `kernelgen-workflow` SubAgent 指定为 `{output-path}/iter_{iteration}/perf_result.json`。
+> **目录提示**：`benchmark.py` 应在验证目录的上级目录执行，`--verify_dir` 参数使用相对路径 `verify` 或绝对路径。
+
+**参数说明**：
+
+| 参数 | 必填 | 说明 |
+|------|------|------|
+| `--op_name` | 是 | 算子名称 |
+| `--verify_dir` | 否 | 验证目录路径，默认当前目录 |
+| `--warmup` | 否 | warmup 次数，默认 5 |
+| `--repeats` | 否 | 正式测试次数，默认 50 |
+| `--output` | 否 | 性能报告输出路径（JSON 格式）|
+| `--device_id` | 否 | 指定 NPU 设备 ID，默认 0 |
+
+---
+
+## Step 5: 收集性能结果
+
+性能测试完成后，从 `--output` 指定的 JSON 文件中读取结果。
+
+### 性能报告格式
+
+```json
+{
+ "op_name": "softmax",
+ "warmup": 5,
+ "repeats": 50,
+ "triton_gpu": {
+ "avg_latency_ms": 1.2345,
+ "peak_memory_mb": 256.00
+ },
+ "triton_ascend": {
+ "avg_latency_ms": 0.5678,
+ "peak_memory_mb": 128.00
+ },
+ "speedup_vs_gpu": 2.17
+}
+```
+
+**指标说明**：
+
+| 指标 | 说明 |
+|------|------|
+| `avg_latency_ms` | 平均延迟（毫秒）|
+| `peak_memory_mb` | 峰值内存占用（MB）|
+| `speedup_vs_gpu` | 相比triton-gpu 实现的加速比 |
+
+**返回**：
+- `perf_result`：dict（完整性能数据）
+- `perf_report_path`：str（性能报告文件路径）
+
+---
+
+## 精度阈值说明
+
+验证使用基于数据类型的**相对误差**比较，与 `torch.allclose` 不同：
+
+| 数据类型 | 精度阈值 (limit) | 说明 |
+|---------|-----------------|------|
+| `float16` | 0.004 | 半精度浮点 |
+| `bfloat16` | 0.03 | BF16 精度较低 |
+| `int8` | 0.01 | 整数量化 |
+| 其他（float32 等） | 0.02 | 默认阈值 |
+
+**比较规则**：
+1. 形状必须一致
+2. NaN 位置必须一致
+3. Inf 位置和符号必须一致
+4. 有限值：计算相对误差，超过阈值的数量不得超过 `有限值总数 × limit`
+
+---
+
+## 脚本位置
+
+验证脚本位于本 skill 的 `scripts/` 目录：
+
+| 脚本 | 用途 |
+|------|------|
+| `scripts/validate_triton_impl.py` | 退化预检查（AST 静态分析） |
+| `scripts/verify.py` | 验证正确性 |
+| `scripts/benchmark.py` | 测试性能 |
+
+**CLI 参数**：
+- `validate_triton_impl.py`: ``, `[--json]`
+- `verify.py`: `--op_name`, `--verify_dir`, `--timeout`, `--device_id`
+- `benchmark.py`: `--op_name`, `--verify_dir`, `--warmup`, `--repeats`, `--output`, `--device_id`
\ No newline at end of file
diff --git a/skills/triton/kernel-triton-verifier/scripts/benchmark.py b/skills/triton/kernel-triton-verifier/scripts/benchmark.py
new file mode 100644
index 00000000..f1a139d8
--- /dev/null
+++ b/skills/triton/kernel-triton-verifier/scripts/benchmark.py
@@ -0,0 +1,467 @@
+#!/usr/bin/env python3
+"""算子验证脚本 — 算子对应.pt文件中包含输入以及预期的输出, 相同输入下, 对比生成算子输出与预期输出的一致性。
+
+用法:
+ python benchmark.py --op_name <算子名> [--verify_dir <验证目录>] [--output <输出路径>] [--device_id <所用设备id>]
+
+前置条件（验证目录下需存在以下文件）:
+ {op_name}.pt — 包含输入，预期输出
+ {op_name}.py — 包含生成算子的主要逻辑
+ vllm_gpu_perf.csv — 包含 Triton-GPU 实现的执行耗时（基准数据）
+"""
+
+import argparse
+import json
+import os
+import shutil
+import sys
+import time
+from typing import Dict, List, Optional, Tuple, Any
+from dataclasses import dataclass
+import importlib
+import gc
+import torch
+import pandas as pd
+
+from test_common import convert_tensor_with_device_type
+
+# ============================================================================
+# 配置常量
+# ============================================================================
+
+WARMUP_DEFAULT = 5
+REPEATS_DEFAULT = 50
+
+
+# ============================================================================
+# 数据类
+# ============================================================================
+
+@dataclass
+class BenchmarkConfig:
+ """性能测试配置"""
+ op_name: str
+ verify_dir: str
+ warmup: int = WARMUP_DEFAULT
+ repeats: int = REPEATS_DEFAULT
+
+
+@dataclass
+class PerformanceResult:
+ """单次性能测试结果"""
+ avg_latency_ms: float
+ peak_memory_mb: float
+ operators: Dict[str, float]
+
+
+@dataclass
+class BenchmarkResult:
+ """完整性能测试结果"""
+ op_name: str
+ warmup: int
+ repeats: int
+ triton_gpu: PerformanceResult
+ triton_ascend: PerformanceResult
+ speedup_vs_gpu: float
+
+
+# ============================================================================
+# 辅助函数
+# ============================================================================
+
+def load_models(op_name: str, verify_dir: str, device: Any):
+ """加载框架实现和Triton实现模型"""
+ import torch
+ import torch_npu
+
+ spec = importlib.util.spec_from_file_location(op_name, f"{verify_dir}/{op_name}.py")
+ triton_npu_module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(triton_npu_module)
+
+ # 获取 kernel 函数
+ triton_npu_func = getattr(triton_npu_module, op_name)
+
+ data = torch.load(f"{verify_dir}/{op_name}.pt", map_location=torch.device('cpu'), weights_only=False)
+
+ return triton_npu_func, data, device
+
+def prepare_triton_fn(triton_func: Any, grid: List[int], input_data: dict) -> callable:
+ # 执行warmup
+ with torch.no_grad():
+ _ = triton_func[grid](**input_data)
+ torch.npu.synchronize()
+
+ # 返回测试函数
+ def test_fn():
+ with torch.no_grad():
+ _ = triton_func[grid](**input_data)
+ torch.npu.synchronize()
+
+ return test_fn
+
+
+def find_profile_file(profile_path: str, filename: str) -> Optional[str]:
+ """在profile目录中查找指定文件"""
+ for root, _, files in os.walk(profile_path):
+ if filename in files:
+ return os.path.join(root, filename)
+ return None
+
+
+def cleanup_profile_path(profile_path: str) -> None:
+ """清理profile目录"""
+ if os.path.exists(profile_path):
+ shutil.rmtree(profile_path, ignore_errors=True)
+
+
+# ============================================================================
+# 性能分析逻辑
+# ============================================================================
+
+def parse_operator_latency(profile_path: str, active_count: int) -> Tuple[Optional[Dict[str, float]], Optional[float]]:
+ """从 profiling 结果文件中提取算子时延数据，计算平均执行时间。"""
+ import pandas as pd
+
+ operator_details_file = find_profile_file(profile_path, "operator_details.csv")
+
+ if not operator_details_file or not os.path.exists(operator_details_file):
+ cleanup_profile_path(profile_path)
+ return None, None
+
+ try:
+ df = pd.read_csv(operator_details_file)
+ except Exception:
+ cleanup_profile_path(profile_path)
+ return None, None
+
+ required_columns = ["Name", "Device Self Duration(us)"]
+ missing_columns = [col for col in required_columns if col not in df.columns]
+ if missing_columns:
+ cleanup_profile_path(profile_path)
+ return None, None
+
+ if "Count" not in df.columns:
+ return _parse_without_count(df, profile_path, active_count)
+
+ return _parse_with_count(df, profile_path, active_count)
+
+
+def _parse_without_count(df: Any, profile_path: str, active_count: int) -> Tuple[Optional[Dict[str, float]], Optional[float]]:
+ """处理没有 Count 列的情况：按操作名称直接累加计算。"""
+ # 按算子名称分组，累加所有测量周期的 Device Self Duration
+ operator_avg_times = {}
+ grouped = df.groupby("Name")["Device Self Duration(us)"].sum()
+ for op_name_str, total_us in grouped.items():
+ # 平均到每次运行（微秒）
+ operator_avg_times[op_name_str] = total_us / active_count
+
+ # 汇总所有算子的平均时间，得到完整的 device 侧执行时间
+ total_avg_us = sum(operator_avg_times.values())
+ total_avg_ms = total_avg_us / 1000.0
+
+ cleanup_profile_path(profile_path)
+
+ return operator_avg_times, round(total_avg_ms, 4)
+
+
+def _parse_with_count(df: Any, profile_path: str, active_count: int) -> Tuple[Optional[Dict[str, float]], Optional[float]]:
+ """解析有 Count 列的情况：按操作名称分组，累加 Self Duration，计算每次运行的平均时间。"""
+ # 筛选出 Count 等于 active_count 的记录（即正式测试阶段的算子）
+ valid_ops = df[df["Count"] == active_count].copy()
+
+ if valid_ops.empty:
+ cleanup_profile_path(profile_path)
+ return None, None
+
+ # 按算子名称分组，累加 Device Self Duration
+ operator_avg_times = {}
+ grouped = valid_ops.groupby("Name")
+ for op_name_str, group in grouped:
+ total_us = group["Device Self Duration(us)"].sum()
+ avg_us = total_us / active_count
+ # 存储单位为微秒（us）
+ operator_avg_times[op_name_str] = avg_us
+
+ # 汇总所有算子的 Self Duration，得到一次完整运行的 device 侧总时间
+ total_avg_us = sum(operator_avg_times.values())
+ # 转换为毫秒
+ total_avg_ms = total_avg_us / 1000.0
+
+ cleanup_profile_path(profile_path)
+
+ return operator_avg_times, round(total_avg_ms, 4)
+
+
+def run_profiler_with_config(test_fn: callable, warmup: int, repeats: int, profile_name: str) -> str:
+ """运行NPU profiler并返回生成的性能分析目录路径。"""
+ import torch
+ import torch_npu
+
+ # 实验性配置
+ experimental_config = torch_npu.profiler._ExperimentalConfig(
+ aic_metrics=torch_npu.profiler.AiCMetrics.PipeUtilization,
+ profiler_level=torch_npu.profiler.ProfilerLevel.Level1,
+ l2_cache=False
+ )
+
+ # 预热一次确保模型准备就绪
+ test_fn()
+ torch.npu.synchronize()
+
+ skip_first = 1 + warmup
+ total_steps = skip_first + repeats
+
+ # 生成唯一的profile路径
+ timestamp = int(time.time() * 1000)
+ profile_path = os.path.join(os.getcwd(), f"{profile_name}_{timestamp}")
+
+ # 创建profiler
+ with torch_npu.profiler.profile(
+ activities=[
+ torch_npu.profiler.ProfilerActivity.NPU,
+ torch_npu.profiler.ProfilerActivity.CPU
+ ],
+ schedule=torch_npu.profiler.schedule(
+ wait=0, warmup=warmup, active=repeats, repeat=1, skip_first=skip_first
+ ),
+ on_trace_ready=torch_npu.profiler.tensorboard_trace_handler(profile_path),
+ record_shapes=False,
+ profile_memory=False,
+ with_stack=False,
+ experimental_config=experimental_config,
+ ) as prof:
+ for _ in range(total_steps):
+ test_fn()
+ prof.step()
+ torch.npu.synchronize()
+
+ return profile_path
+
+
+def measure_single(
+ model: Any,
+ grid,
+ inputs: List[Any],
+ warmup: int,
+ repeats: int,
+ profile_name: str,
+ device: Any
+) -> Tuple[Optional[Dict[str, float]], Optional[float], float]:
+ """测量单次性能（warmup + profiling）"""
+ import torch
+ import torch_npu
+ print(f"measure_single", flush=True)
+ # 重置峰值内存统计
+ torch.npu.reset_peak_memory_stats()
+
+ # 准备测试函数
+ test_fn = prepare_triton_fn(model, grid, inputs)
+
+ try:
+ # 运行profiler
+ profile_path = run_profiler_with_config(test_fn, warmup, repeats, profile_name)
+
+ # 解析结果
+ operators, latency_ms = parse_operator_latency(profile_path, repeats)
+ except Exception as e:
+ print(f"torch_npu.profiler 获取数据失败: {e}，使用兜底测试机制...")
+ operators, latency_ms = None, None
+
+ # 如果profiler获取不到数据，使用兜底机制
+ if operators is None or latency_ms is None:
+ print(f"警告: profiler 无法获取时延数据，将使用 time.perf_counter() 进行兜底测试...")
+ return measure_single_fallback(model, grid, inputs, warmup, repeats, device)
+
+ # 获取峰值内存
+ peak_memory = torch.npu.max_memory_allocated() / (1024 * 1024)
+
+ return operators, latency_ms, round(peak_memory, 2)
+
+
+def measure_single_fallback(
+ model: Any,
+ grid,
+ inputs: List[Any],
+ warmup: int,
+ repeats: int,
+ device: Any
+) -> Tuple[Optional[Dict[str, float]], Optional[float], float]:
+ """使用time.perf_counter()的兜底测试机制"""
+ import torch
+ import torch_npu
+ import time
+ import statistics
+
+ # 执行warmup
+ with torch.no_grad():
+ for _ in range(warmup):
+ _ = model[grid](*inputs)
+ torch.npu.synchronize()
+
+ # 正式测试
+ latencies = []
+ for _ in range(repeats):
+ torch.npu.synchronize()
+ start = time.perf_counter()
+ with torch.no_grad():
+ _ = model[grid](*inputs)
+ torch.npu.synchronize()
+ end = time.perf_counter()
+ latencies.append((end - start) * 1000) # 转换为毫秒
+
+ # 计算平均时延
+ avg_latency_ms = statistics.mean(latencies)
+
+ # 获取峰值内存
+ peak_memory = torch.npu.max_memory_allocated() / (1024 * 1024)
+
+ # 兜底机制不获取算子级别的时延，返回空字典
+ return {}, round(avg_latency_ms, 4), round(peak_memory, 2)
+
+
+# ============================================================================
+# 主测试逻辑
+# ============================================================================
+
+def benchmark_implementations(config: BenchmarkConfig) -> BenchmarkResult:
+ """执行完整的性能测试"""
+ import torch
+ import torch_npu
+
+ device = torch.device("npu")
+
+ # 加载模型和输入
+ triton_npu_func, data, device = load_models(
+ config.op_name,
+ config.verify_dir,
+ device
+ )
+ print(f"load_models", flush=True)
+ # 将输入移到设备上
+ input_data = convert_tensor_with_device_type(data["input_data"], device_type='npu')
+ grid = data['grid']
+ # 测试框架实现
+ df = pd.read_csv(f"{config.verify_dir}/vllm_gpu_perf.csv")
+ framework_latency_us = df.loc[df["Name"] == config.op_name, "Duration(us)"].item()
+ framework_latency_ms = framework_latency_us / 1000.0
+ framework_operators, framework_peak_memory = {},1
+ # 测试生成实现
+ print(f"执行 Implementation warmup 和 profiler (warmup={config.warmup}, active={config.repeats})...")
+ impl_operators, impl_latency_ms, impl_peak_memory = measure_single(
+ triton_npu_func, grid, input_data, config.warmup, config.repeats, "impl_profile", device
+ )
+
+ # 验证结果
+ if framework_latency_ms is None or impl_latency_ms is None:
+ raise RuntimeError("无法从 profiler 结果中提取有效的时延数据")
+
+ # 计算加速比
+ speedup = (
+ framework_latency_ms / impl_latency_ms
+ if impl_latency_ms > 0 and framework_latency_ms > 0
+ else 0
+ )
+
+ # 构建结果
+ return BenchmarkResult(
+ op_name=config.op_name,
+ warmup=config.warmup,
+ repeats=config.repeats,
+ triton_gpu=PerformanceResult(
+ avg_latency_ms=round(framework_latency_ms, 4),
+ peak_memory_mb=round(framework_peak_memory, 2),
+ operators=framework_operators or {}
+ ),
+ triton_ascend=PerformanceResult(
+ avg_latency_ms=round(impl_latency_ms, 4),
+ peak_memory_mb=round(impl_peak_memory, 2),
+ operators=impl_operators or {}
+ ),
+ speedup_vs_gpu=round(speedup, 2)
+ )
+
+
+def result_to_dict(result: BenchmarkResult) -> Dict[str, Any]:
+ """将BenchmarkResult转换为字典格式"""
+ return {
+ "op_name": result.op_name,
+ "warmup": result.warmup,
+ "repeats": result.repeats,
+ "triton_gpu": {
+ "avg_latency_ms": result.triton_gpu.avg_latency_ms,
+ "peak_memory_mb": result.triton_gpu.peak_memory_mb,
+ "operators": {name: round(avg_us, 4) for name, avg_us in result.triton_gpu.operators.items()}
+ },
+ "triton_ascend": {
+ "avg_latency_ms": result.triton_ascend.avg_latency_ms,
+ "peak_memory_mb": result.triton_ascend.peak_memory_mb,
+ "operators": {name: round(avg_us, 4) for name, avg_us in result.triton_ascend.operators.items()}
+ },
+ "speedup_vs_gpu": result.speedup_vs_gpu
+ }
+
+
+# ============================================================================
+# 命令行入口
+# ============================================================================
+
+def main():
+ parser = argparse.ArgumentParser(description="性能测试脚本")
+ parser.add_argument("--op_name", required=True, help="算子名称")
+ parser.add_argument("--verify_dir", default=".", help="验证目录路径（默认当前目录）")
+ parser.add_argument("--warmup", type=int, default=WARMUP_DEFAULT, help="warmup 次数（默认 5）")
+ parser.add_argument("--repeats", type=int, default=REPEATS_DEFAULT, help="正式测试次数（默认 50）")
+ parser.add_argument("--output", help="输出文件路径（JSON 格式）")
+ parser.add_argument(
+ "--device_id", required=True, type=int, default=0, help="指定npu卡"
+ )
+
+ args = parser.parse_args()
+ torch.npu.set_device(args.device_id)
+
+ # 验证目录
+ verify_dir = os.path.abspath(args.verify_dir)
+ if not os.path.isdir(verify_dir):
+ print(f"错误: 验证目录不存在: {verify_dir}", file=sys.stderr)
+ sys.exit(1)
+
+ # 构建配置
+ config = BenchmarkConfig(
+ op_name=args.op_name,
+ verify_dir=verify_dir,
+ warmup=args.warmup,
+ repeats=args.repeats
+ )
+
+ try:
+ # 执行测试
+ result = benchmark_implementations(config)
+ result_dict = result_to_dict(result)
+
+ # 输出结果
+ print("\n性能测试结果:")
+ print(f" Triton-GPU 实现 - 平均延迟: {result_dict['triton_gpu']['avg_latency_ms']:.4f} ms")
+ print(f" Triton-Ascend 实现 - 平均延迟: {result_dict['triton_ascend']['avg_latency_ms']:.4f} ms")
+ print(f" 加速比 (Ascend vs GPU): {result_dict['speedup_vs_gpu']:.2f}x")
+ print(f" Triton-Ascend 实现 - 峰值内存: {result_dict['triton_ascend']['peak_memory_mb']:.2f} MB")
+
+ # 保存到文件或输出
+ if args.output:
+ with open(args.output, "w", encoding="utf-8") as f:
+ json.dump(result_dict, f, indent=2, ensure_ascii=False)
+ print(f"\n结果已保存到: {args.output}")
+ else:
+ print("\n结果:")
+ print(json.dumps(result_dict, indent=2, ensure_ascii=False))
+
+ sys.exit(0)
+
+ except Exception as e:
+ print(f"性能测试失败: {e}", file=sys.stderr)
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
diff --git a/skills/triton/kernel-triton-verifier/scripts/test_common.py b/skills/triton/kernel-triton-verifier/scripts/test_common.py
new file mode 100644
index 00000000..7b84251f
--- /dev/null
+++ b/skills/triton/kernel-triton-verifier/scripts/test_common.py
@@ -0,0 +1,69 @@
+import torch
+from typing import Optional
+
+def convert_tensor_with_device_type(indata: dict, device_type: str):
+ target_device = torch.device(device_type)
+ outdata = {}
+
+ for key, value in indata.items():
+ if isinstance(value, torch.Tensor):
+ if value.device.type != target_device.type:
+ outdata[key] = value.to(target_device)
+ else:
+ outdata[key] = value
+ else:
+ outdata[key] = value
+
+ return outdata
+
+
+def validate_cmp(dtype, y_cal, y_ref, overflow_mode: Optional[str] = None, device_type: Optional[str] = None):
+ if device_type is not None:
+ target_device = torch.device(device_type)
+ y_cal = y_cal.to(target_device)
+ y_ref = y_ref.to(target_device)
+ else:
+ y_cal = y_cal.npu()
+ y_ref = y_ref.npu()
+ if overflow_mode == "saturate":
+ if dtype in ['float32', 'float16']:
+ min_value = -torch.finfo(dtype).min
+ max_value = torch.finfo(dtype).max
+ elif dtype in ['int32', 'int16', 'int8']:
+ min_value = torch.iinfo(dtype).min
+ max_value = torch.iinfo(dtype).max
+ elif dtype == 'bool':
+ min_value = 0
+ max_value = 1
+ else:
+ raise ValueError('Invalid parameter "dtype" is found : {}'.format(dtype))
+ y_ref = torch.clamp(y_ref, min=min_value, max=max_value)
+ if dtype == 'float16':
+ torch.testing.assert_close(y_ref, y_cal, rtol=1e-03, atol=1e-03, equal_nan=True)
+ elif dtype == 'bfloat16':
+ torch.testing.assert_close(y_ref.to(torch.float32), y_cal.to(torch.float32), rtol=5e-03, atol=5e-03,
+ equal_nan=True)
+ elif dtype == 'float32':
+ torch.testing.assert_close(y_ref, y_cal, rtol=1e-05, atol=1e-05, equal_nan=True)
+ elif dtype == 'int32' or dtype == 'int64' or dtype == 'int16' or dtype == 'int8':
+ assert torch.equal(y_cal, y_ref)
+ elif dtype == 'bool':
+ assert torch.equal(y_cal, y_ref)
+ else:
+ raise ValueError('Invalid parameter \"dtype\" is found : {}'.format(dtype))
+
+
+def compare_data_precision(dict_ref: dict, dict_cal: dict, device_type: str):
+ keys_ref, keys_cal = set(dict_ref.keys()), set(dict_cal.keys())
+ if not keys_ref.issubset(keys_cal):
+ raise ValueError("The keys of dict_ref is not subset of dict_cal")
+
+ for key in dict_ref.keys():
+ val_a, val_b = dict_ref[key], dict_cal[key]
+ if type(val_a) != type(val_b):
+ raise ValueError("The data type of two dicts are different")
+
+ if isinstance(val_a, torch.Tensor):
+ validate_cmp(dtype=str(val_a.dtype).split('.')[-1], y_ref=val_a, y_cal=val_b, device_type=device_type)
+ else:
+ raise ValueError("Non-tensor type is not currently supported")
\ No newline at end of file
diff --git a/skills/triton/kernel-triton-verifier/scripts/validate_triton_impl.py b/skills/triton/kernel-triton-verifier/scripts/validate_triton_impl.py
new file mode 100644
index 00000000..44c41519
--- /dev/null
+++ b/skills/triton/kernel-triton-verifier/scripts/validate_triton_impl.py
@@ -0,0 +1,389 @@
+#!/usr/bin/env python3
+"""Triton 实现退化检测脚本 — 通过 AST 静态分析检查生成代码是否退化为 PyTorch 原生实现。
+
+检测两种退化类型：
+ Type 1: 无 @triton.jit kernel，全部使用 PyTorch
+ Type 2: 代码中存在禁止的 PyTorch 计算操作
+
+用法:
+ python validate_triton_impl.py [--json]
+
+退出码: 0 = 通过, 1 = 检测到退化
+"""
+import ast
+import argparse
+import json
+import sys
+
+
+# ---------------------------------------------------------------------------
+# 白名单：允许的 torch 调用和 tensor 方法
+# ---------------------------------------------------------------------------
+
+ALLOWED_TORCH_FUNCS = {
+ # buffer 分配
+ "empty", "empty_like", "empty_strided",
+ "zeros", "zeros_like",
+ "ones", "ones_like",
+ "full", "full_like",
+ # tensor 创建（有时需要用于标量常量 / 索引）
+ "tensor", "arange", "linspace",
+ # 类型 / 设备
+ "as_tensor",
+}
+
+ALLOWED_TENSOR_METHODS = {
+ # 形状 / 元信息
+ "size", "shape", "stride", "numel", "dtype", "device", "dim",
+ "is_contiguous", "data_ptr", "element_size", "storage_offset",
+ # 布局操作（不执行计算）
+ "contiguous", "to", "view", "view_as", "reshape",
+ "permute", "transpose", "expand", "expand_as",
+ "flatten", "unflatten", "unsqueeze", "squeeze",
+ "narrow", "clone", "detach", "t",
+ "type", "float", "half", "bfloat16", "int", "long", "bool", "double",
+ "cpu", "npu", "cuda",
+ "item", "tolist",
+ # 原地标记
+ "requires_grad_", "zero_",
+ # 切片相关（一般通过 __getitem__ 而非方法，但以防万一）
+ "index_select",
+}
+
+ALLOWED_TRITON_ATTRS = {
+ "cdiv", "next_power_of_2",
+}
+
+FORBIDDEN_TENSOR_METHODS = {
+ # 计算操作
+ "sum", "mean", "max", "min", "softmax", "log_softmax",
+ "matmul", "mm", "bmm", "addmm", "add", "sub", "mul", "div",
+ "relu", "sigmoid", "tanh", "gelu", "silu", "elu", "leaky_relu",
+ "exp", "log", "log2", "log10", "sqrt", "pow", "abs",
+ "norm", "layer_norm", "batch_norm", "group_norm",
+ "conv1d", "conv2d", "conv3d", "conv_transpose2d", "linear",
+ "dropout", "softplus", "hardtanh", "hardswish",
+}
+
+
+# ---------------------------------------------------------------------------
+# AST 辅助函数
+# ---------------------------------------------------------------------------
+
+def _decorator_is_triton_jit(decorator):
+ """判断装饰器节点是否为 triton.jit 或 @jit（从 triton 导入）。"""
+ # @triton.jit
+ if isinstance(decorator, ast.Attribute):
+ if (isinstance(decorator.value, ast.Name)
+ and decorator.value.id == "triton"
+ and decorator.attr == "jit"):
+ return True
+ # @jit（直接导入）
+ if isinstance(decorator, ast.Name) and decorator.id == "jit":
+ return True
+ # @triton.jit 作为 Call（如 @triton.jit 带参数，虽然少见）
+ if isinstance(decorator, ast.Call):
+ return _decorator_is_triton_jit(decorator.func)
+ return False
+
+
+def _decorator_is_triton_autotune(decorator):
+ """判断装饰器是否为 triton.autotune。"""
+ if isinstance(decorator, ast.Attribute):
+ if (isinstance(decorator.value, ast.Name)
+ and decorator.value.id == "triton"
+ and decorator.attr == "autotune"):
+ return True
+ if isinstance(decorator, ast.Call):
+ return _decorator_is_triton_autotune(decorator.func)
+ return False
+
+
+def _has_triton_decorator(func_node):
+ """检查函数是否有 @triton.jit（可能与 @triton.autotune 组合）。"""
+ for dec in func_node.decorator_list:
+ if _decorator_is_triton_jit(dec):
+ return True
+ return False
+
+
+def _resolve_call_name(node):
+ """尝试从 ast.Call 节点提取被调用函数的名称字符串。
+
+ 返回 (qualifier, attr) 或 (None, name) 或 None。
+ 例如：torch.empty -> ('torch', 'empty')
+ my_func -> (None, 'my_func')
+ self.conv -> ('self', 'conv')
+ kernel[g] -> 返回 None（kernel launch 通过 Subscript）
+ """
+ func = node.func if isinstance(node, ast.Call) else node
+ if isinstance(func, ast.Attribute):
+ if isinstance(func.value, ast.Name):
+ return (func.value.id, func.attr)
+ # 处理 torch.nn.functional.relu 形式
+ if isinstance(func.value, ast.Attribute):
+ inner = func.value
+ if isinstance(inner.value, ast.Name):
+ return (f"{inner.value.id}.{inner.attr}", func.attr)
+ if isinstance(func, ast.Name):
+ return (None, func.id)
+ return None
+
+
+# ---------------------------------------------------------------------------
+# 核心检查
+# ---------------------------------------------------------------------------
+
+def find_triton_kernels(tree):
+ """查找所有 @triton.jit 装饰的函数名，及其是否使用了 tl.* API。"""
+ kernels = {} # name -> {"has_tl_usage": bool, "line": int, end_line: int}
+ kernel_ranges = [] # 存储 (start_line, end_line)
+
+ for node in ast.walk(tree):
+ if isinstance(node, ast.FunctionDef) and _has_triton_decorator(node):
+ # 检查函数体中是否使用 tl.* API
+ has_tl = False
+ for child in ast.walk(node):
+ if isinstance(child, ast.Attribute):
+ if isinstance(child.value, ast.Name) and child.value.id == "tl":
+ has_tl = True
+ break
+
+ # 获取函数起止行
+ start_line = node.lineno
+ end_line = node.end_lineno if hasattr(node, 'end_lineno') else start_line
+ kernels[node.name] = {
+ "has_tl_usage": has_tl,
+ "line": node.lineno,
+ "end_line": end_line
+ }
+ kernel_ranges.append((start_line, end_line))
+
+ return kernels, kernel_ranges
+
+
+def check_forbidden_torch_ops(tree, kernel_ranges):
+ """检查整个代码中是否使用了禁止的 torch 计算操作 —— 跳过 Triton 内核内部"""
+ violations = []
+
+ def is_inside_kernel(line):
+ # 判断当前行是否在任意一个 @triton.jit 函数内部
+ for (start, end) in kernel_ranges:
+ if start <= line <= end:
+ return True
+ return False
+
+ for node in ast.walk(tree):
+ if not hasattr(node, 'lineno'):
+ continue
+ # 跳过在 kernel 内部的代码
+ line = node.lineno
+ if is_inside_kernel(line):
+ continue
+
+ # --- 检测 @ 运算符（矩阵乘法）---
+ if isinstance(node, ast.BinOp) and isinstance(node.op, ast.MatMult):
+ violations.append({
+ "line": node.lineno,
+ "call": "@",
+ "reason": "矩阵乘法 @ 运算符必须在 Triton kernel 中实现",
+ })
+ continue
+
+ if not isinstance(node, ast.Call):
+ continue
+
+ # --- kernel launch 跳过检查 ---
+ if isinstance(node.func, ast.Subscript):
+ continue
+
+ resolved = _resolve_call_name(node)
+ if resolved is None:
+ continue
+
+ qual, attr = resolved
+
+ # --- torch.xxx(...) ---
+ if qual == "torch":
+ if attr not in ALLOWED_TORCH_FUNCS:
+ violations.append({
+ "line": node.lineno,
+ "call": f"torch.{attr}",
+ "reason": f"torch.{attr} 是计算操作，必须在 Triton kernel 中实现",
+ })
+ continue
+
+ # --- F.xxx(...) / functional.xxx(...) ---
+ if qual in ("F", "functional", "torch.nn.functional", "nn.functional"):
+ violations.append({
+ "line": node.lineno,
+ "call": f"{qual}.{attr}",
+ "reason": f"{qual}.{attr} 是 PyTorch 计算操作，必须在 Triton kernel 中实现",
+ })
+ continue
+
+ # --- triton.cdiv 等 —— 允许 ---
+ if qual == "triton" and attr in ALLOWED_TRITON_ATTRS:
+ continue
+
+ # --- tensor 方法计算操作 ---
+ if attr in FORBIDDEN_TENSOR_METHODS:
+ # 排除已知安全的 qual（torch/F/triton 已在上面处理）
+ if qual not in ("torch", "F", "triton", "functional", "torch.nn.functional", "nn.functional"):
+ violations.append({
+ "line": node.lineno,
+ "call": f"{qual}.{attr}()" if qual else f"{attr}()",
+ "reason": f"{attr} 是计算操作，必须在 Triton kernel 中实现",
+ })
+ continue
+
+ return violations
+
+
+# ---------------------------------------------------------------------------
+# 主验证逻辑
+# ---------------------------------------------------------------------------
+
+def validate(code, filepath=""):
+ """对生成代码执行完整的退化检查。
+
+ 返回结构化结果 dict。
+ """
+ result = {
+ "valid": False,
+ "filepath": filepath,
+ "checks": {
+ "triton_kernel_exists": {"passed": False, "kernels": [], "error": None},
+ "no_forbidden_torch_ops": {"passed": False, "violations": [], "error": None},
+ },
+ "regression_type": None,
+ "suggestion": "",
+ }
+
+ # --- 解析 ---
+ try:
+ tree = ast.parse(code)
+ except SyntaxError as e:
+ result["checks"]["triton_kernel_exists"]["error"] = f"SyntaxError: {e}"
+ result["regression_type"] = 1
+ result["suggestion"] = "代码存在语法错误，无法解析。"
+ return result
+
+ # --- Check 1: kernel 存在性 ---
+ kernels, kernel_ranges = find_triton_kernels(tree)
+ kernel_names = set(kernels.keys())
+ result["checks"]["triton_kernel_exists"]["kernels"] = [
+ {"name": k, "line": v["line"], "has_tl_usage": v["has_tl_usage"]}
+ for k, v in kernels.items()
+ ]
+
+ if not kernel_names:
+ result["checks"]["triton_kernel_exists"]["error"] = "未找到任何 @triton.jit 装饰的 kernel 函数"
+ result["regression_type"] = 1
+ result["suggestion"] = (
+ "代码中没有 Triton kernel。必须创建至少一个 @triton.jit 装饰的函数，"
+ "在其中使用 tl.load/tl.store 实现核心计算逻辑。"
+ )
+ return result
+
+ # 检查 kernel 是否使用了 tl API
+ kernels_without_tl = [k for k, v in kernels.items() if not v["has_tl_usage"]]
+ if len(kernels_without_tl) == len(kernels):
+ result["checks"]["triton_kernel_exists"]["error"] = (
+ f"kernel 函数 {kernels_without_tl} 未使用任何 tl.* API，"
+ "可能是空壳 kernel"
+ )
+ result["regression_type"] = 1
+ result["suggestion"] = (
+ "虽然存在 @triton.jit 装饰的函数，但没有使用 triton.language (tl) API。"
+ "kernel 必须使用 tl.load/tl.store 等进行显式内存操作和计算。"
+ )
+ return result
+
+ result["checks"]["triton_kernel_exists"]["passed"] = True
+
+ # --- Check 2: 禁止的 torch 操作 ---
+ violations = check_forbidden_torch_ops(tree, kernel_ranges)
+ result["checks"]["no_forbidden_torch_ops"]["violations"] = violations
+
+ if violations:
+ result["checks"]["no_forbidden_torch_ops"]["error"] = (
+ f"代码中发现 {len(violations)} 处禁止的 PyTorch 计算操作"
+ )
+ violation_details = "; ".join(
+ f"第{v['line']}行 {v['call']}" for v in violations[:5]
+ )
+ result["regression_type"] = 2
+ result["suggestion"] = (
+ f"代码中使用了禁止的 PyTorch 计算操作: {violation_details}。"
+ "所有核心计算必须在 @triton.jit kernel 中完成，"
+ "仅允许 buffer 分配和形状操作。"
+ )
+ return result
+
+ result["checks"]["no_forbidden_torch_ops"]["passed"] = True
+
+ # --- 全部通过 ---
+ result["valid"] = True
+ return result
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+def main():
+ parser = argparse.ArgumentParser(
+ description="检查生成代码是否退化为 PyTorch 原生实现（AST 静态分析）"
+ )
+ parser.add_argument("file", help="要检查的 Python 文件路径")
+ parser.add_argument("--json", action="store_true", help="JSON 格式输出")
+ args = parser.parse_args()
+
+ try:
+ with open(args.file, "r", encoding="utf-8") as f:
+ code = f.read()
+ except FileNotFoundError:
+ if args.json:
+ print(json.dumps({"valid": False, "error": f"文件不存在: {args.file}"}))
+ else:
+ print(f"[ERROR] 文件不存在: {args.file}")
+ sys.exit(1)
+
+ result = validate(code, filepath=args.file)
+
+ if args.json:
+ print(json.dumps(result, ensure_ascii=False, indent=2))
+ else:
+ if result["valid"]:
+ kernels = result["checks"]["triton_kernel_exists"]["kernels"]
+ print("[PASS] Triton 实现验证通过")
+ print(f" - 发现 {len(kernels)} 个有效 @triton.jit kernel: {', '.join(k['name'] for k in kernels)}")
+ print(" - 代码中无禁止的 PyTorch 计算操作")
+ else:
+ rtype = result["regression_type"]
+ type_desc = {
+ 1: "完全无 Triton kernel（纯 PyTorch）",
+ 2: "部分计算使用 PyTorch（需全部移入 Triton kernel）",
+ }
+ print(f"[FAIL] 检测到 PyTorch 退化 — Type {rtype}: {type_desc.get(rtype, '未知')}")
+
+ # 显示具体检查结果
+ for check_name, check_result in result["checks"].items():
+ status = "PASS" if check_result["passed"] else "FAIL"
+ print(f" [{status}] {check_name}")
+ if check_result["error"]:
+ print(f" {check_result['error']}")
+
+ if result["checks"]["no_forbidden_torch_ops"]["violations"]:
+ print(" 违规详情:")
+ for v in result["checks"]["no_forbidden_torch_ops"]["violations"]:
+ print(f" 第 {v['line']} 行: {v['call']} — {v['reason']}")
+
+ print(f"\n 修复建议: {result['suggestion']}")
+
+ sys.exit(0 if result["valid"] else 1)
+
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
diff --git a/skills/triton/kernel-triton-verifier/scripts/verify.py b/skills/triton/kernel-triton-verifier/scripts/verify.py
new file mode 100644
index 00000000..91700248
--- /dev/null
+++ b/skills/triton/kernel-triton-verifier/scripts/verify.py
@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+"""算子验证脚本 — 算子对应.pt文件中包含输入以及预期的输出, 相同输入下, 对比生成算子输出与预期输出的一致性。
+
+用法:
+ python verify.py --op_name <算子名> [--verify_dir <验证目录>] [--timeout <超时秒数>] [--device_id <所用设备id>]
+
+前置条件（验证目录下需存在以下文件）:
+ {op_name}.pt — 包含输入，预期输出
+ {op_name}.py — 包含生成算子的主要逻辑
+"""
+import argparse
+import os
+import sys
+import torch
+import importlib
+import gc
+
+from test_common import convert_tensor_with_device_type, compare_data_precision
+
+
+# 🔥 强制清空 NPU 缓存 + 内存
+def clear_npu_memory():
+ try:
+ torch.npu.empty_cache() # 清空NPU缓存
+ torch.npu.synchronize() # 强制同步所有操作
+ gc.collect() # 强制Python垃圾回收
+ except:
+ pass
+
+# 🔥 安全卸载动态导入的模块（防止句柄泄漏卡死）
+def unload_module(module_name):
+ if module_name in sys.modules:
+ del sys.modules[module_name]
+ gc.collect()
+
+def verify_implementations(op_name, verify_dir, triton_impl_name):
+ """验证框架实现和生成实现的结果一致性"""
+ try:
+ spec = importlib.util.spec_from_file_location(op_name, f"{verify_dir}/{op_name}.py")
+ triton_npu_module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(triton_npu_module)
+
+ # 获取 kernel 函数
+ triton_npu_func = getattr(triton_npu_module, op_name)
+
+ data = torch.load(f"{verify_dir}/{op_name}.pt", map_location=torch.device('cpu'), weights_only=False)
+
+ input_data = convert_tensor_with_device_type(data["input_data"], device_type='npu')
+
+ triton_npu_func[data["grid"]](**input_data)
+ torch.npu.synchronize()
+
+ compare_data_precision(data["gpu_output"], input_data, device_type='cpu')
+ print("验证成功")
+ except BaseException as e:
+ print(f"❌【失败】执行报错：{str(e)}")
+ finally:
+ # 🔥 终极清理：必须执行，否则连续跑必卡死
+ try:
+ # 清理变量
+ locals().clear()
+ gc.collect()
+
+ # 卸载动态模块
+ unload_module(op_name)
+
+ # 强制清空NPU
+ clear_npu_memory()
+
+ # 移除所有临时引用
+ module = None
+ except:
+ pass
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description="算子验证脚本")
+ parser.add_argument("--op_name", required=True, help="算子名称")
+ parser.add_argument(
+ "--verify_dir", default=".",
+ help="验证目录，包含 {op_name}.pt算子输入 和 test_{op_name}.py算子的triton逻辑",
+ )
+ parser.add_argument("--timeout", type=int, default=900, help="超时秒数（默认 900）")
+ parser.add_argument(
+ "--triton_impl_name", default="triton_ascend_impl",
+ help="Triton 实现模块名（不含 op_name 前缀，默认 triton_ascend_impl）",
+ )
+ parser.add_argument(
+ "--_run", action="store_true",
+ help=argparse.SUPPRESS, # 内部参数：子进程模式，直接执行验证
+ )
+ parser.add_argument(
+ "--device_id", required=True, type=int, default=0, help="指定npu卡"
+ )
+
+ args = parser.parse_args()
+
+ torch.npu.set_device(args.device_id)
+
+ verify_dir = os.path.abspath(args.verify_dir)
+ if not os.path.isdir(verify_dir):
+ print(f"错误: 验证目录不存在: {verify_dir}", file=sys.stderr)
+ sys.exit(1)
+
+ try:
+ verify_implementations(args.op_name, verify_dir, args.triton_impl_name)
+ except Exception as e:
+ print(f"{e}", file=sys.stderr)
+ sys.exit(1)