前言

昇腾NPU作为华为昇腾系列AI处理器的核心算力单元，其软件开发体系依赖CANN（Compute Architecture for Neural Networks）作为基础软件栈。CANN提供完整的算子开发接口与运行时环境，支持开发者在昇腾AI处理器上实现高性能自定义算子。传统Ascend C算子开发采用C++扩展语法，对Python生态开发者存在一定门槛。pyasc项目将Ascend C编程接口完整映射为Python原生语法，使开发者能够使用标准Python语言编写在昇腾NPU上运行的自定义算子。本文以矢量加法算子为实操案例，完整演示从环境配置、算子开发、编译部署到调试验证的端到端流程，所有步骤均可在真实环境中复现。

pyasc是一种用于编写高效自定义算子的编程语言，原生支持Python标准规范。基于pyasc编写的算子程序，通过编译器编译和运行时调度，运行在昇腾AI处理器上。pyasc编程接口与Ascend C类库接口一一对应，旨在提供与Ascend C接口相同的编程能力。目前pyasc正逐步开放对Ascend C API的支持，有关pyasc编程接口的支持范围和约束，可以参考Ascend C API文档。对于编程所需的抽象硬件架构和编程模型的相关知识，可以参考Ascend C算子开发文档。本项目支持的AI处理器包括：Ascend 910C、Ascend 910B。

pyasc技术背景与编程模型

pyasc项目的核心目标是为Python开发者提供与Ascend C一一对应的编程接口。Ascend C是CANN提供的C++算子编程接口，直接操作昇腾NPU的向量、矩阵、标量计算单元。pyasc通过Python前端、MLIR中间表示、LLVM后端的三层编译架构，将Python DSL转换为昇腾NPU可执行的二进制代码。

pyasc编译器前端使用Python AST解析用户代码，将其转换为MLIR的Ascend C IR方言（ascir）。ascir是pyasc定义的中间表示，描述算子的计算过程、内存访问模式、并行化策略。MLIR后端将ascir逐步降低到LLVM IR，最终通过LLVM后端生成昇腾NPU的机器码。

运行时系统负责算子注册、内存管理、内核启动。pyasc生成的算子通过GE（Graph Engine）注册到CANN计算图执行引擎，与框架算子无缝集成。算子执行时，运行时系统自动处理Host到Device的数据搬运、内存分配、流同步等底层细节。

硬件与软件配套关系

pyasc所需的软硬件环境依赖如下：昇腾产品支持Atlas A2训练/推理产品和Atlas A3训练/推理产品。CPU架构支持aarch64和x86_64。操作系统支持经过兼容性查询的Linux发行版。软件依赖包括Python 3.9至3.12版本。

不同pyasc发行版可支持的硬件平台及所需的CANN版本如下表所示：

pyasc社区版本	支持CANN包版本	支持昇腾产品
v1.1.0、v1.1.1	社区版8.5.0.alpha001及以上	Atlas A2训练/推理产品、Atlas A3训练/推理产品
v1.0.0	社区版8.5.0.alpha001、8.5.0.alpha002	Atlas A2训练/推理产品、Atlas A3训练/推理产品

pyasc与Ascend C语法对照

pyasc将Ascend C的C++扩展语法完整映射为Python原生语法，主要接口对应关系如下：

特性	Ascend C (.asc)	pyasc (Python)
编程语言	C++ 扩展语法	原生 Python
核函数定义	global aicore	@asc.jit 装饰器
GlobalTensor	AscendC::GlobalTensor	asc.GlobalTensor()
LocalTensor	AscendC::LocalTensor	asc.LocalTensor()
数据搬运	AscendC::DataCopy()	asc.data_copy()
矢量计算	AscendC::Add()	asc.add()
同步事件	AscendC::SetFlag()/WaitFlag()	asc.set_flag()/wait_flag()
核函数调用	add_custom<<<…>>>()	vadd_kernelnum_blocks, stream

pyasc环境配置手把手实操

系统依赖检查

在开始安装pyasc之前，需要确认系统满足以下基础依赖条件。Python版本需要在3.9.0到3.12.0之间。cmake版本需要大于等于3.20。GLIBC版本需要大于等于2.31。LLVM版本需要大于等于9.4.0。

执行以下命令检查系统环境：

# WHY: 检查Python版本 - pyasc支持CPython 3.9-3.12，版本不匹配会导致安装失败
python3 --version

