pyasc 让你用 Python 写 Ascend C 算子——但需要手动写 C++ kernel。pypto 更进一步：直接把 PTO（虚拟指令集）封装成 Python API，在 Python 里写「指令级」程序，然后编译到 NPU 上执行。定位是「用 Python 语法写汇编」。

PTO 虚拟指令集回顾

pto-isa 是 CANN 的虚拟指令集架构——Ascend C kernel 先编译成 PTO 指令，再翻译成 NPU 固件指令。pypto 让你绕过 Ascend C，直接在 Python 里写 PTO 指令序列。

三层编译（正常路径）
Ascend C → PTO 指令 → NPU 固件指令

pypto 路径（绕过 Ascend C）
Python PTO API → PTO 指令 → NPU 固件指令

绕过 Ascend C 的好处：可以对每一条 PTO 指令做精细控制（调度流水线、寄存器分配、L1 缓存策略），而这些在 Ascend C 里是编译器自动决定的。

pypto 的基本用法

# pypto/examples/vector_add.py

import pypto as pto
import numpy as np

# 定义 kernel（用 PTO 指令）
# 不需要写 C++——Python 函数直接对应 PTO kernel
@pto.kernel
def vector_add(a, b, o, n):
    # PTO 的并行模型：256 个 lane 同时执行
    # get_global_id() 获取当前 lane 的全局编号
    i = pto.get_global_id()

    # 边界检查（类似 CUDA 的 if (i < n）)
    if i < n:
        # PTO 的 LOAD 指令：从 HBM 加载数据到 L1
        a_val = pto.load(a + i)
        b_val = pto.load(b + i)

        # PTO 的 FMA 指令（fused multiply-add）
        # 一条指令完成 o[i] = a[i] + b[i]
        # 不需要单独 add 和 store
        pto.fma(o + i, a_val, b_val, 0.0)

# 准备数据
n = 1024
a = np.random.rand(n).astype(np.float32)
b = np.random.rand(n).astype(np.float32)
o = np.zeros(n, dtype=np.float32)

# 分配 NPU 内存
d_a = pto.alloc_tensor(n * 4)
d_b = pto.alloc_tensor(n * 4)
d_o = pto.alloc_tensor(n * 4)

pto.memcpy_h2d(d_a, a)
pto.memcpy_h2d(d_b, b)

# 启动 kernel（类似 CUDA 的 <<<grid, block>>>）
# 256 个 lane 一组，1024 个元素需要 4 组
pto.launch(vector_add, grid=(4, 1, 1), block=(256, 1, 1),
          args=(d_a, d_b, d_o, n))

pto.memcpy_d2h(o, d_o)

# 验证
assert np.allclose(o, a + b)
print(f"pypto: vector_add {n} elements passed")

核心差异：PTO 指令是显式调度——load/fma/store 每条指令的顺序决定了流水线行为。Ascend C 是隐式调度——编译器自动插入 PipeBarrier。

手动调度流水线

Ascend C 里流水线调度是编译器自动做的。pypto 里需要手动写——因为 Python 可以直接控制每条指令的发射时机。

# pypto/examples/pipeline_matmul.py

@pto.kernel
def matmul_tile(A, B, C, M, N, K, tile_m, tile_n, tile_k):
    # 手动双缓冲流水线
    # 用 pto.pipline 声明流水线阶段
    with pto.pipeline() as pl:
        # 阶段 1：异步加载 A 的 tile
        with pl.stage("load_a"):
            A_tile = pto.alloc_local(tile_m * tile_k)
            pto.async_load(A_tile, A + offset_a, tile_m * tile_k * 4)

        # 阶段 2：异步加载 B 的 tile（和阶段 1 并行）
        with pl.stage("load_b", after="load_a"):
            B_tile = pto.alloc_local(tile_k * tile_n)
            pto.async_load(B_tile, B + offset_b, tile_k * tile_n * 4)

        # 阶段 3：等待加载完成，执行矩阵乘
        with pl.stage("compute", after="load_b"):
            pto.wait_all()  # 等 load_a 和 load_b 完成
            C_tile = pto.alloc_local(tile_m * tile_n)
            # MMA 指令：矩阵乘累加
            pto.mma(C_tile, A_tile, B_tile, tile_m, tile_n, tile_k)

        # 阶段 4：写回结果
        with pl.stage("store", after="compute"):
            pto.store(C + offset_c, C_tile, tile_m * tile_n * 4)

    # 流水线启动：load_a → load_b → compute → store 自动 overlap
    pl.run()

手动调度的收益：双缓冲（load 和 compute 并行）+ 指令级并行（MMA 和 store 并行）。在 GEMM 这种计算密集的算子上，手动流水线比编译器自动调度快 10-15%。

调试 PTO 指令序列

