昇思 MindSpore ：使用 ACL 接口进行应用开发的基本流程与实现方法

本文介绍了基于昇腾AI处理器的统一编程接口AscendCL（ACL）与MindSpore框架的协同开发流程。主要内容包括：1）ACL提供硬件调度、模型加载和推理执行能力，实现训练到部署的无缝衔接；2）详细开发流程涵盖环境准备、模型转换（MindSpore→ONNX→OM）、ACL初始化、资源管理、数据预处理、推理执行和结果解析；3）提供完整Python实现代码示例，展示图像分类任务从模型加载到推理

昇思MindSpore

31人浏览 · 2026-04-22 15:39:09

昇思MindSpore · 2026-04-22 15:39:09 发布

一、概述

AscendCL（ACL）是昇腾 AI 处理器的统一编程接口，为昇思 MindSpore 框架提供底层硬件调度、模型加载、内存管理、推理执行与媒体数据处理能力，实现训练框架与推理部署的无缝衔接。在 MindSpore 中调用 ACL 接口，可高效完成模型转换、离线推理、端侧部署与性能优化，兼顾开发效率与硬件算力释放。以图像分类推理为例，完整说明基于 MindSpore+ACL 的开发流程、关键接口与可运行代码，适配昇腾 310/910 系列硬件。

二、核心开发流程

环境准备：安装 CANN 工具包、MindSpore 与 ACL Python 绑定，配置设备驱动与环境变量。
模型处理：MindSpore 训练 / 导出模型，使用 ATC 工具转换为昇腾离线模型（.om）。
ACL 初始化：初始化 ACL 运行环境，指定设备、创建上下文与流。
资源申请：分配 Host/Device 内存，构建输入输出数据集。
模型加载：加载.om 离线模型，获取模型描述信息。
数据预处理：图像解码、缩放、归一化、格式转换，支持 DVPP 硬件加速。
推理执行：同步 / 异步执行模型推理，等待任务完成。
结果解析：后处理输出数据，得到分类标签与置信度。
资源释放：逆序销毁数据集、卸载模型、释放流 / 上下文、复位设备、反初始化 ACL。

三、环境与依赖

硬件：Atlas 200/200I DK、昇腾 AI 服务器（Ascend 310/910）
软件：CANN 6.0+、Python 3.7+、MindSpore 2.0+、OpenCV、NumPy
关键模块：acl、numpy、cv2

四、模型转换（MindSpore→OM）

MindSpore 导出 ONNX 或 AIR 模型

import mindspore as ms
from mindspore import load_checkpoint, load_param_into_net
from my_model import Net

net = Net()
load_param_into_net(net, load_checkpoint("resnet18.ckpt"))
ms.export(net, ms.numpy.randn(1, 3, 224, 224), file_name="resnet18", file_format="ONNX")

ATC 工具转换命令

atc --model=resnet18.onnx --output=resnet18 --framework=5 \
--input_shape="input:1,3,224,224" \
--soc_version=Ascend310

生成resnet18.om，用于 ACL 加载推理。

五、完整实现代码（Python）

import acl
import numpy as np
import cv2
import os

# 全局变量
DEVICE_ID = 0
MODEL_PATH = "./resnet18.om"
INPUT_WIDTH = 224
INPUT_HEIGHT = 224
context, stream, model_id, model_desc = None, None, None, None

def init_acl():
    global context, stream
    # ACL初始化
    ret = acl.init()
    assert ret == 0, "acl init failed"
    # 指定设备
    ret = acl.rt.set_device(DEVICE_ID)
    assert ret == 0, "set device failed"
    # 创建上下文
    context, ret = acl.rt.create_context(DEVICE_ID)
    assert ret == 0, "create context failed"
    # 创建流
    stream, ret = acl.rt.create_stream()
    assert ret == 0, "create stream failed"
    print("ACL init success")

