CANN Recipes 推理 - 推理应用场景实战

小a彤

377人浏览 · 2026-05-27 13:41:28

小a彤 · 2026-05-27 13:41:28 发布

在这里插入图片描述

前言

cann-recipes 是 CANN 官方提供的应用实践指南覆盖推理和训练两大场景本文专门介绍推理场景的最佳实践

推理场景概述

深度学习推理的应用场景

云端推理服务器部署
边缘推理端侧部署
实时推理低延迟要求
批量推理高吞吐要求

推理优化技术

模型优化

剪枝

import torch.nn.utils.prune as prune

# 结构化剪枝
prune.l1_unstructured(
    model.layer1.conv1,
    name='weight',
    amount=0.5,
)

量化

import torch.quantization

# 量化配置
model.qconfig = torch.quantization.default_qconfig

# 准备量化
torch.quantization.prepare(model, inplace=True)

# 校准
for data in calibrate_loader:
    model(data)

# 转换
 quantized_model = torch.quantization.convert(model)

知识蒸馏

# 教师模型推理
teacher_output = teacher(input)

# 学生模型推理
student_output = student(input)

# 蒸馏损失
loss = F.kl_div(student_output, teacher_output)

推理部署

方式一PyTorch 原生

import torch

model = torch.jit.load("model.pt")
model.eval()

# 推理
with torch.no_grad():
    output = model(input)

方式二ONNX Runtime

import onnxruntime as ort

# 创建会话
session = ort.InferenceSession("model.onnx")

# 推理
output = session.run(None, {"input": input})

方式三Ascend CL

import acl

# 初始化
acl.init()

# 加载模型
model_id = acl.rt.load_model("model.om")

# 创建输入
input_buffer = acl.util.numpy_to_npy(input_array)

# 执行推理
result = acl.rt.execute(model_id, input_buffer)

推理性能优化

批处理

# 静态批处理
batch_size = 32
for i in range(0, dataset_size, batch_size):
    batch = dataset[i:i+batch_size]
    results = model(batch)

异步执行

import asyncio

async def infer_async(model, input_data):
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, model, input_data)
    return result

流水线化

# 预处理流水线
preprocess_queue = Queue()

# 执行流水线
while True:
    batch = preprocess_queue.get()
    result = infer_step(batch)
    postprocess(result)

推理服务

Flask 服务

from flask import Flask, request

app = Flask(__name__)

@app.route('/infer', methods=['POST'])
def infer():
    data = request.json['data']
    result = model.predict(data)
    return {'result': result.tolist()}

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

gRPC 服务

import grpc
from import_inference_pb2 import *
from import_inference_pb2_grpc import *

class InferenceServicer(import_inference_pb2_grpc.InferenceServicer):
    def Infer(self, request, context):
        result = model.infer(request.data)
        return InferResponse(result=result)

TensorRT Serving

# 启动服务
trtexec --onnx=model.onnx --saveEngine=model.plan

# 推理请求
curl -X POST localhost:8000/v1/models/model:predict   -d '{"inputs": [{"name": "input", "data": [...]}]}'

推理案例

案例一ResNet-50 推理

import torch
from torchvision.models import resnet50

# 加载模型
model = resnet50(pretrained=True).npu()
model.eval()

# 预处理
transform = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
])

# 单张图像推理
def predict_image(image_path):
    img = Image.open(image_path)
    img = transform(img).unsqueeze(0).npu()
    
    with torch.no_grad():
        output = model(img)
        pred = output.argmax(dim=1)
    
    return pred.item()

案例二BERT 推理

import torch
from transformers import BertModel

# 加载模型
model = BertModel.from_pretrained("bert-base-chinese").npu()
model.eval()

# 单序列推理
def predict_sequence(text):
    inputs = tokenizer(text, return_tensors="pt")
    inputs = {k: v.npu() for k, v in inputs.items()}
    
    with torch.no_grad():
        outputs = model(**inputs)
    
    return outputs.last_hidden_state

案例三YOLO 检测

import torch

# 加载模型
model = torch.jit.load("yolov5.pt").npu()

# 批量检测
def detect_batch(images):
    results = []
    
    with torch.no_grad():
        for img in images:
            pred = model(img)
            results.append(postprocess(pred))
    
    return results