【昇腾CANN】cann-recipes-infer：大模型推理部署实战手册

前言

上个月帮一个创业公司部署大模型推理服务，从零开始配环境、调性能、压测，整整搞了一周。后来发现cann-recipes-infer这个仓库，里面全是实战配方（recipes），直接照着做就能跑通。这篇文章就来讲讲这个仓库的使用方法。

一、仓库定位与核心价值

cann-recipes-infer是昇腾CANN社区提供的大模型推理部署配方仓库。它不教你理论知识，直接给你可复现的实战步骤，让你从零把一个开源大模型部署到昇腾NPU上，还能跑到可用的性能。

按照仓库的README，它的核心定位是：

• 提供主流开源大模型在昇腾NPU上的推理部署配方
• 覆盖环境配置、模型转换、性能调优、压测全链路
• 提供可复现的脚本和配置文件
• 帮助开发者快速验证昇腾NPU的推理能力

这个仓库在CANN五层架构中横跨第二层（计算服务层）到第四层（计算执行层），因为它涉及算子调用（AOL）、图编译（GE）、运行时管理（Runtime）等多个环节。

仓库地址：https://atomgit.com/cann/cann-recipes-infer

二、核心配方解析

配方1：PyTorch模型迁移推理

这是最常用的配方，把你在GPU上训练好的PyTorch模型迁到昇腾NPU上做推理。

完整步骤：

1. 环境准备

# 1. 安装CANN社区版（已经包含Runtime、算子库等）
# 去昇腾社区下载对应版本，比如CANN 8.0

# 2. 安装PyTorch适配层
pip install torch-npu  # 昇腾定制的PyTorch版本

# 3. 验证安装
python -c "import torch; print('NPU可用:', torch.npu.is_available())"
# 应该输出：NPU可用: True

2. 模型转换

import torch
import torch.nn as nn

# 1. 加载在GPU上训练好的模型（假设是GPT-2）
model = GPT2LMHeadModel.from_pretrained("gpt2")
model.eval()  # 推理模式

# 2. 把模型迁到NPU
model = model.npu()

# 3. 保存为NPU格式（方便后续加载）
torch.save(model.state_dict(), "gpt2_npu.pth")

print("模型已转换并保存到 gpt2_npu.pth")

3. 推理脚本

import torch
from transformers import GPT2Tokenizer

# 1. 加载模型和分词器
model = GPT2LMHeadModel.from_pretrained("gpt2")
model.load_state_dict(torch.load("gpt2_npu.pth"))
model = model.npu()
model.eval()

tokenizer = GPT2Tokenizer.from_pretrained("gpt2")

# 2. 推理函数
def generate_text(prompt, max_length=50):
    input_ids = tokenizer.encode(prompt, return_tensors="pt").npu()
    
    with torch.no_grad():  # 不计算梯度，节省显存
        output_ids = model.generate(
            input_ids,
            max_length=max_length,
            num_beams=5,  # beam search
            early_stopping=True
        )
    
    return tokenizer.decode(output_ids[0], skip_special_tokens=True)

# 3. 测试推理
prompt = "人工智能的未来是"
generated_text = generate_text(prompt, max_length=100)
print("生成文本:", generated_text)

4. 性能调优

import torch
import time

# 1. 启用算子融合（提升性能）
torch_npu.set_compile_option("fusion_level", 2)

# 2. 启用KV缓存（减少重复计算）
model.config.use_cache = True

# 3. 性能测试
prompt = "人工智能的未来是"
input_ids = tokenizer.encode(prompt, return_tensors="pt").npu()

# 预热（JIT编译需要一点时间）
with torch.no_grad():
    _ = model.generate(input_ids, max_length=50)

# 正式测试
torch.npu.synchronize()
start = time.perf_counter()

with torch.no_grad():
    output_ids = model.generate(input_ids, max_length=100)

torch.npu.synchronize()
elapsed = time.perf_counter() - start

print("推理耗时: {:.2f} ms".format(elapsed * 1000))
print("生成tokens数: {}".format(output_ids.shape[1]))
print("吞吐: {:.2f} tokens/s".format(
    (output_ids.shape[1] - input_ids.shape[1]) / elapsed
))

配方2：FasterTransformer加速推理

对于大模型推理，FasterTransformer（FT）是个很流行的加速库。cann-recipes-infer提供了FT在昇腾NPU上的适配版本。

完整步骤：

1. 编译安装FT