def load_model():
    global model_id, model_desc
    # 加载离线模型
    model_id, ret = acl.mdl.load_from_file(MODEL_PATH)
    assert ret == 0, "load model failed"
    # 获取模型描述
    model_desc = acl.mdl.create_desc()
    ret = acl.mdl.get_desc(model_desc, model_id)
    assert ret == 0, "get model desc failed"
    print("Model load success, model_id:", model_id)

def preprocess(image_path):
    # 图像预处理：resize、归一化、HWC→CHW、转float32
    img = cv2.imread(image_path)
    img = cv2.resize(img, (INPUT_WIDTH, INPUT_HEIGHT))
    img = img.astype(np.float32) / 255.0
    img = img.transpose(2, 0, 1)
    img = img[np.newaxis, ...]
    return img

def create_dataset(data):
    # 创建输入数据集
    dataset = acl.mdl.create_dataset()
    # 申请Device内存
    data_ptr = acl.rt.malloc_host(data.nbytes)
    acl.rt.memcpy(data_ptr, data.nbytes, data.ctypes.data, data.nbytes,
                  acl.memcpy_host_to_device)
    data_buf = acl.create_data_buffer(data_ptr, data.nbytes)
    acl.mdl.add_dataset_buffer(dataset, data_buf)
    return dataset, data_ptr

def create_output_dataset():
    # 创建输出数据集
    dataset = acl.mdl.create_dataset()
    out_num = acl.mdl.get_num_outputs(model_desc)
    for i in range(out_num):
        size = acl.mdl.get_output_size_by_index(model_desc, i)
        ptr = acl.rt.malloc_host(size)
        buf = acl.create_data_buffer(ptr, size)
        acl.mdl.add_dataset_buffer(dataset, buf)
    return dataset

def inference(input_ds, output_ds):
    # 执行推理
    ret = acl.mdl.execute(model_id, input_ds, output_ds)
    assert ret == 0, "execute failed"
    # 同步流
    ret = acl.rt.synchronize_stream(stream)
    assert ret == 0, "sync stream failed"

def postprocess(output_ds):
    # 解析输出
    buf = acl.mdl.get_dataset_buffer(output_ds, 0)
    ptr = acl.get_data_buffer_addr(buf)
    size = acl.get_data_buffer_size(buf)
    output = np.zeros((1, 1000), dtype=np.float32)
    acl.rt.memcpy(output.ctypes.data, size, ptr, size,
                  acl.memcpy_device_to_host)
    # 获取Top1类别
    class_id = np.argmax(output)
    confidence = output[0][class_id]
    return class_id, confidence

def release_resource(input_ds, output_ds, data_ptr):
    # 释放资源
    acl.destroy_data_buffer(acl.mdl.get_dataset_buffer(input_ds, 0))
    acl.mdl.destroy_dataset(input_ds)
    acl.rt.free_host(data_ptr)
    out_num = acl.mdl.get_num_outputs(model_desc)
    for i in range(out_num):
        buf = acl.mdl.get_dataset_buffer(output_ds, i)
        ptr = acl.get_data_buffer_addr(buf)
        acl.rt.free_host(ptr)
        acl.destroy_data_buffer(buf)
    acl.mdl.destroy_dataset(output_ds)
    acl.mdl.unload(model_id)
    acl.mdl.destroy_desc(model_desc)
    acl.rt.destroy_stream(stream)
    acl.rt.destroy_context(context)
    acl.rt.reset_device(DEVICE_ID)
    acl.finalize()
    print("Resource released")

def main(image_path):
    # 初始化
    init_acl()
    load_model()
    # 预处理
    input_data = preprocess(image_path)
    # 构建输入输出
    input_ds, data_ptr = create_dataset(input_data)
    output_ds = create_output_dataset()
    # 推理
    inference(input_ds, output_ds)
    # 后处理
    cls_id, conf = postprocess(output_ds)
    print(f"Class ID: {cls_id}, Confidence: {conf:.4f}")
    # 释放
    release_resource(input_ds, output_ds, data_ptr)

if __name__ == "__main__":
    main("./test.jpg")