---

• CANN官网： https://www.hiascend.com/cann

项目背景

在肺部疾病筛查与诊断领域，CT影像分析是临床诊断的重要依据。某三甲医院每天产生超过2000例肺部CT影像数据，传统的人工阅片方式存在效率低、漏诊率高、受医生经验影响大等问题。尤其在早期肺癌筛查中，直径小于5mm的小结节极易被漏诊，而早期发现对患者预后至关重要。

为提升肺部疾病诊断效率和准确性，该医院与华为合作，基于CANN（神经网络异构计算架构）构建了一套肺部CT智能辅助诊断系统，实现了肺结节的自动检测、良恶性鉴别和结构化报告生成。

项目挑战

在系统开发过程中，团队面临以下核心挑战：

• 数据规模大：单例肺部CT包含300-500层图像，每层512×512像素，每天2000例数据量达数百GB。
• 模型复杂度高：3D卷积神经网络模型参数量超过100M，计算复杂度高，传统计算平台难以满足实时性要求。
• 精度要求严格：医疗诊断对模型精度要求极高，肺结节检测召回率需达到95%以上，假阳性率需控制在5个/例以下。
• 部署环境多样：需要在不同环境（医院数据中心、科室工作站、移动设备）灵活部署。

系统整体架构

系统采用基于CANN的分层异构计算架构，充分发挥昇腾AI处理器的计算优势：

核心架构组件：

• 数据采集层：负责DICOM格式CT影像数据的采集、传输和存储
• 预处理层：基于CANN的图像处理加速库，实现CT图像的去噪、增强、标准化等预处理
• 计算引擎层：采用昇腾910和昇腾310处理器混合部署，实现模型训练和推理加速
• 模型层：包含3D CNN肺结节检测模型、特征提取模型和良恶性鉴别模型
• 应用层：提供医生工作站、报告系统和远程会诊接口

通过CANN的统一编程接口AscendCL，实现了从底层硬件到上层应用的无缝对接，确保了系统的高效运行和灵活部署。

CANN技术应用方案

1. 异构计算资源调度

系统基于CANN的Device管理能力，实现了对昇腾910和昇腾310处理器的统一管理和调度：

import ascendcl as acldl
import threading
import queue

class DeviceManager:
    def __init__(self):
        # 初始化CANN环境
        self.acl_config = acldl.AclConfig()
        self.acl_config.init()
        
        # 获取可用设备数量
        self.device_count = self.acl_config.get_device_count()
        
        # 创建设备上下文池
        self.device_contexts = []
        self.context_locks = []
        
        for i in range(self.device_count):
            context = acldl.AclContext()
            context.create(i)
            self.device_contexts.append(context)
            self.context_locks.append(threading.Lock())
            
        # 任务队列
        self.task_queue = queue.Queue()
        self.workers = []
        
        # 启动工作线程
        for i in range(self.device_count):
            worker = threading.Thread(target=self._process_task, args=(i,))
            worker.daemon = True
            worker.start()
            self.workers.append(worker)
    
    def submit_task(self, task_type, data, callback=None):
        # 提交任务到队列
        task = (task_type, data, callback)
        self.task_queue.put(task)
    
    def _process_task(self, device_id):
        # 处理任务
        while True:
            task_type, data, callback = self.task_queue.get()
            
            # 加锁确保设备上下文的独占访问
            with self.context_locks[device_id]:
                try:
                    # 切换到当前设备上下文
                    self.device_contexts[device_id].set_current()
                    
                    # 根据任务类型处理数据
                    if task_type == "preprocess":
                        result = self._preprocess(data)
                    elif task_type == "inference":
                        result = self._inference(data)
                    elif task_type == "postprocess":
                        result = self._postprocess(data)
                    
                    # 执行回调函数
                    if callback:
                        callback(result)
                except Exception as e:
                    print(f"Error processing task on device {device_id}: {e}")
                finally:
                    # 标记任务完成
                    self.task_queue.task_done()
    
    def _preprocess(self, data):
        # 图像预处理实现
        # ...
        return processed_data
    
    def _inference(self, data):
        # 模型推理实现
        # ...
        return inference_result
    
    def _postprocess(self, data):
        # 后处理实现
        # ...
        return postprocessed_result
    
    def __del__(self):
        # 释放资源
        for context in self.device_contexts:
            context.destroy()
        self.acl_config.finalize()