# WHY: 检查cmake版本 - pyasc基于CMake构建系统，版本低于3.20会导致构建脚本报错
cmake --version

# WHY: 检查GLIBC版本 - pyasc依赖的系统库与GLIBC版本强相关，版本过低会导致运行时动态链接失败
ldd --version

# WHY: 检查CPU架构 - pyasc提供ARM64和x86_64两种架构的预编译包，需要确认当前系统架构
uname -m

CANN软件包安装

pyasc运行依赖CANN软件包。根据是否有NPU设备，选择对应的安装方式。

无NPU设备的环境准备

对于无NPU设备的开发者，可使用云开发环境提供的NPU计算资源。进入开源仓Gitcode页面，单击"云开发"按钮，使用已认证过的华为云账号登录。

若需要在本地环境安装CANN包，执行以下步骤：

# WHY: 下载CANN toolkit包 - toolkit包包含算子开发所需的基础库和头文件，是pyasc编译运行的必要依赖
# 从昇腾社区下载对应版本的CANN toolkit包
chmod +x Ascend-cann-toolkit_${cann_version}_linux-$(uname -m).run

# WHY: 使用--install参数 - 该参数执行全自动安装，无需交互式确认，适合脚本化部署
./Ascend-cann-toolkit_${cann_version}_linux-$(uname -m).run --install --install-path=${install_path}

# WHY: 安装CANN ops包 - ops包包含预编译的算子实现库，pyasc生成的自定义算子需要与这些库链接
# 910B为Ascend-cann-910b-ops_8.5.0_linux-x86_64.run
# 910C为Ascend-cann-A3-ops_8.5.0_linux-x86_64.run
chmod +x Ascend-cann-${soc_name}-ops_${cann_version}_linux-$(uname -m).run
./Ascend-cann-${soc_name}-ops_${cann_version}_linux-$(uname -m).run --install --install-path=${install_path}

有NPU设备的环境准备

对于有NPU设备的开发者，推荐使用CANN官方Docker镜像进行开发。

# WHY: 使用Docker镜像 - 官方镜像已预集成NPU驱动、固件和CANN包，避免手动配置的兼容性问题
# 从昇腾镜像仓库拉取已预集成CANN镜像
docker pull swr.cn-south-1.myhuaweicloud.com/ascendhub/cann:9.0.0-beta.2-910b-ubuntu22.04-py3.11

# WHY: 需要这些参数 - --ipc=host和--net=host确保容器与宿主机共享IPC和网络栈，NPU设备访问所需
# --privileged赋予容器完整设备访问权限
# --device参数将宿主机的NPU设备映射到容器内
docker run --name pyasc_dev \
  --ipc=host --net=host --privileged \
  --device /dev/davinci0 \
  --device /dev/davinci_manager \
  --device /dev/devmm_svm \
  --device /dev/hisi_hdc \
  -v /usr/local/dcmi:/usr/local/dcmi \
  -v /usr/local/bin/npu-smi:/usr/local/bin/npu-smi \
  -v /usr/local/Ascend/driver/lib64/:/usr/local/Ascend/driver/lib64/ \
  -v /usr/local/Ascend/driver/version.info:/usr/local/Ascend/driver/version.info \
  -v /etc/ascend_install.info:/etc/ascend_install.info \
  -v $(pwd):/workspace \
  -it swr.cn-south-1.myhuaweicloud.com/ascendhub/cann:9.0.0-beta.2-910b-ubuntu22.04-py3.11 \
  bash

pyasc源码下载与依赖安装

# WHY: 从Gitcode克隆源码 - pyasc项目托管在Gitcode平台，克隆master分支获取最新开发代码
git clone https://gitcode.com/cann/pyasc.git
cd pyasc

# WHY: 安装build-time依赖 - requirements-build.txt包含CMake、setuptools等构建时所需的Python包
python3 -m pip install -r requirements-build.txt

# WHY: 安装run-time依赖 - requirements-runtime.txt包含pyasc运行时所需的Python包，如numpy等
python3 -m pip install -r requirements-runtime.txt

LLVM预编译包安装

pyasc的编译后端基于LLVM框架，需要安装LLVM 19.1.7或更高版本。

