前言

深度学习模型从训练到实际部署，需要经历复杂的优化和适配过程。昇腾 CANN 的 cann-recipes-infer 仓提供了丰富的推理优化食谱，覆盖图像分类、目标检测、自然语言处理等多个领域。本文深入解析 cann-recipes-infer 的架构设计、核心内容和实际应用方法。

cann-recipes-infer 在 CANN 生态中的位置

CANN 生态包含多个层次的组件，cann-recipes-infer 位于应用使能层，起到最佳实践传递和快速原型开发的关键作用：

CANN 生态架构：
├── 硬件层：昇腾 AI 处理器（910、310、610 等）
├── 驱动层：Driver（设备驱动、内存管理、任务调度）
├── 运行时：Runtime（模型加载、内存管理、任务调度）
├── 编译器：GE 图引擎、Blaze 张量引擎
├── 算子库：ops-transformer、ops-nn、ops-math 等
├── 应用框架：PyTorch、TensorFlow、MindSpore 适配层
└── 应用使能层：cann-recipes-infer ← 本文重点

cann-recipes-infer 的主要功能包括：

  1. 推理优化食谱：提供各种模型的推理优化方法和代码
  2. 性能分析工具：提供性能分析和调优工具
  3. 部署最佳实践：提供模型部署的最佳实践指南
  4. 示例代码库：提供丰富的推理示例代码

cann-recipes-infer 架构设计

整体架构

cann-recipes-infer 采用模块化设计，各组件职责清晰：

cann-recipes-infer
├── 图像分类食谱（Image Classification Recipes）
│   ├── ResNet-50 推理优化
│   ├── MobileNet 推理优化
│   └── EfficientNet 推理优化
├── 目标检测食谱（Object Detection Recipes）
│   ├── YOLOv8 推理优化
│   ├── Faster R-CNN 推理优化
│   └── SSD 推理优化
├── 自然语言处理食谱（NLP Recipes）
│   ├── BERT 推理优化
│   ├── GPT 推理优化
│   └── LLaMA 推理优化
├── 推荐系统食谱（Recommendation Recipes）
│   ├── DIN 推理优化
│   ├── DIEN 推理优化
│   └── DeepFM 推理优化
├── 性能分析工具（Performance Analysis Tools）
│   ├── 推理延迟分析工具
│   ├── 吞吐量分析工具
│   └── 内存使用分析工具
└── 部署最佳实践（Deployment Best Practices）
    ├── 单机部署最佳实践
    ├── 集群部署最佳实践
    └── 云部署最佳实践

核心组件详解

1. 图像分类食谱

针对图像分类模型，提供完整的推理优化食谱。

核心内容：

  1. 模型转换：将训练好的模型转换为 CANN 支持的格式
  2. 推理优化：应用算子融合、内存优化等技术提升推理性能
  3. 精度验证：验证优化后模型的精度是否满足要求
  4. 性能测试：测试优化后模型的推理延迟和吞吐量
# ResNet-50 推理优化食谱
import torch
import torchvision
from cann import ge, runtime

# 1. 模型转换
model = torchvision.models.resnet50(pretrained=True)
model.eval()

# 导出为 ONNX 格式
dummy_input = torch.randn(1, 3, 224, 224)
torch.onnx.export(model, dummy_input, 'resnet50.onnx')

# 2. 使用 GE 优化模型
ge_parser = ge.FrontendParser()
ge_graph = ge_parser.parse_onnx_model('resnet50.onnx')

ge_optimizer = ge.GraphOptimizer()
ge_optimizer.enable_operator_fusion(True)
ge_optimizer.enable_constant_folding(True)
ge_optimizer.enable_memory_optimization(True)

optimized_graph = ge_optimizer.optimize(ge_graph)

ge_compiler = ge.GraphCompiler()
ge_compiler.set_target_device('ascend310')
ge_compiler.set_precision_mode('fp16')

compiled_model = ge_compiler.compile(optimized_graph)
compiled_model.save('resnet50_optimized.om')