2. 模型优化与加速

针对3D CNN模型计算复杂度高的问题，团队利用CANN的算子优化和模型压缩技术进行了深度优化：

import ascendcl as acldl
import numpy as np
import torch
from torch import nn

class LungNoduleModel:
    def __init__(self, model_path, device_id=0):
        # 初始化CANN环境
        self.acl_config = acldl.AclConfig()
        self.acl_config.init()
        
        # 创建设备上下文
        self.context = acldl.AclContext()
        self.context.create(device_id)
        self.context.set_current()
        
        # 加载优化后的模型
        self.model = acldl.AclModel(model_path)
        
        # 获取模型输入输出信息
        self.input_desc = self.model.get_input_desc(0)
        self.output_desc = self.model.get_output_desc(0)
        
        # 创建输入输出缓冲区
        self.input_buf = acldl.AclBuffer(self.input_desc.get_size())
        self.output_buf = acldl.AclBuffer(self.output_desc.get_size())
        
        print(f"LungNoduleModel initialized on device {device_id}")
    
    def preprocess(self, ct_data):
        # CT数据预处理
        # 1. 窗宽窗位调整
        ct_data = self._adjust_window(ct_data, window_center=-600, window_width=1500)
        
        # 2. 归一化
        ct_data = (ct_data - np.mean(ct_data)) / np.std(ct_data)
        
        # 3. 尺寸调整为模型输入大小
        ct_data = self._resize(ct_data, target_shape=(1, 1, 64, 64, 64))
        
        # 4. 数据格式转换
        ct_data = ct_data.astype(np.float32)
        return ct_data
    
    def _adjust_window(self, data, window_center, window_width):
        # 窗宽窗位调整
        min_value = window_center - window_width // 2
        max_value = window_center + window_width // 2
        data = np.clip(data, min_value, max_value)
        data = (data - min_value) / (max_value - min_value)
        return data
    
    def _resize(self, data, target_shape):
        # 图像尺寸调整
        # ... (实现3D图像resize逻辑)
        return resized_data
    
    def inference(self, preprocessed_data):
        # 执行模型推理
        # 1. 将数据拷贝到设备内存
        self.input_buf.copy_from_host(preprocessed_data.tobytes())
        
        # 2. 执行推理计算
        self.model.execute([self.input_buf], [self.output_buf])
        
        # 3. 从设备内存读取结果
        output_data = np.frombuffer(self.output_buf.copy_to_host(), dtype=np.float32)
        
        # 4. 解析输出结果
        output_shape = self.output_desc.get_shape()
        output_data = output_data.reshape(output_shape)
        
        return output_data
    
    def postprocess(self, output_data, threshold=0.7):
        # 后处理，提取肺结节信息
        nodules = []
        
        # 遍历输出结果，提取置信度大于阈值的结节
        for i in range(output_data.shape[1]):
            for j in range(output_data.shape[2]):
                for k in range(output_data.shape[3]):
                    prob = output_data[0, i, j, k, 0]
                    if prob > threshold:
                        # 计算结节的世界坐标
                        # ... (坐标转换逻辑)
                        
                        # 提取结节特征
                        # ... (特征提取逻辑)
                        
                        nodules.append({
                            'position': (i, j, k),
                            'probability': float(prob),
                            'size': 5.0,  # 示例值
                            'features': nodule_features
                        })
        
        return nodules
    
    def process(self, ct_data):
        # 完整处理流程
        preprocessed_data = self.preprocess(ct_data)
        output_data = self.inference(preprocessed_data)
        nodules = self.postprocess(output_data)
        return nodules
    
    def __del__(self):
        # 释放资源
        self.input_buf.destroy()
        self.output_buf.destroy()
        self.context.destroy()
        self.acl_config.finalize()

3. 基于TBE的自定义算子开发

为进一步提升系统性能，团队针对3D卷积等计算密集型操作，利用CANN的TBE（Tensor Boost Engine）开发框架，开发了专用的优化算子：