# WHY: 下载LLVM预编译包 - pyasc使用LLVM作为编译后端，将Python DSL编译为Ascend C中间表示
# 根据系统架构选择对应命令，以下为ARM架构示例
wget https://cann-ai.obs.cn-north-4.myhuaweicloud.com/llvm/llvm-19.1.7-aarch64.tar.xz

# WHY: 使用tar -xJf解压 - xz压缩格式提供更高的压缩比，-J参数指定使用xz解压算法
tar -xJf llvm-19.1.7-aarch64.tar.xz

# WHY: 设置LLVM_INSTALL_PREFIX环境变量 - pyasc构建脚本通过该变量定位LLVM安装路径
export LLVM_INSTALL_PREFIX=$PWD/llvm-19.1.7-aarch64

# WHY: 验证LLVM安装 - 确认LLVM正确安装后，才能确保后续pyasc构建过程不出现链接错误
${LLVM_INSTALL_PREFIX}/bin/llvm-config --version

# WHY: 检查libz和libzstd - 这两个库是LLVM和pyasc运行时的依赖库，缺失会导致导入pyasc时动态链接失败
test -f /usr/lib/$(uname -m)-linux-gnu/libz.so && echo "libz.so: [OK]" || echo "libz.so: [MISSING]"
test -f /usr/lib/$(uname -m)-linux-gnu/libzstd.so && echo "libzstd.so: [OK]" || echo "libzstd.so: [MISSING]"

# 如果缺少libz或libzstd，执行安装
sudo apt-get install zlib1g-dev libzstd-dev

pyasc安装验证

# WHY: 使用pip安装 - pyasc提供Python包封装，通过pip安装可自动处理依赖关系和安装路径
# 普通模式：将项目安装到Python环境的site-packages目录中，本地修改不影响已安装版本
python3 -m pip install .

# WHY: 开发者模式 - 仅创建符号链接，本地修改实时生效，无需重新安装，适合开发阶段
# python3 -m pip install -e .

# WHY: 检查pyasc是否安装成功 - 确认安装后，才能确保后续算子开发过程中能够正确导入asc模块
pip3 list | grep -w "pyasc"

运行环境变量配置

# WHY: 需要设置环境变量 - CANN软件包的路径需要加入系统环境变量，pyasc运行时通过环境变量定位CANN库
# 默认路径安装，以root用户为例（非root用户，将/usr/local替换为${HOME}）
source /usr/local/Ascend/cann/set_env.sh

# WHY: 仿真器模式需要额外设置 - 仿真器模式下，pyasc使用Ascend910B1 simulator运行算子，需要加载仿真动态库
export LD_LIBRARY_PATH=$ASCEND_HOME_PATH/tools/simulator/Ascend910B1/lib:$LD_LIBRARY_PATH

# WHY: 设置LD_PRELOAD - torch_npu默认只支持NPU上板，仿真器模式运行需要提前加载libruntime_camodel.so
export LD_PRELOAD=libruntime_camodel.so

# WHY: 取消LD_PRELOAD - 若pyasc后端采用NPU处理器运行，需取消LD_PRELOAD环境变量，避免加载错误的运行时库
# unset LD_PRELOAD

第一个pyasc算子：矢量加法手把手实现

算子分析与设计规格

在实现算子前，需要对算子进行完整分析，明确数学表达式、输入输出规格以及所需接口。

以Add算子为例，算子的数学表达式为：

z = x + y

计算逻辑是：从外部存储Global Memory搬运数据至内部存储Local Memory，再使用pyasc计算接口完成两个输入参数相加，得到最终结果，再搬运到Global Memory上。

Add算子的设计规格如下：

项目	规格
算子类型(OpType)	Add
算子输入x	shape: (8, 2048), data type: float, format: ND
算子输入y	shape: (8, 2048), data type: float, format: ND
算子输出z	shape: (8, 2048), data type: float, format: ND
核函数名	vadd_kernel
主要接口	asc.data_copy：数据搬运接口
	asc.add：矢量基础算术接口
	asc.GlobalTensor/LocalTensor：内存管理接口
	asc.set_flag/wait_flag：同步接口
算子实现文件名称	add.py

核函数开发完整代码

以下是完整的矢量加法算子实现代码，包含详细的注释说明：

# WHY: 导入asc模块 - asc模块提供pyasc的核心编程接口，包括Tensor管理、数据搬运、矢量计算等功能
import asc
# WHY: 导入runtime模块 - runtime模块提供pyasc的运行时配置和流管理接口
import asc.lib.runtime as rt
import asc.runtime.config as config

