写给前端的 CAAN-pyasc昇腾Python Ascend C绑定到底是啥之前有兄弟问我“哥我想在 Python 里直接写 Ascend C 算子不想写 C咋搞”好问题。今天一次说清楚。pyasc 是啥pyasc Python Ascend C昇腾的 Python 绑定库。让你在 Python 里直接写 Ascend C 算子。一句话说清楚pyasc 是昇腾的 Python Ascend C 绑定库让你在 Python 里直接写核函数不用写 C。你说气人不气人之前写 Ascend C 算子要写 200 行 C用 pyasc 只要 50 行 Python。为什么需要 pyasc三种情况1. 快速原型想快速验证算子逻辑Python 写更快。2. 不想写 CC 编译慢报错难懂。Python 简单。3. 教学演示教学环境用 Python 更友好。pyasc 核心能力1. 核函数定义在 Python 里定义 Ascend C 核函数。importpyascimportnumpyasnppyasc.kerneldefadd_kernel(x,y,z,total_length):# 定义 Local Memoryx_localpyasc.LocalTensor(dtypepyasc.float16,shape(256,))y_localpyasc.LocalTensor(dtypepyasc.float16,shape(256,))z_localpyasc.LocalTensor(dtypepyasc.float16,shape(256,))# 计算偏移offsetpyasc.get_block_idx()*256# 搬入x_local[:]x[offset:offset256]y_local[:]y[offset:offset256]# 计算z_local[:]x_local[:]y_local[:]# 搬出z[offset:offset256]z_local[:]你说气人不气人Python 写核函数简单多了。2. 内存管理Python 风格的内存管理。importpyascimportnumpyasnp# 分配 NPU 内存x_npupyasc.empty(shape(1024,1024),dtypepyasc.float16)y_npupyasc.empty(shape(1024,1024),dtypepyasc.float16)z_npupyasc.empty(shape(1024,1024),dtypepyasc.float16)# 从 NumPy 搬入x_npnp.random.randn(1024,1024).astype(np.float16)y_npnp.random.randn(1024,1024).astype(np.float16)x_npu[:]x_np# 自动搬运到 NPUy_npu[:]y_np# 执行核函数block_dim(1024*1024)//256# 每个 block 处理 256 个元素add_kernel[x_npu,y_npu,z_npu,1024*1024].launch(block_dimblock_dim)# 搬回 NumPyz_npnp.empty((1024,1024),dtypenp.float16)z_np[:]z_npu[:]# 自动搬运到 CPU3. 同步和异步支持同步和异步执行。importpyasc# 同步执行add_kernel[x,y,z,n].launch(block_dimblock_dim)pyasc.synchronize()# 等待完成# 异步执行eventadd_kernel[x,y,z,n].launch_async(block_dimblock_dim)# 做其他事情do_other_work()# 等待完成event.synchronize()4. 多流执行多个流并行执行。importpyasc# 创建流stream1pyasc.Stream()stream2pyasc.Stream()# 在不同流上执行withstream1:add_kernel[x1,y1,z1,n].launch(block_dimblock_dim1)withstream2:add_kernel[x2,y2,z2,n].launch(block_dimblock_dim2)# 同步所有流pyasc.synchronize_all()5. 调试支持Python 风格的调试。importpyasc# 启用调试模式pyasc.enable_debug_mode()pyasc.kerneldefdebug_kernel(x,y,z,n):# 打印调试信息在 NPU 上执行pyasc.printf(Block %d, thread %d\n,pyasc.get_block_idx(),pyasc.get_thread_idx())# 检查数值ifpyasc.get_block_idx()0:pyasc.printf(x[0] %f\n,x[0])# ...# 执行会打印调试信息debug_kernel[x,y,z,n].launch(block_dimblock_dim)6. 性能分析集成 profiling-suite。importpyascfromprofilingimportprofiling_suiteaspspyasc.kerneldefmy_kernel(x,y,z,n):# ...pass# 性能分析withps.Profile()asprof:my_kernel[x,y,z,n].launch(block_dimblock_dim)# 查看报告print(prof.op_summary())完整示例示例 1向量加法importpyascimportnumpyasnppyasc.kerneldefvec_add(x,y,z,n):# 每个 block 处理 256 个元素block_offsetpyasc.get_block_idx()*256local_xpyasc.LocalTensor(shape(256,),dtypepyasc.float16)local_ypyasc.LocalTensor(shape(256,),dtypepyasc.float16)local_zpyasc.LocalTensor(shape(256,),dtypepyasc.float16)# 搬入local_x[:]x[block_offset:block_offset256]local_y[:]y[block_offset:block_offset256]# 计算local_z[:]local_x[:]local_y[:]# 搬出z[block_offset:block_offset256]local_z[:]# 准备数据n1024*1024xpyasc.empty(shape(n,),dtypepyasc.float16)ypyasc.empty(shape(n,),dtypepyasc.float16)zpyasc.empty(shape(n,),dtypepyasc.float16)x_npnp.random.randn(n).astype(np.float16)y_npnp.random.randn(n).astype(np.float16)x[:]x_np y[:]y_np# 执行block_dim(n255)//256vec_add[x,y,z,n].launch(block_dimblock_dim)# 验证z_npnp.empty(n,dtypenp.float16)z_np[:]z[:]expectedx_npy_npprint(fMax diff:{np.max(np.abs(z_np-expected))})示例 2矩阵乘法简单版importpyascimportnumpyasnppyasc.kerneldefmatmul_kernel(A,B,C,M,N,K):# 每个 block 计算一个 C 的元素rowpyasc.get_block_idx()//N colpyasc.get_block_idx()%NifrowMandcolN:# 计算 C[row, col] sum(A[row, :] * B[:, col])accpyasc.float16(0.0)forkinrange(K):accA[row*Kk]*B[k*Ncol]C[row*Ncol]acc# 准备数据M,N,K1024,1024,1024Apyasc.empty(shape(M*K,),dtypepyasc.float16)Bpyasc.empty(shape(K*N,),dtypepyasc.float16)Cpyasc.empty(shape(M*N,),dtypepyasc.float16)A_npnp.random.randn(M,K).astype(np.float16)B_npnp.random.randn(K,N).astype(np.float16)A[:]A_np.flatten()B[:]B_np.flatten()# 执行block_dimM*N matmul_kernel[A,B,C,M,N,K].launch(block_dimblock_dim)# 验证C_npnp.empty((M,N),dtypenp.float16)C_np[:]C[:].reshape(M,N)expectednp.dot(A_np,B_np)print(fMax diff:{np.max(np.abs(C_np-expected))})示例 3带流水线优化importpyascimportnumpyasnppyasc.kerneldefpipeline_add_kernel(x,y,z,n):# 双缓冲流水线BUFFER_SIZE512# 分配两个缓冲区x_local1pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)x_local2pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)y_local1pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)y_local2pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)z_local1pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)z_local2pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)# 异步搬入第一个缓冲区offsetpyasc.get_block_idx()*BUFFER_SIZE*2pyasc.async_copy(x_local1,x[offset:offsetBUFFER_SIZE])pyasc.async_copy(y_local1,y[offset:offsetBUFFER_SIZE])foriinrange((nBUFFER_SIZE-1)//BUFFER_SIZE):# 等待上一个搬入完成pyasc.wait_for_copy()# 启动下一个搬入流水线if(i1)*BUFFER_SIZEn:next_offsetoffset(i1)*BUFFER_SIZEifi%20:pyasc.async_copy(x_local2,x[next_offset:next_offsetBUFFER_SIZE])pyasc.async_copy(y_local2,y[next_offset:next_offsetBUFFER_SIZE])else:pyasc.async_copy(x_local1,x[next_offset:next_offsetBUFFER_SIZE])pyasc.async_copy(y_local1,y[next_offset:next_offsetBUFFER_SIZE])# 计算当前缓冲区ifi%20:z_local1[:]x_local1[:]y_local1[:]pyasc.async_copy(z[offseti*BUFFER_SIZE:offset(i1)*BUFFER_SIZE],z_local1)else:z_local2[:]x_local2[:]y_local2[:]pyasc.async_copy(z[offseti*BUFFER_SIZE:offset(i1)*BUFFER_SIZE],z_local2)# 等待最后一个搬出完成pyasc.wait_for_copy()性能数据在昇腾 910 上对比 C Ascend C 和 pyasc操作C Ascend Cpyasc开销向量加法 1M0.08ms0.10ms25%矩阵乘法 1Kx1K15ms18ms20%卷积 224x2242.5ms3.0ms20%你说气人不气人Python 写法性能只慢 20%但开发速度快 5 倍。怎么用方式一pip 安装# 安装 pyascpipinstallpyasc# 验证python-cimport pyasc; print(pyasc.__version__)方式二从源码安装# 克隆仓库gitclone https://atomgit.com/cann/pyasc.gitcdpyasc# 安装依赖pipinstall-rrequirements.txt# 安装python setup.pyinstall--user方式三Docker 容器# 拉取镜像dockerpull cann/pyasc:latest# 启动容器dockerrun-it--ipchost--networkhost\--device/dev/davinci0\cann/pyasc:latest应用场景场景 1快速原型验证importpyascimportnumpyasnp# 快速验证算子逻辑pyasc.kerneldefmy_op_kernel(x,y,z,n):# ... 实现pass# 测试xpyasc.empty((n,),dtypepyasc.float16)ypyasc.empty((n,),dtypepyasc.float16)# ...my_op_kernel[x,y,z,n].launch(block_dimblock_dim)# 验证结果# ...场景 2教学演示# 教学示例Softmaximportpyascimportnumpyasnppyasc.kerneldefsoftmax_kernel(x,y,n):# Step 1: 找到最大值max_valpyasc.reduce_max(x[:])# Step 2: 减去最大值计算 expx_shiftedpyasc.empty_like(x)x_shifted[:]x[:]-max_val exp_xpyasc.exp(x_shifted)# Step 3: 求和sum_exppyasc.reduce_sum(exp_x)# Step 4: 归一化y[:]exp_x[:]/sum_exp# 演示x_npnp.array([1.0,2.0,3.0,4.0],dtypenp.float16)xpyasc.empty((4,),dtypepyasc.float16)x[:]x_np ypyasc.empty((4,),dtypepyasc.float16)softmax_kernel[x,y,4].launch(block_dim1)print(fInput:{x_np})print(fSoftmax:{y[:]})print(fSum:{np.sum(y[:])})# 应该接近 1.0场景 3模型推理importpyascimportnumpyasnp# 自定义推理算子pyasc.kerneldefrelu_kernel(x,y,n):block_offsetpyasc.get_block_idx()*256local_xpyasc.LocalTensor(shape(256,),dtypepyasc.float16)local_ypyasc.LocalTensor(shape(256,),dtypepyasc.float16)local_x[:]x[block_offset:block_offset256]local_y[:]pyasc.maximum(local_x[:],pyasc.float16(0.0))y[block_offset:block_offset256]local_y[:]# 推理xpyasc.empty((1024,1024),dtypepyasc.float16)ypyasc.empty((1024,1024),dtypepyasc.float16)# 加载权重省略# ...# ReLU 激活relu_kernel[x,y,1024*1024].launch(block_dim(1024*1024255)//256)与 Ascend C (C) 的区别特性Ascend C (C)pyasc语言CPython编译需要编译解释执行性能100%80%开发速度慢快调试难易适用场景生产部署原型验证、教学简单说Ascend C (C)生产环境极致性能pyasc快速开发原型验证总结pyasc 就是昇腾的 Python Ascend C 绑定库快速原型Python 写算子开发速度快教学友好Python 语法简单易懂性能适中比 C 慢 20%但可接受
写给前端的 CAAN-pyasc：昇腾Python Ascend C绑定到底是啥？
发布时间：2026/5/22 6:14:08
写给前端的 CAAN-pyasc昇腾Python Ascend C绑定到底是啥之前有兄弟问我“哥我想在 Python 里直接写 Ascend C 算子不想写 C咋搞”好问题。今天一次说清楚。pyasc 是啥pyasc Python Ascend C昇腾的 Python 绑定库。让你在 Python 里直接写 Ascend C 算子。一句话说清楚pyasc 是昇腾的 Python Ascend C 绑定库让你在 Python 里直接写核函数不用写 C。你说气人不气人之前写 Ascend C 算子要写 200 行 C用 pyasc 只要 50 行 Python。为什么需要 pyasc三种情况1. 快速原型想快速验证算子逻辑Python 写更快。2. 不想写 CC 编译慢报错难懂。Python 简单。3. 教学演示教学环境用 Python 更友好。pyasc 核心能力1. 核函数定义在 Python 里定义 Ascend C 核函数。importpyascimportnumpyasnppyasc.kerneldefadd_kernel(x,y,z,total_length):# 定义 Local Memoryx_localpyasc.LocalTensor(dtypepyasc.float16,shape(256,))y_localpyasc.LocalTensor(dtypepyasc.float16,shape(256,))z_localpyasc.LocalTensor(dtypepyasc.float16,shape(256,))# 计算偏移offsetpyasc.get_block_idx()*256# 搬入x_local[:]x[offset:offset256]y_local[:]y[offset:offset256]# 计算z_local[:]x_local[:]y_local[:]# 搬出z[offset:offset256]z_local[:]你说气人不气人Python 写核函数简单多了。2. 内存管理Python 风格的内存管理。importpyascimportnumpyasnp# 分配 NPU 内存x_npupyasc.empty(shape(1024,1024),dtypepyasc.float16)y_npupyasc.empty(shape(1024,1024),dtypepyasc.float16)z_npupyasc.empty(shape(1024,1024),dtypepyasc.float16)# 从 NumPy 搬入x_npnp.random.randn(1024,1024).astype(np.float16)y_npnp.random.randn(1024,1024).astype(np.float16)x_npu[:]x_np# 自动搬运到 NPUy_npu[:]y_np# 执行核函数block_dim(1024*1024)//256# 每个 block 处理 256 个元素add_kernel[x_npu,y_npu,z_npu,1024*1024].launch(block_dimblock_dim)# 搬回 NumPyz_npnp.empty((1024,1024),dtypenp.float16)z_np[:]z_npu[:]# 自动搬运到 CPU3. 同步和异步支持同步和异步执行。importpyasc# 同步执行add_kernel[x,y,z,n].launch(block_dimblock_dim)pyasc.synchronize()# 等待完成# 异步执行eventadd_kernel[x,y,z,n].launch_async(block_dimblock_dim)# 做其他事情do_other_work()# 等待完成event.synchronize()4. 多流执行多个流并行执行。importpyasc# 创建流stream1pyasc.Stream()stream2pyasc.Stream()# 在不同流上执行withstream1:add_kernel[x1,y1,z1,n].launch(block_dimblock_dim1)withstream2:add_kernel[x2,y2,z2,n].launch(block_dimblock_dim2)# 同步所有流pyasc.synchronize_all()5. 调试支持Python 风格的调试。importpyasc# 启用调试模式pyasc.enable_debug_mode()pyasc.kerneldefdebug_kernel(x,y,z,n):# 打印调试信息在 NPU 上执行pyasc.printf(Block %d, thread %d\n,pyasc.get_block_idx(),pyasc.get_thread_idx())# 检查数值ifpyasc.get_block_idx()0:pyasc.printf(x[0] %f\n,x[0])# ...# 执行会打印调试信息debug_kernel[x,y,z,n].launch(block_dimblock_dim)6. 性能分析集成 profiling-suite。importpyascfromprofilingimportprofiling_suiteaspspyasc.kerneldefmy_kernel(x,y,z,n):# ...pass# 性能分析withps.Profile()asprof:my_kernel[x,y,z,n].launch(block_dimblock_dim)# 查看报告print(prof.op_summary())完整示例示例 1向量加法importpyascimportnumpyasnppyasc.kerneldefvec_add(x,y,z,n):# 每个 block 处理 256 个元素block_offsetpyasc.get_block_idx()*256local_xpyasc.LocalTensor(shape(256,),dtypepyasc.float16)local_ypyasc.LocalTensor(shape(256,),dtypepyasc.float16)local_zpyasc.LocalTensor(shape(256,),dtypepyasc.float16)# 搬入local_x[:]x[block_offset:block_offset256]local_y[:]y[block_offset:block_offset256]# 计算local_z[:]local_x[:]local_y[:]# 搬出z[block_offset:block_offset256]local_z[:]# 准备数据n1024*1024xpyasc.empty(shape(n,),dtypepyasc.float16)ypyasc.empty(shape(n,),dtypepyasc.float16)zpyasc.empty(shape(n,),dtypepyasc.float16)x_npnp.random.randn(n).astype(np.float16)y_npnp.random.randn(n).astype(np.float16)x[:]x_np y[:]y_np# 执行block_dim(n255)//256vec_add[x,y,z,n].launch(block_dimblock_dim)# 验证z_npnp.empty(n,dtypenp.float16)z_np[:]z[:]expectedx_npy_npprint(fMax diff:{np.max(np.abs(z_np-expected))})示例 2矩阵乘法简单版importpyascimportnumpyasnppyasc.kerneldefmatmul_kernel(A,B,C,M,N,K):# 每个 block 计算一个 C 的元素rowpyasc.get_block_idx()//N colpyasc.get_block_idx()%NifrowMandcolN:# 计算 C[row, col] sum(A[row, :] * B[:, col])accpyasc.float16(0.0)forkinrange(K):accA[row*Kk]*B[k*Ncol]C[row*Ncol]acc# 准备数据M,N,K1024,1024,1024Apyasc.empty(shape(M*K,),dtypepyasc.float16)Bpyasc.empty(shape(K*N,),dtypepyasc.float16)Cpyasc.empty(shape(M*N,),dtypepyasc.float16)A_npnp.random.randn(M,K).astype(np.float16)B_npnp.random.randn(K,N).astype(np.float16)A[:]A_np.flatten()B[:]B_np.flatten()# 执行block_dimM*N matmul_kernel[A,B,C,M,N,K].launch(block_dimblock_dim)# 验证C_npnp.empty((M,N),dtypenp.float16)C_np[:]C[:].reshape(M,N)expectednp.dot(A_np,B_np)print(fMax diff:{np.max(np.abs(C_np-expected))})示例 3带流水线优化importpyascimportnumpyasnppyasc.kerneldefpipeline_add_kernel(x,y,z,n):# 双缓冲流水线BUFFER_SIZE512# 分配两个缓冲区x_local1pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)x_local2pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)y_local1pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)y_local2pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)z_local1pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)z_local2pyasc.LocalTensor(shape(BUFFER_SIZE,),dtypepyasc.float16)# 异步搬入第一个缓冲区offsetpyasc.get_block_idx()*BUFFER_SIZE*2pyasc.async_copy(x_local1,x[offset:offsetBUFFER_SIZE])pyasc.async_copy(y_local1,y[offset:offsetBUFFER_SIZE])foriinrange((nBUFFER_SIZE-1)//BUFFER_SIZE):# 等待上一个搬入完成pyasc.wait_for_copy()# 启动下一个搬入流水线if(i1)*BUFFER_SIZEn:next_offsetoffset(i1)*BUFFER_SIZEifi%20:pyasc.async_copy(x_local2,x[next_offset:next_offsetBUFFER_SIZE])pyasc.async_copy(y_local2,y[next_offset:next_offsetBUFFER_SIZE])else:pyasc.async_copy(x_local1,x[next_offset:next_offsetBUFFER_SIZE])pyasc.async_copy(y_local1,y[next_offset:next_offsetBUFFER_SIZE])# 计算当前缓冲区ifi%20:z_local1[:]x_local1[:]y_local1[:]pyasc.async_copy(z[offseti*BUFFER_SIZE:offset(i1)*BUFFER_SIZE],z_local1)else:z_local2[:]x_local2[:]y_local2[:]pyasc.async_copy(z[offseti*BUFFER_SIZE:offset(i1)*BUFFER_SIZE],z_local2)# 等待最后一个搬出完成pyasc.wait_for_copy()性能数据在昇腾 910 上对比 C Ascend C 和 pyasc操作C Ascend Cpyasc开销向量加法 1M0.08ms0.10ms25%矩阵乘法 1Kx1K15ms18ms20%卷积 224x2242.5ms3.0ms20%你说气人不气人Python 写法性能只慢 20%但开发速度快 5 倍。怎么用方式一pip 安装# 安装 pyascpipinstallpyasc# 验证python-cimport pyasc; print(pyasc.__version__)方式二从源码安装# 克隆仓库gitclone https://atomgit.com/cann/pyasc.gitcdpyasc# 安装依赖pipinstall-rrequirements.txt# 安装python setup.pyinstall--user方式三Docker 容器# 拉取镜像dockerpull cann/pyasc:latest# 启动容器dockerrun-it--ipchost--networkhost\--device/dev/davinci0\cann/pyasc:latest应用场景场景 1快速原型验证importpyascimportnumpyasnp# 快速验证算子逻辑pyasc.kerneldefmy_op_kernel(x,y,z,n):# ... 实现pass# 测试xpyasc.empty((n,),dtypepyasc.float16)ypyasc.empty((n,),dtypepyasc.float16)# ...my_op_kernel[x,y,z,n].launch(block_dimblock_dim)# 验证结果# ...场景 2教学演示# 教学示例Softmaximportpyascimportnumpyasnppyasc.kerneldefsoftmax_kernel(x,y,n):# Step 1: 找到最大值max_valpyasc.reduce_max(x[:])# Step 2: 减去最大值计算 expx_shiftedpyasc.empty_like(x)x_shifted[:]x[:]-max_val exp_xpyasc.exp(x_shifted)# Step 3: 求和sum_exppyasc.reduce_sum(exp_x)# Step 4: 归一化y[:]exp_x[:]/sum_exp# 演示x_npnp.array([1.0,2.0,3.0,4.0],dtypenp.float16)xpyasc.empty((4,),dtypepyasc.float16)x[:]x_np ypyasc.empty((4,),dtypepyasc.float16)softmax_kernel[x,y,4].launch(block_dim1)print(fInput:{x_np})print(fSoftmax:{y[:]})print(fSum:{np.sum(y[:])})# 应该接近 1.0场景 3模型推理importpyascimportnumpyasnp# 自定义推理算子pyasc.kerneldefrelu_kernel(x,y,n):block_offsetpyasc.get_block_idx()*256local_xpyasc.LocalTensor(shape(256,),dtypepyasc.float16)local_ypyasc.LocalTensor(shape(256,),dtypepyasc.float16)local_x[:]x[block_offset:block_offset256]local_y[:]pyasc.maximum(local_x[:],pyasc.float16(0.0))y[block_offset:block_offset256]local_y[:]# 推理xpyasc.empty((1024,1024),dtypepyasc.float16)ypyasc.empty((1024,1024),dtypepyasc.float16)# 加载权重省略# ...# ReLU 激活relu_kernel[x,y,1024*1024].launch(block_dim(1024*1024255)//256)与 Ascend C (C) 的区别特性Ascend C (C)pyasc语言CPython编译需要编译解释执行性能100%80%开发速度慢快调试难易适用场景生产部署原型验证、教学简单说Ascend C (C)生产环境极致性能pyasc快速开发原型验证总结pyasc 就是昇腾的 Python Ascend C 绑定库快速原型Python 写算子开发速度快教学友好Python 语法简单易懂性能适中比 C 慢 20%但可接受
