他当时的情况是:装了 CANN,跑了 PyTorch 模型,NPU 也认到了,但不理解 ops-transformer 到底属于哪一层——是模型代码的一部分,还是 CANN 的一部分,还是昇腾硬件的一部分?

这个问题太常见了。用一个厨房的比喻把整个系统说清楚。

先把整个系统想象成一个厨房

你的大模型(比如 LLaMA)就是一份菜谱。菜谱上写着"先炒鸡蛋,再加西红柿,最后出锅"——这对应模型代码里的 forward 计算图。

昇腾 NPU 是厨房里的灶台。灶台本身只认一种指令——“点火”“调火”“关火”,不认"把鸡蛋炒到七分熟"这种人类语言。

那问题来了:菜谱怎么变成灶台能懂的操作?

昇腾异构计算架构 CANN 就是中间那层翻译。 它把你的模型代码翻译成 NPU 能执行的指令,再调度算子真正在硬件上跑起来。

ops-transformer 是这整套流程里,专门负责"炒蛋"那个步骤的专用锅具。FlashAttention、MoE、MC2 这些算子,就是这个锅具里的核心功能。

五层架构的通俗版解释

CANN 五层架构听起来吓人,但其实每层干的事情一句话就能说清楚:

第一层:AscendCL——这是给应用层用的接口。你可以把它理解为厨房的控制面板,上面有"开火"“调温”"定时"几个按钮。你写 Python 代码调用 torch.npu() 的时候,其实就是在按这个控制面板。

怎么验证你在用这一层:

import torch
# 检查 NPU 是否可用
print(torch.npu.is_available())  # True 说明 AscendCL 正常
print(torch.npu.device_count())  # 应该看到 NPU 数量

# 或者用 ascendcl 的原生接口
import acl
acl.init()  # 初始化 AscendCL
print(f"ACL version: {acl.__version__}")

第二层:AOL 算子库——这是真正能干活的地方。算子库里有几百个现成的算子,每个算子对应一种常见的计算操作。ops-transformer 就在这一层,它提供 FlashAttention、MoE 这些大模型训练里用得最多的算子。

怎么验证你的 ops-transformer 算子已注册:

# 检查 ops-transformer 算子是否在 PyTorch 里注册成功
import torch_npu
from torch_npu.contrib import register_acc_ops

# 打印已注册的算子列表
print(dir(torch_npu.contrib))

# 验证 FlashAttention 是否可调用
try:
    from flash_attention_ops import flash_attention_npu
    print("FlashAttention 算子注册成功")
except ImportError as e:
    print(f"算子未注册: {e}")

第三层:Graph Compiler(GE)——这是灶台旁边的自动炒菜机。它把你的整个计算图(所有算子的组合)做一个全局优化:哪里可以合并、哪里可以跳过、哪里需要重新排顺序,让整个流程跑得更快。GE 会做算子融合——本来要分三步走的计算,合并成一步直接在芯片上跑,省掉中间来回搬数据的时间。

查看 GE 融合日志的方法:

# 设置环境变量让 GE 输出详细的融合信息
export ASCEND_GLOBAL_LOG_LEVEL=3
export ENABLE_OOL_OP痕过融合LOG=1

# 重新跑训练脚本,日志会输出 GE 的融合决策过程
python train_llama.py 2>&1 | tee ge_fusion_log.txt

# 搜索融合相关的输出
grep -E "(Fusion|flash_attention|融合)" ge_fusion_log.txt

融合成功的日志长这样:

[GE Fusion] 子图 #15 检测到算子序列:
  MatMul[qkt] + Softmax + MatMul[pv]
  匹配融合规则: flash_attention_fusion_pass
  融合为单一算子: FlashAttentionKernel
  输出 shape: (batch, heads, seq_len, dim)

第四层:Runtime——这是调度员。它负责把计算任务分配到具体的 NPU 计算单元上,决定谁先跑谁后跑,多个任务怎么排队,怎么管理显存。

监控 Runtime 调度效率的方法:

# 用 npu-smi 实时监控 NPU 利用率和 HBM 带宽
watch -n 0.5 npu-smi dmon -c 1 -s puc -d 10