import tbe
from tbe import tvm
from tbe.common.register import register_op_compute
from tbe.common.utils import shape_util

@register_op_compute("optimized_3d_conv")
def optimized_3d_conv_compute(input_x, weight, output_y, 
                             stride, pad, dilation, 
                             groups, dtype="float32", 
                             kernel_name="optimized_3d_conv"):
    # 定义3D卷积算子计算逻辑
    # 1. 获取输入和权重的形状信息
    shape_input = shape_util.shape_to_list(input_x.shape)
    shape_weight = shape_util.shape_to_list(weight.shape)
    
    # 2. 检查输入参数合法性
    if groups != 1:
        raise ValueError("Only groups=1 is supported currently")
    
    # 3. 计算输出形状
    n, c, d, h, w = shape_input
    kd, kh, kw, ic, oc = shape_weight
    
    # 考虑padding和stride计算输出维度
    out_d = (d + 2 * pad[0] - kd) // stride[0] + 1
    out_h = (h + 2 * pad[1] - kh) // stride[1] + 1
    out_w = (w + 2 * pad[2] - kw) // stride[2] + 1
    
    # 4. 实现优化的3D卷积计算
    # 使用TVM的计算表达式描述3D卷积
    # 针对昇腾AI处理器的特性进行优化
    
    # 4.1 输入数据填充
    if sum(pad) > 0:
        padded_input = tbe.tdnn.pad(input_x, pad, pad_mode="constant", pad_value=0.0)
    else:
        padded_input = input_x
    
    # 4.2 执行3D卷积计算
    # 使用滑动窗口和矩阵乘法相结合的优化策略
    # 充分利用昇腾AI Core的矩阵计算能力
    k = tvm.reduce_axis((0, ic), name='k')
    kd_axis = tvm.reduce_axis((0, kd), name='kd')
    kh_axis = tvm.reduce_axis((0, kh), name='kh')
    kw_axis = tvm.reduce_axis((0, kw), name='kw')
    
    conv = tvm.compute(
        (n, oc, out_d, out_h, out_w),
        lambda n, oc, d, h, w: tvm.sum(
            padded_input[n, k, d*stride[0]+kd_axis, h*stride[1]+kh_axis, w*stride[2]+kw_axis].astype(dtype) *
            weight[kd_axis, kh_axis, kw_axis, k, oc].astype(dtype),
            axis=[k, kd_axis, kh_axis, kw_axis]
        ),
        name="conv"
    )
    
    # 5. 返回计算结果
    return conv

# 注册算子调度策略
@register_op_schedule("optimized_3d_conv")
def optimized_3d_conv_schedule(attrs):
    # 定义算子调度策略
    # 针对3D卷积的特点进行调度优化
    # ... (调度策略实现)
    return sch

# 注册算子原型
@register_op_proto("optimized_3d_conv")
def optimized_3d_conv_proto():
    # 定义算子原型
    # ... (原型定义)

4. 内存优化与并行处理

系统利用CANN的内存管理和Stream机制，实现了数据的高效处理和任务的并行执行：

import ascendcl as acldl
import numpy as np

class MemoryOptimizer:
    def __init__(self, device_id=0):
        # 初始化CANN环境
        self.acl_config = acldl.AclConfig()
        self.acl_config.init()
        
        # 创建设备上下文
        self.context = acldl.AclContext()
        self.context.create(device_id)
        self.context.set_current()
        
        # 创建Stream用于异步执行
        self.stream = acldl.AclStream()
        self.stream.create()
        
        # 内存池管理
        self.memory_pool = {}
    
    def alloc_memory(self, size, memory_type="device"):
        # 分配内存
        if size in self.memory_pool:
            # 如果内存池中已有合适大小的内存块，直接复用
            return self.memory_pool[size].pop()
        else:
            # 否则创建新的内存块
            if memory_type == "device":
                return acldl.AclBuffer(size)
            elif memory_type == "host":
                return np.zeros(size, dtype=np.byte)
    
    def free_memory(self, buffer, size):
        # 释放内存到内存池
        if size not in self.memory_pool:
            self.memory_pool[size] = []
        self.memory_pool[size].append(buffer)
    
    def copy_data_async(self, src, dst, size):
        # 异步数据拷贝
        if isinstance(src, np.ndarray) and isinstance(dst, acldl.AclBuffer):
            # 主机到设备
            return dst.copy_from_host_async(src.tobytes(), size, self.stream)
        elif isinstance(src, acldl.AclBuffer) and isinstance(dst, np.ndarray):
            # 设备到主机
            return src.copy_to_host_async(dst.tobytes(), size, self.stream)
    
    def synchronize(self):
        # 同步Stream
        self.stream.synchronize()
    
    def __del__(self):
        # 释放所有资源
        for buffers in self.memory_pool.values():
            for buffer in buffers:
                if isinstance(buffer, acldl.AclBuffer):
                    buffer.destroy()
        
        self.stream.destroy()
        self.context.destroy()
        self.acl_config.finalize()