# WHY: 定义USE_CORE_NUM - 指定并行计算的核数，多核并行是提升NPU利用率的关键手段
USE_CORE_NUM = 8
# WHY: 定义BUFFER_NUM - 双缓冲机制允许数据搬入和计算过程重叠执行，提升流水线效率
BUFFER_NUM = 2
# WHY: 定义TILE_NUM - 每个核上数据分块个数，分块计算是实现流水线并行的基础
TILE_NUM = 8

# WHY: 使用@asc.jit装饰器 - 该装饰器将Python函数标记为pyasc核函数，触发Python DSL到Ascend C的编译流程
@asc.jit
def vadd_kernel(x: asc.GlobalAddress, y: asc.GlobalAddress, z: asc.GlobalAddress, block_length: int):
    # WHY: 获取当前核的索引 - 多核并行场景下，每个核处理不同的数据分片，需要根据核索引计算数据偏移
    offset = asc.get_block_idx() * block_length
    
    # WHY: 创建GlobalTensor - GlobalTensor管理Global Memory上的数据，是NPU与外部存储交互的接口
    x_gm = asc.GlobalTensor()
    y_gm = asc.GlobalTensor()
    z_gm = asc.GlobalTensor()
    
    # WHY: 设置global buffer - 将GlobalTensor绑定到具体的全局内存地址和长度，建立地址映射关系
    x_gm.set_global_buffer(x + offset, block_length)
    y_gm.set_global_buffer(y + offset, block_length)
    z_gm.set_global_buffer(z + offset, block_length)
    
    # WHY: 计算每个tile的长度 - 将单核数据进一步分块，实现流水线并行，tile_length是每个数据块的元素个数
    tile_length = block_length // TILE_NUM // BUFFER_NUM
    
    # WHY: 获取数据类型信息 - 不同数据类型的sizeof不同，计算buffer大小需要依据具体的数据类型
    data_type = x.dtype
    # WHY: 计算buffer_size - Local Memory空间有限，需要精确计算每个Tensor所需的存储空间
    buffer_size = tile_length * BUFFER_NUM * data_type.sizeof()
    
    # WHY: 创建LocalTensor - LocalTensor管理Local Memory上的数据，计算操作直接在Local Memory上进行
    # WHY: 指定TPosition.VECIN - VECIN是矢量计算输入数据的逻辑位置，指定位置有助于编译器优化内存访问
    x_local = asc.LocalTensor(data_type, asc.TPosition.VECIN, 0, tile_length * BUFFER_NUM)
    y_local = asc.LocalTensor(data_type, asc.TPosition.VECIN, buffer_size, tile_length * BUFFER_NUM)
    # WHY: 指定TPosition.VECOUT - VECOUT是矢量计算输出数据的逻辑位置
    z_local = asc.LocalTensor(data_type, asc.TPosition.VECOUT, buffer_size + buffer_size, tile_length * BUFFER_NUM)
    
    # WHY: 循环TILE_NUM * BUFFER_NUM次 - 双缓冲机制需要每个buffer处理两次（ping-pong），循环次数为分块数*缓冲数
    for i in range(TILE_NUM * BUFFER_NUM):
        # WHY: 使用buf_id = i % BUFFER_NUM - 通过取模运算实现ping-pong缓冲区的切换
        buf_id = i % BUFFER_NUM
        
        # Step1: 搬入 - WHY: 使用data_copy - data_copy是pyasc提供的数据搬运接口，实现Global Memory到Local Memory的数据拷贝
        asc.data_copy(x_local[buf_id * tile_length:], x_gm[i * tile_length:], tile_length)
        asc.data_copy(y_local[buf_id * tile_length:], y_gm[i * tile_length:], tile_length)
        
        # WHY: 设置同步事件 - 数据搬入是异步操作，需要等待搬入完成后才能进行计算，set_flag/wait_flag实现事件同步
        asc.set_flag(asc.HardEvent.MTE2_V, buf_id)
        asc.wait_flag(asc.HardEvent.MTE2_V, buf_id)
        
        # Step2: 计算 - WHY: 使用asc.add - asc.add是pyasc提供的矢量加法接口，在Local Memory上执行矢量加法
        asc.add(z_local[buf_id * tile_length:], x_local[buf_id * tile_length:],
                y_local[buf_id * tile_length:], tile_length)
        
        # WHY: 设置同步事件 - 矢量计算是异步操作，需要等待计算完成后才能搬出数据
        asc.set_flag(asc.HardEvent.V_MTE3, buf_id)
        asc.wait_flag(asc.HardEvent.V_MTE3, buf_id)
        
        # Step3: 搬出 - WHY: 使用data_copy - 将计算结果从Local Memory拷贝回Global Memory
        asc.data_copy(z_gm[i * tile_length:], z_local[buf_id * tile_length:], tile_length)
        
        # WHY: 需要MTE3_MTE2同步 - 确保数据搬出完成后再启动下一轮数据搬入，避免读写冲突
        asc.set_flag(asc.HardEvent.MTE3_MTE2, buf_id)
        asc.wait_flag(asc.HardEvent.MTE3_MTE2, buf_id)