# 1. 克隆FT仓库（昇腾适配版）
git clone https://atomgit.com/cann/cann-recipes-infer.git
cd cann-recipes-infer/faster_transformer

# 2. 编译（需要CMake 3.18+）
mkdir build && cd build
cmake .. -DSM=80  # SM=80对应Ascend 910
make -j32

# 3. 安装Python包
pip install -e .

2. 使用FT推理

import torch
from faster_transformer import GlmModel

# 1. 加载模型（FT格式）
model = GlmModel.from_pretrained(
    "glm-10b",
    device="npu:0"
)

# 2. 推理
prompt = "人工智能的未来是"
input_ids = tokenizer.encode(prompt, return_tensors="pt").npu()

output_ids = model.generate(
    input_ids,
    max_length=100,
    top_k=50,
    top_p=0.95
)

print("生成文本:", tokenizer.decode(output_ids[0]))

3. 性能对比

import time

# 1. 原生PyTorch推理
model_pt = GPT2LMHeadModel.from_pretrained("gpt2").npu()
input_ids = tokenizer.encode("测试", return_tensors="pt").npu()

torch.npu.synchronize()
start = time.perf_counter()
_ = model_pt.generate(input_ids, max_length=100)
torch.npu.synchronize()
pt_time = time.perf_counter() - start

# 2. FT推理
model_ft = GlmModel.from_pretrained("gpt2", device="npu:0")
input_ids = tokenizer.encode("测试", return_tensors="pt").npu()

torch.npu.synchronize()
start = time.perf_counter()
_ = model_ft.generate(input_ids, max_length=100)
torch.npu.synchronize()
ft_time = time.perf_counter() - start

print("原生PyTorch耗时: {:.2f} ms".format(pt_time * 1000))
print("FasterTransformer耗时: {:.2f} ms".format(ft_time * 1000))
print("加速比: {:.2f}x".format(pt_time / ft_time))

配方3：多卡推理（模型并行）

对于特别大的模型（比如100B+参数），一张NPU放不下，需要做模型并行推理。

完整步骤：

1. 切分模型

import torch
import torch.nn as nn

class ParallelGPT(nn.Module):
    def __init__(self, num_layers, hidden_dim, num_gpus):
        super().__init__()
        self.num_layers = num_layers
        self.hidden_dim = hidden_dim
        self.num_gpus = num_gpus
        
        # 把层数均分到每张GPU
        layers_per_gpu = num_layers // num_gpus
        
        self.gpu_layers = nn.ModuleList()
        for gpu_id in range(num_gpus):
            start = gpu_id * layers_per_gpu
            end = start + layers_per_gpu if gpu_id < num_gpus - 1 else num_layers
            
            # 把这部分的层放到对应GPU上
            layers = nn.Sequential(
                *[GPTLayer(hidden_dim) for _ in range(start, end)]
            ).npu(gpu_id)
            
            self.gpu_layers.append(layers)
    
    def forward(self, x):
        # 输入在GPU 0上
        for gpu_id, layers in enumerate(self.gpu_layers):
            # 把输入移到对应GPU
            x = x.to(gpu_id)
            # 前向传播
            x = layers(x)
        # 输出在最后一张GPU上
        return x.to(0)

2. 推理脚本

# 1. 创建模型（假设有80层，放在8张NPU上）
model = ParallelGPT(num_layers=80, hidden_dim=8192, num_gpus=8)

# 2. 推理
input_ids = torch.randint(0, 32000, (1, 512)).npu(0)
output = model(input_ids)

print("输出形状:", output.shape)

3. 性能监控

import torch
import time

# 1. 监控每张NPU的显存和利用率
def monitor_npus():
    for i in range(torch.npu.device_count()):
        memory_allocated = torch.npu.memory_allocated(i) / 1024**3
        memory_reserved = torch.npu.memory_reserved(i) / 1024**3
        print("NPU {}: 已分配 {:.2f} GB, 缓存 {:.2f} GB".format(
            i, memory_allocated, memory_reserved
        ))

# 2. 推理时监控
input_ids = torch.randint(0, 32000, (1, 512)).npu(0)

monitor_npus()  # 推理前

start = time.perf_counter()
output = model(input_ids)
torch.npu.synchronize()
elapsed = time.perf_counter() - start

monitor_npus()  # 推理后

print("推理耗时: {:.2f} s".format(elapsed))

三、实战踩坑与解决方案

坑1：算子不支持

错误信息：RuntimeError: Op XXX not implemented for NPU