pypto 的调试比 Ascend C 更细粒度——可以单步执行 PTO 指令、查看每条指令的 L1 缓存状态、模拟指令发射时序。

# pypto 的指令级调试器

import pypto.debug as dbg

# 把 kernel 加载到调试器
ctx = dbg.debug_kernel(vector_add, args=(d_a, d_b, d_o, n))

# 单步执行 PTO 指令
ctx.step()  # 执行 1 条 PTO 指令
# → 输出：
#   [Lane 0] LOAD d_a[0] → L1[0] (4 bytes)
#   [Lane 1] LOAD d_a[1] → L1[1] (4 bytes)
#   ...

# 查看 L1 缓存内容
l1_data = ctx.inspect_l1(0, 16)  # lane 0 的前 16 个 float
print(l1_data)
# → [0.123, 0.456, ...]

# 查看指令时序（哪条指令在哪 cycle 执行）
timeline = ctx.get_instruction_timeline()
for entry in timeline:
    print(f"Cycle {entry.cycle}: Lane {entry.lane}: {entry.inst}")
# → Cycle 0: Lane 0-255: LOAD
# → Cycle 3: Lane 0-255: FMA
# → Cycle 5: Lane 0-255: STORE

指令级调试在优化算子延迟时非常有用——可以看到哪条指令成了流水线气泡（bubble）。

踩坑一：PTO 指令的延迟不匹配 NPU 实际延迟

pypto 模拟器里的指令延迟是查表得到的（PTO 指令 → 预估 Cycle 数）。但 NPU 固件会把多条 PTO 指令融合成一条固件指令——实际延迟比模拟器显示的短。

错误：根据 pypto 模拟器的指令时序做优化决策。

# 模拟器显示 MMA 需要 8 cycles
# 实际 NPU 固件把 MMA + 前一条 LOAD 融合了 → 实际 5 cycles
# 基于模拟器的优化（在 MMA 前插入 NOP 对齐）反而让实际性能变差

正确做法：用 pto.profile_on_npu() 在真实 NPU 上跑性能分析。

# 真实 NPU 上的指令级 profiling
prof = pto.profile_on_npu(vector_add, args=(d_a, d_b, d_o, n))
print(prof.cycle_breakdown())
# → LOAD: 3 cycles (not 4 as simulator said)
# → FMA: 5 cycles (not 8 as simulator said)
# → STORE: 2 cycles (not 3 as simulator said)

踩坑二：Python 侧的 tensor 生命周期管理

pypto 的 pto.alloc_tensor() 在 NPU 上分配内存。但 Python 的垃圾回收不知道这块内存在被 NPU kernel 使用——Python 侧 del 了 tensor，NPU 上还在跑，导致段错误。

错误：

d_a = pto.alloc_tensor(n * 4)
pto.memcpy_h2d(d_a, a)

pto.launch(kernel, args=(d_a, ...))

del d_a  # Python 侧释放 → NPU 上的内存可能被回收
# kernel 还在跑 → 访问已释放内存 → Segmentation fault

正确做法：用 pto.Tensor 的上下文管理器，或者显式调用 pto.sync() 等 kernel 跑完。

with pto.Tensor(n * 4) as d_a:
    pto.memcpy_h2d(d_a, a)
    pto.launch(kernel, args=(d_a, ...))
# 退出 with 块时自动等 kernel 完成 + 释放 NPU 内存

# 或者显式同步
d_a = pto.alloc_tensor(n * 4)
pto.memcpy_h2d(d_a, a)
pto.launch(kernel, args=(d_a, ...))
pto.sync()  # 等 NPU kernel 完成
del d_a  # 安全释放

踩坑三：block 内同步语义和 CUDA 不同

CUDA 的 __syncthreads() 同步整个 block（所有 thread）。pypto 的 pto.barrier() 只同步当前 warp（32 个 lane）——因为 NPU 的调度单位是 warp，不是整个 block。

错误：用 pto.barrier() 同步所有 256 个 lane。

# 错误假设：barrier() 同步 256 个 lane
if pto.get_global_id() < 128:
    pto.store(shared_mem, data)
pto.barrier()  # 只同步当前 warp（32 个 lane）！
# 后 128 个 lane 可能还没执行 store → 数据不一致

正确做法：用 pto.barrier_block() 同步整个 block。

if pto.get_global_id() < 128:
    pto.store(shared_mem, data)
pto.barrier_block()  # 同步整个 block（所有 warp）
# 安全：所有 lane 的 store 都完成了

---

pypto 的价值不在日常算子开发（Ascend C 已经够用了），而在需要指令级控制的场景：算子融合的流水线优化、L1 缓存的精细管理、NPU 新特性的快速验证。用 Python 写汇编看起来奇怪，但 PTO 指令的数量很少（~50 条），Python 的表达力足够描述所有调度决策。