核函数调用封装

# WHY: 封装launch函数 - 将核函数调用封装为Python函数，提供与PyTorch一致的调用接口
def vadd_launch(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
    # WHY: 创建输出Tensor - 算子输出需要预先分配内存，torch.zeros_like创建与输入相同shape和dtype的输出Tensor
    z = torch.zeros_like(x)
    
    # WHY: 计算total_length - 获取张量中元素的总数，用于计算分片大小
    total_length = z.numel()
    # WHY: 计算每个核的处理长度 - 将总数据量平均分配到多个核上并行处理
    block_length = total_length // USE_CORE_NUM
    
    # WHY: 使用内核调用符 - vadd_kernel[USE_CORE_NUM, rt.current_stream()]指定核数和流，触发核函数执行
    vadd_kernel[USE_CORE_NUM, rt.current_stream()](x, y, z, block_length)
    return z

完整验证程序

# WHY: 需要完整的验证程序 - 验证程序提供端到端的算子功能测试，包括环境配置、数据准备、算子调用和结果校验
import logging
import argparse
import torch
try:
    import torch_npu
except ModuleNotFoundError:
    pass

import asc
import asc.runtime.config as config
import asc.lib.runtime as rt

USE_CORE_NUM = 8
BUFFER_NUM = 2
TILE_NUM = 8

logging.basicConfig(level=logging.INFO)

@asc.jit
def vadd_kernel(x: asc.GlobalAddress, y: asc.GlobalAddress, z: asc.GlobalAddress, block_length: int):
    offset = asc.get_block_idx() * block_length
    x_gm = asc.GlobalTensor()
    y_gm = asc.GlobalTensor()
    z_gm = asc.GlobalTensor()
    x_gm.set_global_buffer(x + offset, block_length)
    y_gm.set_global_buffer(y + offset, block_length)
    z_gm.set_global_buffer(z + offset, block_length)
    
    tile_length = block_length // TILE_NUM // BUFFER_NUM
    data_type = x.dtype
    buffer_size = tile_length * BUFFER_NUM * data_type.sizeof()
    
    x_local = asc.LocalTensor(data_type, asc.TPosition.VECIN, 0, tile_length * BUFFER_NUM)
    y_local = asc.LocalTensor(data_type, asc.TPosition.VECIN, buffer_size, tile_length * BUFFER_NUM)
    z_local = asc.LocalTensor(data_type, asc.TPosition.VECOUT, buffer_size + buffer_size, tile_length * BUFFER_NUM)
    
    for i in range(TILE_NUM * BUFFER_NUM):
        buf_id = i % BUFFER_NUM
        asc.data_copy(x_local[buf_id * tile_length:], x_gm[i * tile_length:], tile_length)
        asc.data_copy(y_local[buf_id * tile_length:], y_gm[i * tile_length:], tile_length)
        asc.set_flag(asc.HardEvent.MTE2_V, buf_id)
        asc.wait_flag(asc.HardEvent.MTE2_V, buf_id)
        asc.add(z_local[buf_id * tile_length:], x_local[buf_id * tile_length:],
                y_local[buf_id * tile_length:], tile_length)
        asc.set_flag(asc.HardEvent.V_MTE3, buf_id)
        asc.wait_flag(asc.HardEvent.V_MTE3, buf_id)
        asc.data_copy(z_gm[i * tile_length:], z_local[buf_id * tile_length:], tile_length)
        asc.set_flag(asc.HardEvent.MTE3_MTE2, buf_id)
        asc.wait_flag(asc.HardEvent.MTE3_MTE2, buf_id)

def vadd_launch(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
    z = torch.zeros_like(x)
    total_length = z.numel()
    block_length = total_length // USE_CORE_NUM
    vadd_kernel[USE_CORE_NUM, rt.current_stream()](x, y, z, block_length)
    return z

# WHY: 区分backend和platform - backend指定运行模式（仿真或上板），platform指定昇腾AI处理器型号
def vadd_custom(backend: config.Backend, platform: config.Platform):
    config.set_platform(backend, platform)
    # WHY: 选择device - 根据backend选择计算设备，NPU模式使用npu设备，仿真模式使用cpu设备
    device = "npu" if config.Backend(backend) == config.Backend.NPU else "cpu"
    # WHY: 创建随机输入 - 随机输入能够更全面地测试算子的数值正确性
    size = 8 * 2048
    x = torch.rand(size, dtype=torch.float32, device=device)
    y = torch.rand(size, dtype=torch.float32, device=device)
    z = vadd_launch(x, y)
    # WHY: 使用torch.allclose校验 - allclose允许浮点数误差，比精确相等比较更适合验证数值计算正确性
    assert torch.allclose(z, x + y)
    logging.info("[INFO] Assertion passed: z equals x + y")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    # WHY: 提供-r参数 - 允许用户在命令行指定运行模式，Model为仿真模式，NPU为上板模式
    parser.add_argument("-r", type=str, default="Model", help="backend to run")
    # WHY: 提供-v参数 - 允许用户在命令行指定昇腾AI处理器型号，如Ascend910B1
    parser.add_argument("-v", type=str, default=None, help="platform to run")
    args = parser.parse_args()
    backend = args.r
    platform = args.v
    if backend not in config.Backend.__members__:
        raise ValueError("Unsupported Backend! Supported: ['Model', 'NPU']")
    backend = config.Backend(backend)
    if platform is not None:
        platform_values = [p.value for p in config.Platform]
        if platform not in platform_values:
            raise ValueError(f"Unsupported Platform! Supported: {platform_values}")
        platform = config.Platform(platform)
    logging.info("[INFO] start process sample add.")
    vadd_custom(backend, platform)
    logging.info("[INFO] Sample add run success.")

编译与运行

# WHY: 需要-r参数 - RUN_MODE指定编译执行方式，Model为仿真器模式，NPU为上板模式
# 仿真器模式运行 - WHY: 无NPU设备时可使用仿真器验证算子逻辑正确性
python3 add.py -r Model -v Ascend910B1

# NPU上板模式运行 - WHY: 在真实NPU设备上验证算子的性能表现
python3 add.py -r NPU -v Ascend910B1

# WHY: 检查打屏信息 - 用例执行完成出现"Sample add run success"，说明样例执行成功，算子功能正确

Python DSL到Ascend C的编译链路

编译流程架构

pyasc的编译链路将Python DSL转换为可在昇腾NPU上执行的机器码，整个流程分为前端、中端和后端三个阶段。

前端负责解析Python源码，生成抽象语法树（AST）。pyasc使用Python的ast模块解析用户编写的算子代码，提取核函数定义、Tensor操作和数据搬运等关键信息。

中端基于MLIR框架实现，将AST转换为Ascend C中间表示（ASC-IR）。ASC-IR是一种针对昇腾NPU架构优化的中间表示，包含矢量计算、矩阵计算和内存访问等专用操作。

后端将ASC-IR转换为昇腾NPU的机器码。pyasc使用LLVM的代码生成能力，将ASC-IR转换为二进制代码，并生成算子运行时所需的元数据信息。

pyasc目录结构与编译模块

pyasc项目的目录结构设计如下：

pyasc/
├── bin/              # 工具文件
├── docs/             # 说明文档
│   ├── figures/      # 文档图片
│   └── python-api/   # API接口文档
├── include/          # 后端头文件和td文件
│   └── ascir/        # ascir头文件和td文件
├── lib/              # 后端源文件
│   ├── Dialect/      # mlir方言定义源文件
│   ├── TableGen/     # tablegen扩展代码文件
│   └── Target/       # mlir目标代码转换源文件
├── python/           # python前端代码
│   ├── asc/          # 用户可见的python包
│   ├── src/          # pybind相关代码，cpp格式
│   ├── test/         # python格式的测试用例集
│   └── tutorials/    # 供用户参考的样例集
└── test/             # 后端的测试用例集

关键编译模块说明：

• include/ascir/：定义ASC-IR的方言接口，包括Tensor操作、同步原语和内存管理接口。
• lib/Dialect/：实现ASC-IR方言的具体操作，包括矢量计算操作、矩阵计算操作和内存访问操作。
• lib/Target/：实现ASC-IR到机器码的转换逻辑，生成昇腾NPU可执行的二进制代码。
• python/asc/：提供用户可见的Python编程接口，包括@asc.jit装饰器、Tensor类和计算接口。

@asc.jit装饰器工作原理

@asc.jit装饰器是pyasc编译链路的入口，其核心工作流程如下：

1. 捕获被装饰函数的Python字节码和AST。
2. 分析函数参数类型注解，确定GlobalAddress、GlobalTensor等类型信息。
3. 将Python函数体转换为MLIR表示，生成对应的ASC-IR操作序列。
4. 触发ASC-IR到机器码的编译流程，生成核函数的二进制代码。
5. 返回可被内核调用符（如vadd_kernel[num_blocks, stream]()）调用的可执行对象。

算子注册到GE的完整流程

GE（Graph Engine）概述

GE是CANN软件栈中的图引擎，负责算子的图编译和运行时调度。自定义算子开发完成后，需要注册到GE中，才能被神经网络框架（如PyTorch、TensorFlow）调用。

算子注册步骤

算子注册到GE需要完成以下工作：

算子原型定义

算子原型定义描述算子的输入输出规格、属性参数和类型推导规则。pyasc算子注册需要使用CANN提供的算子注册接口。

# WHY: 需要算子原型定义 - GE通过算子原型了解算子的输入输出规格，完成图编译时的形状推导和类型检查
# 以下为算子原型定义的示例结构（实际注册需使用CANN提供的注册宏）
# REG_OP(Add)
#     .INPUT(x, TensorType({DT_FLOAT}))
#     .INPUT(y, TensorType({DT_FLOAT}))
#     .OUTPUT(z, TensorType({DT_FLOAT}))
#     .OP_END_FACTORY_REG(Add)

算子实现注册

算子实现注册将pyasc实现的核函数与算子原型关联起来，使GE能够调度该算子执行计算。

# WHY: 需要算子实现注册 - 将pyasc实现的核函数注册到GE的算子库中，使框架能够调用该算子
# 以下为算子实现注册的示例（实际注册需使用CANN提供的注册接口）
# 算子注册信息包括算子类型、输入输出描述和核函数入口地址

算子适配插件开发

对于接入PyTorch等深度学习框架的场景，需要开发算子适配插件，将框架的算子调用转换为GE的算子执行请求。

# WHY: 需要算子适配插件 - 深度学习框架使用自己的算子抽象，需要通过适配插件将框架算子映射到GE算子
# 以下为PyTorch算子适配的示例结构
import torch
import torch_npu

# WHY: 使用torch.library - torch.library是PyTorch提供的扩展算子注册接口，允许注册自定义算子
@torch.library.custom_op("myops::vadd", mutates_args=())
def vadd(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
    # 调用pyasc实现的算子
    return vadd_launch(x, y)

# WHY: 需要注册符号推导函数 - PyTorch需要推导算子的输出形状，确保图编译正确性
@vadd.register_symbolic
def _(x, y):
    return torch.empty_like(x)

调试方法：打印中间结果与性能分析

中间结果打印

pyasc提供多种调试手段，帮助开发者定位算子实现中的问题。

使用logging模块打印调试信息

# WHY: 使用logging而非print - logging模块提供日志级别控制，可灵活开关调试信息输出
import logging
logging.basicConfig(level=logging.DEBUG)

# 在核函数中插入调试信息打印
@asc.jit
def vadd_kernel_with_debug(x, y, z, block_length):
    offset = asc.get_block_idx() * block_length
    logging.debug(f"Block index: {asc.get_block_idx()}, offset: {offset}")
    
    # WHY: 在搬入后打印 - 确认数据正确搬入到Local Memory，是调试数据搬运问题的关键手段
    asc.data_copy(x_local[...], x_gm[...], tile_length)
    logging.debug(f"Data copied in: x_local[0] = {x_local[0].item()}")
    
    asc.add(z_local[...], x_local[...], y_local[...], tile_length)
    # WHY: 在计算后打印 - 确认计算结果正确，是调试计算逻辑问题的关键手段
    logging.debug(f"Computation result: z_local[0] = {z_local[0].item()}")

使用仿真器模式进行功能调试

仿真器模式在CPU上模拟NPU的执行行为，支持更灵活的调试手段。

# WHY: 使用仿真器模式调试 - 仿真器模式支持单步执行和变量查看，适合调试算子逻辑错误
python3 add.py -r Model -v Ascend910B1

# WHY: 配合gdb使用 - gdb可调试pyasc的后端C++代码，定位编译器或运行时的问题
# gdb --args python3 add.py -r Model -v Ascend910B1

性能分析方法

使用CANN工具链进行性能分析

CANN提供专业的性能分析工具，帮助开发者定位算子的性能瓶颈。

# WHY: 使用msprof工具 - msprof是CANN提供的性能分析命令行工具，可采集算子的执行时间、内存带宽等性能指标
# 采集算子性能数据
msprof --output=./profiling_output python3 add.py -r NPU -v Ascend910B1

# WHY: 分析profiling数据 - 通过profiling数据了解算子的计算效率，找出性能优化方向
# 查看profiling报告
cat ./profiling_output/report/summary.csv

性能优化建议

根据性能分析结果，可从以下维度优化算子性能：

1. 内存访问优化：调整数据分块大小，提升Global Memory到Local Memory的数据搬运效率。
2. 流水线优化：调整TILE_NUM和BUFFER_NUM参数，提升搬运和计算的并行度。
3. 向量化优化：使用pyasc提供的向量化接口，充分发挥NPU的SIMD计算能力。

效率对比表

下表展示使用pyasc开发Ascend C算子与传统C++开发方式的效率对比：

对比维度	传统C++开发	pyasc Python开发	差异来源
代码行数	约150行（含头文件、命名空间）	约80行（纯Python代码）	Python语法简洁，无需头文件和类型声明
编译次数	每次修改需完整重新编译C++项目	修改后自动重新编译变更部分	pyasc支持增量编译和开发者模式
调试周期	需交叉编译、部署到NPU设备	可在仿真器模式快速验证	仿真器模式避免设备部署开销
学习成本	需掌握Ascend C扩展语法和C++模板	仅需掌握Python和标准Ascend C接口	Python开发者无需学习C++扩展语法
生态集成	需手动编写Python绑定代码	原生支持PyTorch Tensor作为输入输出	pyasc内置PyTorch集成，无需额外绑定

常见问题排查

编译阶段问题

问题1：LLVM找不到或版本不匹配

错误信息：LLVM not found 或 LLVM version mismatch

排查步骤：

# WHY: 检查LLVM_INSTALL_PREFIX - pyasc构建脚本依赖该环境变量定位LLVM安装路径
echo $LLVM_INSTALL_PREFIX

# WHY: 使用llvm-config --version - 确认LLVM版本是否满足pyasc要求（>=19.1.7）
${LLVM_INSTALL_PREFIX}/bin/llvm-config --version

问题2：Python依赖缺失

错误信息：ModuleNotFoundError: No module named 'xxx'

解决方法：

# WHY: 使用pip安装缺失依赖 - pyasc的构建和运行依赖多个Python包，缺失会导致导入失败
python3 -m pip install -r requirements-build.txt
python3 -m pip install -r requirements-runtime.txt

运行阶段问题

问题3：CANN环境变量未正确设置

错误信息：ASCEND_HOME_PATH not set 或 Failed to load CANN library

解决方法：

# WHY: 需要source set_env.sh - 该脚本设置ASCEND_HOME_PATH等环境变量，pyasc运行时依赖这些变量
source /usr/local/Ascend/cann/set_env.sh

# WHY: 检查环境变量 - 确认CANN环境变量正确设置后，才能确保pyasc正常运行
echo $ASCEND_HOME_PATH

问题4：NPU设备无法访问

错误信息：npu-smi info 无输出或报错

排查步骤：

# WHY: 检查NPU驱动 - NPU驱动未正确安装会导致设备无法访问
npu-smi info

# WHY: 检查设备文件 - /dev/davinci*设备文件不存在说明NPU驱动未加载
ls -l /dev/davinci*

---

仓库地址：https://atomgit.com/cann/pyasc