# 输出解读:puc = Power Utilization + Compute utilization
# 如果 NPU 利用率长期 < 60%,问题大概率出在 Runtime 调度

第五层:硬件驱动——这是灶台的底层电路,直接对接昇腾 NPU 的计算单元 Cube/Vector 和存储层次。

ops-transformer 在第二层,处于一个"承上启下"的位置:上层被 GE 的算子融合优化调用,下层依赖 opbase 提供的基础算子组件。

算子从模型代码到 NPU 执行的全过程

用一个具体例子说清楚这个流程。

当你用 PyTorch 写一行 output = attention(q, k, v) 时,背后发生的事情大概是这个顺序:

第一步:Framework Adaptor 注册算子。 PyTorch 的 npu() 调用会通过 CANN 的框架适配层,把 ops-transformer 里的 FlashAttention 算子注册到系统中。这一步做的是"告诉系统我需要这个算子,它在哪里"。

# PyTorch 模型代码中,调用的入口是这个
model = model.npu()  # 触发 Framework Adaptor 的注册

# 或者用更直接的方式,把整个模型搬到 NPU 上
model = model.to("npu:0")

# 注册验证:看算子有没有被正确识别
import torch_npu
print(torch_npu.list_registered_ops())  # 看 ops-transformer 的算子是否在列表里

第二步:GE 图编译。 CANN 的图编译器拿到完整的计算图,开始做优化。它发现 FlashAttention 这个算子可以用 ops-transformer 里的融合版本实现(不需要分别调用 MatMul → Softmax → MatMul 三个算子),于是把三个算子合并成一次调用。这就是 GE 做的事——全局视角的算子融合和调度优化。

# 在模型里触发 GE 的融合
# 融合发生的时间点:在首次 forward 时触发 JIT 编译
# 编译完成后,后续调用直接走融合后的执行计划

output = model(batch_input)  # 第一次 forward,GE 做图编译 + 融合
# 第二次开始,融合执行计划已经生成
output = model(batch_input2)  # 走的是融合后的路径

第三步:Runtime 调度。 优化后的算子被拆分成具体的计算任务,按时间片分配给 NPU 的 Cube 计算单元(矩阵乘专用)和 Vector 计算单元(向量操作专用)。算子的输入数据从 HBM 加载到片上统一缓冲区(UB),计算在 UB 内完成,结果写回 HBM。

# 用 Profiler 观察 Runtime 的调度行为
from torch_npu.profiler import profile

with profile(
    activities=[
        torch_npu.profiler.ProfilerActivity.NPU,
        torch_npu.profiler.ProfilerActivity.CPU,
    ],
    record_shapes=True,
    with_stack=True,
    export_name="trace.json")
:
    output = model(batch_input)

# 打开 Profiler GUI 分析 timeline
# ascendsnorkel-validate -i trace.json

第四步:ops-transformer 执行。 具体到 FlashAttention 的实现,它在 UB 内完成 tile 级别的 QK^T → Softmax → 乘 V 的全部计算,利用昇腾 NPU 的异步搬运能力把下一个 tile 的数据提前预加载进来,数据搬运和计算并行。

# ops-transformer 的实际调用方式
from flash_attention_ops import flash_attention_npu

# 构造输入
batch, heads, seq_len, dim = 4, 32, 2048, 64
q = torch.randn(batch, heads, seq_len, dim, dtype=torch.float16).npu()
k = torch.randn(batch, heads, seq_len, dim, dtype=torch.float16).npu()
v = torch.randn(batch, heads, seq_len, dim, dtype=torch.float16).npu()

# 调用融合算子
output = flash_attention_npu(q, k, v, causal=True)
print(f"输出 shape: {output.shape}")

这四步中,ops-transformer 只负责第四步——真正的计算执行。前三步由其他组件完成,但正是这种分层设计让系统可以分别优化、灵活组合。

理解这个分层有什么实际用处

知道 ops-transformer 在第二层,不是为了考试,是为了解决实际问题。