技术创新点

1. 3D卷积优化技术

团队针对3D CNN模型的特点，利用CANN的TBE开发框架，设计了优化的3D卷积算子。通过以下技术手段提升性能：

• 分块计算：将3D卷积分解为多个2D卷积操作，充分利用昇腾AI处理器的2D矩阵计算能力
• 权值重排：优化卷积核的存储格式，提高内存访问效率
• 数据预取：通过预取数据到缓存，减少内存访问延迟

优化后的3D卷积算子计算效率提升了2.5倍，模型推理速度提升了1.8倍。

2. 混合精度计算

系统采用CANN的混合精度计算功能，在保证模型精度的前提下，提升计算效率：

• 前向传播使用FP16精度，减少内存占用和计算量
• 关键层和反向传播使用FP32精度，确保训练稳定性和模型精度
• 使用CANN提供的精度校准工具，对量化误差进行补偿

混合精度计算使模型训练速度提升了1.5倍，推理速度提升了1.3倍，同时内存占用减少了50%。

3. 多任务并行处理

基于CANN的Stream机制，系统实现了多任务的并行处理：

• 图像预处理、模型推理和后处理在不同Stream中并行执行
• 多个患者的CT数据在不同设备上并行处理
• 同一患者的不同部位数据在同一设备上的不同Stream中并行处理

多任务并行处理使系统整体吞吐量提升了2倍以上，单例CT的处理时间从原来的45秒缩短至18秒。

项目成效

通过CANN技术的深度应用，肺部CT智能辅助诊断系统取得了显著成效：

• 诊断效率大幅提升：单例肺部CT的处理时间从人工阅片的10-15分钟缩短至18秒，每天2000例CT影像的处理时间从原来的10天缩短至2小时以内。
• 诊断准确性显著提高：肺结节检测召回率达到97.3%，假阳性率控制在3.2个/例以下，相比人工阅片的92%召回率有明显提升。
• 医生工作强度降低：医生从繁琐的影像细节排查中解放出来，专注于疑难病例的诊断和治疗方案制定，工作效率提升了40%以上。
• 早期肺癌检出率提升：直径小于5mm的肺小结节检出率从原来的75%提升至95%以上，为患者争取了宝贵的治疗时间。

总结

• 该肺部CT智能辅助诊断系统的成功实践，充分展示了CANN在医疗AI领域的巨大潜力。通过CANN的高性能计算能力、灵活的适配特性和丰富的优化工具，项目团队成功解决了传统医疗影像分析中面临的效率低、精度差、成本高等问题，为医疗行业的智能化转型提供了可复制、可推广的解决方案。

• 随着CANN技术的不断发展和完善，未来将在更多医疗场景中发挥重要作用，如多模态医学影像分析、智能手术规划、医疗资源优化等。特别是在当前背景下，CANN 作为自主创新的异构计算架构，将为我国医疗 AI 的发展提供有力支撑。

正如项目负责人所说：“CANN技术不仅提升了我们的诊断效率和准确性，更重要的是，它为医生提供了一个强大的辅助工具，使他们能够为患者提供更好的医疗服务。我们相信，随着CANN生态的不断完善，将会有更多创新应用在医疗领域落地。”

---