# 3. 推理优化
class InferenceOptimizer:
    def __init__(self, model_path):
        self.model_path = model_path
        
        # 初始化 ACL
        acl.init()
        acl.rt.set_device(0)
        
        # 加载模型
        self.model_id = acl.mdl.load_from_file(model_path)
        
        # 创建流
        self.stream = acl.rt.create_stream()
        
        print(f"模型加载成功: {model_path}")
    
    def optimize_memory(self):
        """内存优化"""
        # 设置内存分配策略
        acl.rt.set_device_mem_pool_size(0, 2 * 1024 * 1024 * 1024)  # 2GB
        
        # 启用内存复用
        acl.rt.enable_memory_reuse(True)
        
        print("内存优化完成")
    
    def optimize_pipeline(self):
        """流水线优化"""
        # 创建多个流实现流水线
        self.stream1 = acl.rt.create_stream()
        self.stream2 = acl.rt.create_stream()
        self.stream3 = acl.rt.create_stream()
        
        print("流水线优化完成")
    
    def infer(self, input_data):
        """执行推理"""
        # 分配设备内存
        input_size = input_data.nbytes
        dev_input = acl.rt.malloc(input_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        output_size = input_size
        dev_output = acl.rt.malloc(output_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        
        # 主机到设备数据传输
        acl.rt.memcpy_async(
            dev_input, input_size,
            input_data.tobytes(), input_size,
            ACL_MEMCPY_HOST_TO_DEVICE, self.stream
        )
        
        # 执行推理
        acl.mdl.execute_async(self.model_id, [dev_input], [dev_output], self.stream)
        
        # 设备到主机数据传输
        output_data = np.empty_like(input_data)
        acl.rt.memcpy_async(
            output_data.tobytes(), output_size,
            dev_output, output_size,
            ACL_MEMCPY_DEVICE_TO_HOST, self.stream
        )
        
        # 等待执行完成
        acl.rt.synchronize_stream(self.stream)
        
        # 清理资源
        acl.rt.free(dev_input)
        acl.rt.free(dev_output)
        
        return output_data
    
    def cleanup(self):
        """清理资源"""
        acl.mdl.unload(self.model_id)
        acl.rt.destroy_stream(self.stream)
        acl.rt.reset_device(0)
        acl.finalize()
        print("资源清理完成")

# 使用示例
optimizer = InferenceOptimizer('resnet50_optimized.om')
optimizer.optimize_memory()
optimizer.optimize_pipeline()

# 预处理输入数据
input_data = preprocess_image('test.jpg')

# 执行推理
output = optimizer.infer(input_data)

# 后处理输出数据
result = postprocess_output(output)

print(f"推理结果: {result}")

optimizer.cleanup()
2. 目标检测食谱

针对目标检测模型，提供完整的推理优化食谱。

核心内容：

  1. 模型转换：将训练好的目标检测模型转换为 CANN 支持的格式
  2. 推理优化：应用算子融合、内存优化等技术提升推理性能
  3. 后处理优化：优化 NMS、边界框解码等后处理操作
  4. 精度验证：验证优化后模型的精度是否满足要求
# YOLOv8 推理优化食谱
import torch
from cann import ge, runtime

# 1. 模型转换
model = torch.hub.load('ultralytics/yolov8', 'yolov8s')

# 导出为 ONNX 格式
dummy_input = torch.randn(1, 3, 640, 640)
model.export(format='onnx', imgsz=640)

# 2. 使用 GE 优化模型
ge_parser = ge.FrontendParser()
ge_graph = ge_parser.parse_onnx_model('yolov8s.onnx')

ge_optimizer = ge.GraphOptimizer()
ge_optimizer.enable_operator_fusion(True)
ge_optimizer.enable_constant_folding(True)
ge_optimizer.enable_memory_optimization(True)

optimized_graph = ge_optimizer.optimize(ge_graph)

ge_compiler = ge.GraphCompiler()
ge_compiler.set_target_device('ascend310')
ge_compiler.set_precision_mode('fp16')

compiled_model = ge_compiler.compile(optimized_graph)
compiled_model.save('yolov8s_optimized.om')

# 3. 后处理优化
class PostprocessOptimizer:
    def __init__(self, model_path):
        self.model_path = model_path
        
        # 初始化 ACL
        acl.init()
        acl.rt.set_device(0)
        
        # 加载模型
        self.model_id = acl.mdl.load_from_file(model_path)
        
        # 创建流
        self.stream = acl.rt.create_stream()
        
        print(f"模型加载成功: {model_path}")
    
    def optimize_nms(self):
        """NMS 优化"""
        # 使用昇腾的 NMS 算子
        self.nms_op = acl.op.create_operator('NMS')
        acl.op.set_attr(self.nms_op, 'iou_threshold', 0.45)
        acl.op.set_attr(self.nms_op, 'score_threshold', 0.25)
        
        print("NMS 优化完成")
    
    def optimize_bbox_decode(self):
        """边界框解码优化"""
        # 使用昇腾的边界框解码算子
        self.bbox_decode_op = acl.op.create_operator('BBoxDecode')
        
        print("边界框解码优化完成")
    
    def postprocess(self, output_data):
        """后处理"""
        # NMS
        nms_output = acl.op.execute(self.nms_op, [output_data])
        
        # 边界框解码
        bbox_output = acl.op.execute(self.bbox_decode_op, [nms_output])
        
        return bbox_output
    
    def infer(self, input_data):
        """执行推理"""
        # 分配设备内存
        input_size = input_data.nbytes
        dev_input = acl.rt.malloc(input_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        output_size = input_size * 100  # 假设输出大小是输入的 100 倍
        dev_output = acl.rt.malloc(output_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        
        # 主机到设备数据传输
        acl.rt.memcpy_async(
            dev_input, input_size,
            input_data.tobytes(), input_size,
            ACL_MEMCPY_HOST_TO_DEVICE, self.stream
        )
        
        # 执行推理
        acl.mdl.execute_async(self.model_id, [dev_input], [dev_output], self.stream)
        
        # 设备到主机数据传输
        output_data = np.empty(output_size // 4, dtype=np.float32)  # 假设是 FP32
        acl.rt.memcpy_async(
            output_data.tobytes(), output_size,
            dev_output, output_size,
            ACL_MEMCPY_DEVICE_TO_HOST, self.stream
        )
        
        # 等待执行完成
        acl.rt.synchronize_stream(self.stream)
        
        # 后处理
        post_output = self.postprocess(output_data)
        
        # 清理资源
        acl.rt.free(dev_input)
        acl.rt.free(dev_output)
        
        return post_output
    
    def cleanup(self):
        """清理资源"""
        acl.op.destroy_operator(self.nms_op)
        acl.op.destroy_operator(self.bbox_decode_op)
        acl.mdl.unload(self.model_id)
        acl.rt.destroy_stream(self.stream)
        acl.rt.reset_device(0)
        acl.finalize()
        print("资源清理完成")

# 使用示例
optimizer = PostprocessOptimizer('yolov8s_optimized.om')
optimizer.optimize_nms()
optimizer.optimize_bbox_decode()

# 预处理输入数据
input_data = preprocess_image('test.jpg', target_size=640)

# 执行推理
output = optimizer.infer(input_data)

# 可视化结果
visualize_results(output, 'test.jpg')

optimizer.cleanup()
3. 自然语言处理食谱

针对自然语言处理模型，提供完整的推理优化食谱。

核心内容：

  1. 模型转换：将训练好的 NLP 模型转换为 CANN 支持的格式
  2. 推理优化：应用算子融合、内存优化等技术提升推理性能
  3. KV Cache 优化：优化 Transformer 模型的 KV Cache 管理
  4. 精度验证：验证优化后模型的精度是否满足要求
# BERT 推理优化食谱
import torch
from transformers import BertModel, BertTokenizer
from cann import ge, runtime

# 1. 模型转换
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertModel.from_pretrained('bert-base-uncased')
model.eval()

# 导出为 ONNX 格式
dummy_input = tokenizer('Hello, world!', return_tensors='pt')['input_ids']
torch.onnx.export(model, dummy_input, 'bert-base.onnx')

# 2. 使用 GE 优化模型
ge_parser = ge.FrontendParser()
ge_graph = ge_parser.parse_onnx_model('bert-base.onnx')

ge_optimizer = ge.GraphOptimizer()
ge_optimizer.enable_operator_fusion(True)
ge_optimizer.enable_constant_folding(True)
ge_optimizer.enable_memory_optimization(True)

optimized_graph = ge_optimizer.optimize(ge_graph)

ge_compiler = ge.GraphCompiler()
ge_compiler.set_target_device('ascend310')
ge_compiler.set_precision_mode('fp16')

compiled_model = ge_compiler.compile(optimized_graph)
compiled_model.save('bert-base_optimized.om')

# 3. KV Cache 优化
class KVCacheOptimizer:
    def __init__(self, model_path, max_seq_len=512, n_heads=12, head_dim=64):
        self.model_path = model_path
        self.max_seq_len = max_seq_len
        self.n_heads = n_heads
        self.head_dim = head_dim
        
        # 初始化 ACL
        acl.init()
        acl.rt.set_device(0)
        
        # 加载模型
        self.model_id = acl.mdl.load_from_file(model_path)
        
        # 创建流
        self.stream = acl.rt.create_stream()
        
        # 初始化 KV Cache
        self.k_cache = acl.rt.malloc(
            max_seq_len * n_heads * head_dim * 2,  # FP16
            ACL_MEM_MALLOC_NORMAL_ONLY
        )
        self.v_cache = acl.rt.malloc(
            max_seq_len * n_heads * head_dim * 2,  # FP16
            ACL_MEM_MALLOC_NORMAL_ONLY
        )
        self.cache_len = 0
        
        print(f"模型加载成功: {model_path}")
        print(f"KV Cache 初始化成功: {max_seq_len} tokens")
    
    def update_cache(self, k, v):
        """更新 KV Cache"""
        # 计算新 Cache 的偏移量
        offset = self.cache_len * self.n_heads * self.head_dim * 2
        
        # 将新的 k, v 复制到 Cache
        k_size = k.nbytes
        v_size = v.nbytes
        
        acl.rt.memcpy(
            self.k_cache + offset, k_size,
            k.tobytes(), k_size,
            ACL_MEMCPY_HOST_TO_DEVICE
        )
        acl.rt.memcpy(
            self.v_cache + offset, v_size,
            v.tobytes(), v_size,
            ACL_MEMCPY_HOST_TO_DEVICE
        )
        
        # 更新 Cache 长度
        self.cache_len += k.shape[1]
        
        print(f"KV Cache 更新: {self.cache_len} tokens")
    
    def get_cache(self):
        """获取 KV Cache"""
        # 计算 Cache 大小
        cache_size = self.cache_len * self.n_heads * self.head_dim * 2
        
        # 分配设备内存
        dev_k = acl.rt.malloc(cache_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        dev_v = acl.rt.malloc(cache_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        
        # 从 Cache 复制数据
        acl.rt.memcpy(
            dev_k, cache_size,
            self.k_cache, cache_size,
            ACL_MEMCPY_DEVICE_TO_DEVICE
        )
        acl.rt.memcpy(
            dev_v, cache_size,
            self.v_cache, cache_size,
            ACL_MEMCPY_DEVICE_TO_DEVICE
        )
        
        return dev_k, dev_v
    
    def infer(self, input_ids):
        """执行推理"""
        # 准备输入数据
        input_data = input_ids.numpy().astype(np.int32)
        input_size = input_data.nbytes
        
        # 分配设备内存
        dev_input = acl.rt.malloc(input_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        
        # 主机到设备数据传输
        acl.rt.memcpy_async(
            dev_input, input_size,
            input_data.tobytes(), input_size,
            ACL_MEMCPY_HOST_TO_DEVICE, self.stream
        )
        
        # 获取 KV Cache
        dev_k, dev_v = self.get_cache()
        
        # 执行推理
        output_size = input_size * 768  # 假设输出大小是输入的 768 倍
        dev_output = acl.rt.malloc(output_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        
        # 将 KV Cache 作为额外的输入
        acl.mdl.execute_async(
            self.model_id,
            [dev_input, dev_k, dev_v],
            [dev_output],
            self.stream
        )
        
        # 设备到主机数据传输
        output_data = np.empty(output_size // 4, dtype=np.float32)  # 假设是 FP32
        acl.rt.memcpy_async(
            output_data.tobytes(), output_size,
            dev_output, output_size,
            ACL_MEMCPY_DEVICE_TO_HOST, self.stream
        )
        
        # 等待执行完成
        acl.rt.synchronize_stream(self.stream)
        
        # 清理资源
        acl.rt.free(dev_input)
        acl.rt.free(dev_k)
        acl.rt.free(dev_v)
        acl.rt.free(dev_output)
        
        return output_data
    
    def cleanup(self):
        """清理资源"""
        acl.rt.free(self.k_cache)
        acl.rt.free(self.v_cache)
        acl.mdl.unload(self.model_id)
        acl.rt.destroy_stream(self.stream)
        acl.rt.reset_device(0)
        acl.finalize()
        print("资源清理完成")

# 使用示例
optimizer = KVCacheOptimizer('bert-base_optimized.om')
optimizer.optimize_kv_cache()

# 预处理输入数据
input_text = "Hello, world!"
input_ids = tokenizer(input_text, return_tensors='pt')['input_ids']

# 执行推理
output = optimizer.infer(input_ids)

# 后处理输出数据
result = postprocess_output(output)

print(f"推理结果: {result}")

optimizer.cleanup()

性能分析工具

cann-recipes-infer 提供了丰富的性能分析工具，帮助用户分析和优化模型推理性能。

1. 推理延迟分析工具

分析模型推理的端到端延迟，找出性能瓶颈。

# 推理延迟分析工具示例
import time

class LatencyAnalyzer:
    def __init__(self, model_path, num_iterations=100):
        self.model_path = model_path
        self.num_iterations = num_iterations
        
        # 初始化 ACL
        acl.init()
        acl.rt.set_device(0)
        
        # 加载模型
        self.model_id = acl.mdl.load_from_file(model_path)
        
        # 创建流
        self.stream = acl.rt.create_stream()
        
        print(f"模型加载成功: {model_path}")
    
    def analyze(self, input_data):
        """分析推理延迟"""
        # 预热
        for i in range(10):
            self.infer(input_data)
        
        # 测试
        latencies = []
        for i in range(self.num_iterations):
            start = time.time()
            self.infer(input_data)
            end = time.time()
            latencies.append((end - start) * 1000)  # 转换为毫秒
        
        # 统计分析
        mean_latency = np.mean(latencies)
        std_latency = np.std(latencies)
        min_latency = np.min(latencies)
        max_latency = np.max(latencies)
        p50_latency = np.percentile(latencies, 50)
        p90_latency = np.percentile(latencies, 90)
        p99_latency = np.percentile(latencies, 99)
        
        print(f"推理延迟分析 (n={self.num_iterations}):")
        print(f"  平均延迟: {mean_latency:.2f} ms")
        print(f"  标准差: {std_latency:.2f} ms")
        print(f"  最小延迟: {min_latency:.2f} ms")
        print(f"  最大延迟: {max_latency:.2f} ms")
        print(f"  P50 延迟: {p50_latency:.2f} ms")
        print(f"  P90 延迟: {p90_latency:.2f} ms")
        print(f"  P99 延迟: {p99_latency:.2f} ms")
        
        return {
            'mean': mean_latency,
            'std': std_latency,
            'min': min_latency,
            'max': max_latency,
            'p50': p50_latency,
            'p90': p90_latency,
            'p99': p99_latency
        }
    
    def infer(self, input_data):
        """执行推理"""
        # 分配设备内存
        input_size = input_data.nbytes
        dev_input = acl.rt.malloc(input_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        output_size = input_size
        dev_output = acl.rt.malloc(output_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        
        # 主机到设备数据传输
        acl.rt.memcpy_async(
            dev_input, input_size,
            input_data.tobytes(), input_size,
            ACL_MEMCPY_HOST_TO_DEVICE, self.stream
        )
        
        # 执行推理
        acl.mdl.execute_async(self.model_id, [dev_input], [dev_output], self.stream)
        
        # 设备到主机数据传输
        output_data = np.empty_like(input_data)
        acl.rt.memcpy_async(
            output_data.tobytes(), output_size,
            dev_output, output_size,
            ACL_MEMCPY_DEVICE_TO_HOST, self.stream
        )
        
        # 等待执行完成
        acl.rt.synchronize_stream(self.stream)
        
        # 清理资源
        acl.rt.free(dev_input)
        acl.rt.free(dev_output)
        
        return output_data
    
    def cleanup(self):
        """清理资源"""
        acl.mdl.unload(self.model_id)
        acl.rt.destroy_stream(self.stream)
        acl.rt.reset_device(0)
        acl.finalize()
        print("资源清理完成")

# 使用示例
analyzer = LatencyAnalyzer('resnet50_optimized.om')
input_data = preprocess_image('test.jpg')
result = analyzer.analyze(input_data)
analyzer.cleanup()

2. 吞吐量分析工具

分析模型推理的吞吐量，评估模型的并发处理能力。

# 吞吐量分析工具示例
import time

class ThroughputAnalyzer:
    def __init__(self, model_path, num_iterations=100):
        self.model_path = model_path
        self.num_iterations = num_iterations
        
        # 初始化 ACL
        acl.init()
        acl.rt.set_device(0)
        
        # 加载模型
        self.model_id = acl.mdl.load_from_file(model_path)
        
        # 创建流
        self.stream = acl.rt.create_stream()
        
        print(f"模型加载成功: {model_path}")
    
    def analyze(self, input_data, batch_size=1):
        """分析吞吐量"""
        # 预热
        for i in range(10):
            self.infer(input_data, batch_size)
        
        # 测试
        start = time.time()
        for i in range(self.num_iterations):
            self.infer(input_data, batch_size)
        end = time.time()
        
        # 计算吞吐量
        total_samples = self.num_iterations * batch_size
        elapsed_time = end - start
        throughput = total_samples / elapsed_time
        
        print(f"吞吐量分析 (n={self.num_iterations}, batch_size={batch_size}):")
        print(f"  总样本数: {total_samples}")
        print(f"  总时间: {elapsed_time:.2f} s")
        print(f"  吞吐量: {throughput:.2f} samples/s")
        
        return {
            'total_samples': total_samples,
            'elapsed_time': elapsed_time,
            'throughput': throughput
        }
    
    def infer(self, input_data, batch_size=1):
        """执行推理"""
        # 扩展 batch 维度
        input_data = input_data.repeat(batch_size, 1, 1, 1)
        
        # 分配设备内存
        input_size = input_data.nbytes
        dev_input = acl.rt.malloc(input_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        output_size = input_size
        dev_output = acl.rt.malloc(output_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        
        # 主机到设备数据传输
        acl.rt.memcpy_async(
            dev_input, input_size,
            input_data.tobytes(), input_size,
            ACL_MEMCPY_HOST_TO_DEVICE, self.stream
        )
        
        # 执行推理
        acl.mdl.execute_async(self.model_id, [dev_input], [dev_output], self.stream)
        
        # 设备到主机数据传输
        output_data = np.empty_like(input_data)
        acl.rt.memcpy_async(
            output_data.tobytes(), output_size,
            dev_output, output_size,
            ACL_MEMCPY_DEVICE_TO_HOST, self.stream
        )
        
        # 等待执行完成
        acl.rt.synchronize_stream(self.stream)
        
        # 清理资源
        acl.rt.free(dev_input)
        acl.rt.free(dev_output)
        
        return output_data
    
    def cleanup(self):
        """清理资源"""
        acl.mdl.unload(self.model_id)
        acl.rt.destroy_stream(self.stream)
        acl.rt.reset_device(0)
        acl.finalize()
        print("资源清理完成")

# 使用示例
analyzer = ThroughputAnalyzer('resnet50_optimized.om')
input_data = preprocess_image('test.jpg')
result = analyzer.analyze(input_data, batch_size=4)
analyzer.cleanup()

部署最佳实践

cann-recipes-infer 提供了丰富的部署最佳实践，帮助用户在不同环境下部署模型。

1. 单机部署最佳实践

针对单机环境，提供最佳的部署方案。

核心内容：

  1. 模型优化：应用各种优化技术提升模型性能
  2. 资源管理：合理管理设备资源，避免资源泄漏
  3. 并发处理：支持多并发推理请求
  4. 监控告警：监控模型性能，及时发现问题
# 单机部署最佳实践示例
import threading
import queue

class SingleNodeDeployment:
    def __init__(self, model_path, num_workers=4):
        self.model_path = model_path
        self.num_workers = num_workers
        
        # 初始化 ACL
        acl.init()
        acl.rt.set_device(0)
        
        # 加载模型（多个 worker 共享同一个模型）
        self.model_id = acl.mdl.load_from_file(model_path)
        
        # 创建任务队列
        self.task_queue = queue.Queue()
        
        # 创建 worker 线程
        self.workers = []
        for i in range(num_workers):
            worker = threading.Thread(target=self.worker_loop)
            worker.start()
            self.workers.append(worker)
        
        print(f"单机部署初始化成功: {model_path}")
        print(f"  Worker 数量: {num_workers}")
    
    def worker_loop(self):
        """Worker 线程循环"""
        # 每个 worker 创建自己的流
        stream = acl.rt.create_stream()
        
        while True:
            # 从任务队列获取任务
            task = self.task_queue.get()
            
            if task is None:
                # 收到结束信号
                break
            
            # 执行推理
            input_data, callback = task
            output = self.infer(input_data, stream)
            
            # 调用回调函数
            callback(output)
            
            # 标记任务完成
            self.task_queue.task_done()
        
        # 清理流
        acl.rt.destroy_stream(stream)
    
    def infer(self, input_data, stream):
        """执行推理"""
        # 分配设备内存
        input_size = input_data.nbytes
        dev_input = acl.rt.malloc(input_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        output_size = input_size
        dev_output = acl.rt.malloc(output_size, ACL_MEM_MALLOC_NORMAL_ONLY)
        
        # 主机到设备数据传输
        acl.rt.memcpy_async(
            dev_input, input_size,
            input_data.tobytes(), input_size,
            ACL_MEMCPY_HOST_TO_DEVICE, stream
        )
        
        # 执行推理
        acl.mdl.execute_async(self.model_id, [dev_input], [dev_output], stream)
        
        # 设备到主机数据传输
        output_data = np.empty_like(input_data)
        acl.rt.memcpy_async(
            output_data.tobytes(), output_size,
            dev_output, output_size,
            ACL_MEMCPY_DEVICE_TO_HOST, stream
        )
        
        # 等待执行完成
        acl.rt.synchronize_stream(stream)
        
        # 清理资源
        acl.rt.free(dev_input)
        acl.rt.free(dev_output)
        
        return output_data
    
    def submit_task(self, input_data, callback):
        """提交推理任务"""
        self.task_queue.put((input_data, callback))
    
    def shutdown(self):
        """关闭部署服务"""
        # 发送结束信号给所有 worker
        for i in range(self.num_workers):
            self.task_queue.put(None)
        
        # 等待所有 worker 结束
        for worker in self.workers:
            worker.join()
        
        # 清理资源
        acl.mdl.unload(self.model_id)
        acl.rt.reset_device(0)
        acl.finalize()
        
        print("单机部署关闭成功")

# 使用示例
deployment = SingleNodeDeployment('resnet50_optimized.om', num_workers=4)

# 定义回调函数
def callback(output):
    result = postprocess_output(output)
    print(f"推理结果: {result}")

# 提交推理任务
input_data = preprocess_image('test.jpg')
deployment.submit_task(input_data, callback)

# 等待所有任务完成
deployment.task_queue.join()

# 关闭部署服务
deployment.shutdown()

总结

cann-recipes-infer 作为昇腾 CANN 的推理食谱集合，提供了丰富的推理优化食谱、性能分析工具和部署最佳实践，大幅降低了模型部署的难度。通过学习和应用这些食谱，可以快速掌握 CANN 的推理优化技能，并应用于实际项目中。

完整的 cann-recipes-infer 文档和示例代码可以在昇腾官方文档中心找到。

# 大数据 # 人工智能 # 架构
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

iPaaS平台全景扫描：五款主流集成产品解读

avatar AtomGit开源社区
cover

AI大模型赋能数据分析：3大场景+5大避坑指南，附代码Prompt，即学即用！

avatar AtomGit开源社区
cover

iPaaS集成平台：五大产品关键能力速查

avatar AtomGit开源社区