当你发现训练速度不达预期时,你不会直接去看 ops-transformer 的源码,而是先用 CANN Profiler 抓一次 trace,看 GE 的算子融合有没有生效。等你确认问题出在 FlashAttention 本身没被正确调用时,再去看 ops-transformer 的实现。

反过来,如果你发现某个算子在 ops-transformer 里没有对应的融合实现,而你需要手动改一个融合策略,那你需要往 GE 层去了解算子融合的规则是什么、怎么注册自定义融合 pass——这是另一个层级的事情。

# 调优排查的标准流程
# 1. 先用 Profiler 确认瓶颈在哪一层
# 如果 Timeline 上 FusionKernel 的耗时占比高 → 看 ops-transformer 实现
# 如果数据搬运的等待时间占比高 → 看 Runtime 调度(batch size / seq_len)
# 如果多个小算子紧挨着没融合 → 看 GE 融合日志(shape/dtype 对齐了吗)

# 2. 检查 GE 融合状态
import os
print(os.environ.get("ASCEND_GLOBAL_LOG_LEVEL", "未设置"))
# 融合开启时,日志中会有 [GE Fusion] 字样

# 3. 验证算子注册
from torch_npu import profiler
print("Profiler 可用:", hasattr(profiler, "profile"))

分层架构的核心价值在于:每一层都有自己的关注点,定位清晰,出了问题知道往哪找。

GE 和 Runtime 在 ops-transformer 场景里的实际角色

有人问:ops-transformer 跟 GE 是什么关系?跟 Runtime 又是什么关系?

简单说:ops-transformer 是被调用的对象,GE 是优化器,Runtime 是调度员。

当你没有 GE 时,ops-transformer 的算子还是能跑——一次调一个,按顺序执行。但加了 GE 之后,GE 会把你的模型里所有算子做一个全局视图的优化,决定 ops-transformer 里的哪些算子可以被合并、哪些可以跳过、哪些需要重排执行顺序。Runtime 则负责把这些决策变成具体的时间片分配,让多个算子在不同计算单元上并行。

GE 和 Runtime 属于 CANN 第三层和第四层的基础设施,不属于 ops-transformer 本身。但你在学习 ops-transformer 时,需要了解它们是怎么影响算子执行的——不然你调 API 的时候完全不知道背后发生了什么。

为什么这整套架构值得学

说一个实际的理由。

过去要搞懂这套东西,要么参加华为的官方培训,要么有内部渠道才能接触文档。2025年8月昇腾 CANN 全面开源之后,整个体系全部在 AtomGit 上公开了——GE 的源码、Runtime 的实现、ops-transformer 的算子代码、cann-learning-hub 的全套教程。

也就是说,现在你完全可以通过自学搞清楚一条从应用层到硬件层的完整链路。以前这是需要内部权限才能做到的事情。

你现在缺的只是时间和耐心,不是资源。

怎么开始学

建议从 cann-learning-hub 的架构概览开始,先把五层架构搞清楚。然后选一个你关心的具体场景(比如大模型训练),顺着 cann-samples 的 FlashAttention 示例跑通一次调用。再往后是读 ops-transformer 的源码,理解融合算子的内部实现。

学完这一圈,你会对整个系统有完全不同的理解——不再是一个"调用 PyTorch 的黑盒",而是一套有层次、有逻辑、有分工的工程体系。

# 快速上手的完整命令流
# 第一步:检查环境
npu-smi info
python -c "import torch_npu; print(torch_npu.__version__)"

# 第二步:克隆仓库
git clone https://atomgit.com/cann/ops-transformer
cd ops-transformer

# 第三步:跑通示例
cd examples
python flash_attention_benchmark.py \
    --batch 4 --heads 32 --seq_len 2048 --dtype float16

# 第四步:验证 GE 融合状态
# 输出中如果有 "GE fusion: enabled" → 融合已触发
# 如果是 "GE fusion: disabled" → 检查 dtype/shape 对齐

相关仓库:

https://atomgit.com/cann/ops-transformer

https://atomgit.com/cann/cann-learning-hub

https://atomgit.com/cann/ge

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