解决方案：

# 1. 检查算子是否在ops-nn/ops-transformer等仓库中
# 2. 如果不在，可以自己用Ascend C写算子
# 3. 或者暂时用CPU替代（性能会差）
x = x.cpu()
output = unsupported_op(x)
output = output.npu()

坑2：显存溢出

错误信息：RuntimeError: NPU out of memory

解决方案：

# 1. 减小batch size
batch_size = 1  # 从16减小到1

# 2. 启用梯度检查点（如果训练）
from torch.utils.checkpoint import checkpoint

# 3. 及时释放不需要的张量
del intermediate_tensor
torch.npu.empty_cache()

# 4. 使用模型并行（如果模型太大）
model = ParallelGPT(...)

坑3：性能不佳

解决方案：

# 1. 启用算子融合
torch_npu.set_compile_option("fusion_level", 2)

# 2. 启用混合精度
model = model.half()
input_data = input_data.half()

# 3. 使用FasterTransformer等加速库
from faster_transformer import GlmModel
model = GlmModel.from_pretrained(...)

# 4. 做性能分析，找到瓶颈
import torch_npu.profiler as profiler
with profiler.profile(activities=[profiler.ProfilerActivity.NPU]) as prof:
    _ = model(input_data)
print(prof.key_averages().table(sort_by="self_npu_time_total"))

四、性能压测与调优

1. 吞吐量压测

import torch
import time

def benchmark_throughput(model, input_length, output_length, num_iterations=100):
    """测试推理吞吐量（tokens/s）"""
    input_ids = torch.randint(0, 32000, (1, input_length)).npu()
    
    # 预热
    with torch.no_grad():
        _ = model.generate(input_ids, max_length=output_length)
    
    # 正式测试
    torch.npu.synchronize()
    start = time.perf_counter()
    
    for _ in range(num_iterations):
        with torch.no_grad():
            output_ids = model.generate(input_ids, max_length=output_length)
    
    torch.npu.synchronize()
    elapsed = time.perf_counter() - start
    
    num_tokens = (output_ids.shape[1] - input_length) * num_iterations
    throughput = num_tokens / elapsed
    
    print("吞吐量: {:.2f} tokens/s".format(throughput))
    return throughput

# 使用
model = GPT2LMHeadModel.from_pretrained("gpt2").npu()
benchmark_throughput(model, input_length=50, output_length=100)

2. 延迟压测

def benchmark_latency(model, input_length, output_length, num_iterations=100):
    """测试推理延迟（ms）"""
    input_ids = torch.randint(0, 32000, (1, input_length)).npu()
    
    latencies = []
    for _ in range(num_iterations):
        torch.npu.synchronize()
        start = time.perf_counter()
        
        with torch.no_grad():
            _ = model.generate(input_ids, max_length=output_length)
        
        torch.npu.synchronize()
        end = time.perf_counter()
        latencies.append((end - start) * 1000)  # 转为ms
    
    import numpy as np
    print("延迟统计 (ms):")
    print("  平均: {:.2f}".format(np.mean(latencies)))
    print("  中位数: {:.2f}".format(np.median(latencies)))
    print("  P90: {:.2f}".format(np.percentile(latencies, 90)))
    print("  P99: {:.2f}".format(np.percentile(latencies, 99)))
    
    return latencies

# 使用
benchmark_latency(model, input_length=50, output_length=100)

3. 调优建议

根据压测结果，可以有针对性地调优：

• 如果吞吐量低：启用算子融合、使用FasterTransformer、增大batch size（如果显存够）
• 如果延迟高：启用KV缓存、使用混合精度、减少beam search的beam数
• 如果显存溢出：减小batch size、启用梯度检查点、使用模型并行

五、总结

cann-recipes-infer这个仓库，对于想在昇腾NPU上部署大模型推理的开发者来说，价值巨大。它提供了可复现的实战配方，让你不用从零摸索，直接照着做就能跑通。

仓库里的配方覆盖了很多主流大模型（GPT、GLM、LLaMA等），也提供了多种推理优化技术（算子融合、模型并行、FasterTransformer等）。遇到问题时，可以先查仓库的Issues，或者到昇腾社区论坛提问。

当然，这个仓库也不是万能的。有些特别新的模型可能还没适配，需要你自己去改配方。但这种改配方的过程，也是深入理解CANN推理链路的好机会。

希望这篇文章对你有帮助。如果有其他问题，欢迎在评论区讨论。