From 93675c97a6c7a4c87b81059ba4bfc7b94f44cee1 Mon Sep 17 00:00:00 2001
From: zhumingming <zhumingming1@huawei.com>
Date: Mon, 13 Apr 2026 09:22:49 +0800
Subject: [PATCH] =?UTF-8?q?=E3=80=90benchmark=E3=80=91=E6=96=B0=E5=A2=9Eke?=
 =?UTF-8?q?rnel-triton-verifier=20skill,=20=E6=94=AF=E6=8C=81=E4=B8=8Etrit?=
 =?UTF-8?q?on-gpu=E7=AE=97=E5=AD=90=E8=BF=9B=E8=A1=8C=E7=B2=BE=E5=BA=A6/?=
 =?UTF-8?q?=E6=80=A7=E8=83=BD=E6=AF=94=E8=BE=83?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 skills/triton/kernel-triton-verifier/SKILL.md | 313 ++++++++++++
 .../scripts/benchmark.py                      | 467 ++++++++++++++++++
 .../scripts/test_common.py                    |  69 +++
 .../scripts/validate_triton_impl.py           | 389 +++++++++++++++
 .../kernel-triton-verifier/scripts/verify.py  | 109 ++++
 5 files changed, 1347 insertions(+)
 create mode 100644 skills/triton/kernel-triton-verifier/SKILL.md
 create mode 100644 skills/triton/kernel-triton-verifier/scripts/benchmark.py
 create mode 100644 skills/triton/kernel-triton-verifier/scripts/test_common.py
 create mode 100644 skills/triton/kernel-triton-verifier/scripts/validate_triton_impl.py
 create mode 100644 skills/triton/kernel-triton-verifier/scripts/verify.py

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
+
+<role>
+你是一个内核代码验证专家。你的任务是按照标准验证流程，创建验证项目并运行，检查生成的算子代码是否能正确编译运行且与参考实现的输出一致。验证通过后，执行性能测试并收集性能数据。
+</role>
+
+## 验证流程
+
+```
+输入：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 <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`: `<file_path>`, `[--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 <file_path> [--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="<unknown>"):
+    """对生成代码执行完整的退化检查。
+
+    返回结构化结果 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)