用PyTorch从零实现UNet3+:手把手教你搞定医学图像分割(附完整代码)
用PyTorch从零实现UNet3+:手把手教你搞定医学图像分割(附完整代码)
医学图像分割一直是计算机视觉领域的重要研究方向,尤其在临床诊断和治疗规划中发挥着关键作用。传统的UNet网络虽然表现优异,但在处理多尺度信息融合方面仍有提升空间。UNet3+作为UNet系列的最新演进,通过全尺度跳跃连接和分类引导模块等创新设计,显著提升了分割精度。本文将带你从零开始实现UNet3+,完整解析代码细节,并分享实际训练中的技巧。
1. 环境准备与数据加载
在开始构建UNet3+之前,我们需要配置合适的开发环境。推荐使用Python 3.8+和PyTorch 1.7+版本,这些版本在兼容性和性能方面都有良好表现。
首先安装必要的依赖库:
pip install torch torchvision opencv-python numpy scikit-image对于医学图像数据,通常使用DICOM或NIfTI格式。这里我们以ISIC皮肤病变数据集为例,展示如何构建数据加载器:
import torch from torch.utils.data import Dataset, DataLoader import cv2 import os class MedicalImageDataset(Dataset): def __init__(self, img_dir, mask_dir, transform=None): self.img_dir = img_dir self.mask_dir = mask_dir self.transform = transform self.images = os.listdir(img_dir) def __len__(self): return len(self.images) def __getitem__(self, idx): img_path = os.path.join(self.img_dir, self.images[idx]) mask_path = os.path.join(self.mask_dir, self.images[idx].replace('.jpg', '_mask.gif')) image = cv2.imread(img_path) image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) mask = cv2.imread(mask_path, cv2.IMREAD_GRAYSCALE) if self.transform: augmented = self.transform(image=image, mask=mask) image = augmented['image'] mask = augmented['mask'] image = image.transpose(2, 0, 1).astype('float32') / 255.0 mask = mask.astype('float32') / 255.0 return torch.tensor(image), torch.tensor(mask).unsqueeze(0)提示:医学图像通常需要特定的预处理,如窗宽窗位调整、归一化等,这些操作可以集成到transform中。
2. UNet3+核心架构解析
UNet3+的核心创新在于全尺度跳跃连接和分类引导模块。相比传统UNet只在相同尺度层间跳跃连接,UNet3+实现了编码器和解码器所有层级间的信息流动。
2.1 编码器设计
编码器部分采用经典的卷积+下采样结构,共5个层级:
class EncoderBlock(nn.Module): def __init__(self, in_channels, out_channels): super().__init__() self.conv = nn.Sequential( nn.Conv2d(in_channels, out_channels, 3, padding=1), nn.BatchNorm2d(out_channels), nn.ReLU(inplace=True), nn.Conv2d(out_channels, out_channels, 3, padding=1), nn.BatchNorm2d(out_channels), nn.ReLU(inplace=True) ) self.pool = nn.MaxPool2d(2) def forward(self, x): x = self.conv(x) skip = x # 保存特征图用于跳跃连接 x = self.pool(x) return x, skip2.2 全尺度跳跃连接
这是UNet3+最具创新性的部分,每个解码层接收来自所有编码层的特征:
class FullScaleSkipConnection(nn.Module): def __init__(self, in_channels, out_channels, scale_factor): super().__init__() self.scale_factor = scale_factor if scale_factor > 1: # 需要下采样 self.downsample = nn.MaxPool2d(scale_factor) elif scale_factor < 1: # 需要上采样 self.upsample = nn.Upsample(scale_factor=1/scale_factor, mode='bilinear') self.conv = nn.Sequential( nn.Conv2d(in_channels, out_channels, 3, padding=1), nn.BatchNorm2d(out_channels), nn.ReLU(inplace=True) ) def forward(self, x): if hasattr(self, 'downsample'): x = self.downsample(x) elif hasattr(self, 'upsample'): x = self.upsample(x) return self.conv(x)2.3 分类引导模块(CGM)
CGM模块通过额外的分类任务减少假阳性预测:
class ClassificationGuidedModule(nn.Module): def __init__(self, in_channels): super().__init__() self.classifier = nn.Sequential( nn.Dropout(0.5), nn.Conv2d(in_channels, 2, 1), # 二分类 nn.AdaptiveMaxPool2d(1), nn.Sigmoid() ) def forward(self, x): cls = self.classifier(x) # [B, 2, 1, 1] cls = cls.squeeze(3).squeeze(2) # [B, 2] cls_max = cls.argmax(dim=1).float() # [B] return cls_max.view(-1, 1, 1, 1) # 扩展维度用于后续乘法3. 完整UNet3+实现
将上述组件整合,我们得到完整的UNet3+架构:
class UNet3Plus(nn.Module): def __init__(self, in_channels=3, n_classes=1): super().__init__() filters = [64, 128, 256, 512, 1024] # 编码器 self.encoder1 = EncoderBlock(in_channels, filters[0]) self.encoder2 = EncoderBlock(filters[0], filters[1]) self.encoder3 = EncoderBlock(filters[1], filters[2]) self.encoder4 = EncoderBlock(filters[2], filters[3]) self.encoder5 = EncoderBlock(filters[3], filters[4]) # 分类引导模块 self.cgm = ClassificationGuidedModule(filters[4]) # 解码器 self.decoder1 = self._make_decoder(filters[0], [1, 2, 4, 8, 16]) self.decoder2 = self._make_decoder(filters[0], [2, 1, 2, 4, 8]) self.decoder3 = self._make_decoder(filters[0], [4, 2, 1, 2, 4]) self.decoder4 = self._make_decoder(filters[0], [8, 4, 2, 1, 2]) self.decoder5 = self._make_decoder(filters[0], [16, 8, 4, 2, 1]) # 输出层 self.final_conv = nn.Conv2d(filters[0]*5, n_classes, 1) def _make_decoder(self, out_channels, scale_factors): modules = [] for sf in scale_factors: modules.append(FullScaleSkipConnection(out_channels, out_channels, sf)) return nn.ModuleList(modules) def forward(self, x): # 编码器前向传播 x, skip1 = self.encoder1(x) x, skip2 = self.encoder2(x) x, skip3 = self.encoder3(x) x, skip4 = self.encoder4(x) x, skip5 = self.encoder5(x) # 分类引导 cls_mask = self.cgm(skip5) # 解码器前向传播 dec1 = self._decode(self.decoder1, [skip1, skip2, skip3, skip4, skip5]) dec2 = self._decode(self.decoder2, [skip1, skip2, skip3, skip4, skip5]) dec3 = self._decode(self.decoder3, [skip1, skip2, skip3, skip4, skip5]) dec4 = self._decode(self.decoder4, [skip1, skip2, skip3, skip4, skip5]) dec5 = self._decode(self.decoder5, [skip1, skip2, skip3, skip4, skip5]) # 特征聚合 fused = torch.cat([dec1, dec2, dec3, dec4, dec5], dim=1) output = self.final_conv(fused) # 应用分类引导 return output * cls_mask def _decode(self, decoder, skips): features = [] for i, skip in enumerate(skips): features.append(decoder[i](skip)) return torch.cat(features, dim=1)4. 训练策略与技巧
UNet3+的训练需要特别注意损失函数的选择和学习率调度:
4.1 混合损失函数
论文中提出的混合损失函数结合了三种不同层次的损失:
class HybridLoss(nn.Module): def __init__(self): super().__init__() self.focal = FocalLoss() self.iou = IoULoss() self.ms_ssim = MSSSIM() def forward(self, pred, target): focal_loss = self.focal(pred, target) iou_loss = self.iou(pred, target) ms_ssim_loss = 1 - self.ms_ssim(pred, target) return focal_loss + iou_loss + ms_ssim_loss4.2 学习率调度
医学图像分割通常需要精细调整学习率:
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau( optimizer, mode='min', factor=0.5, patience=5, verbose=True )4.3 数据增强策略
针对医学图像的特性,推荐使用以下增强组合:
import albumentations as A train_transform = A.Compose([ A.RandomRotate90(), A.Flip(), A.ElasticTransform(alpha=120, sigma=120*0.05, alpha_affine=120*0.03), A.GridDistortion(), A.RandomBrightnessContrast(p=0.5), A.Resize(256, 256) ])5. 模型评估与部署
训练完成后,我们需要评估模型性能并准备部署:
5.1 评估指标
除了常见的Dice系数,医学图像分割还关注以下指标:
def calculate_metrics(pred, target): pred = (pred > 0.5).float() target = (target > 0.5).float() tp = (pred * target).sum() fp = (pred * (1-target)).sum() fn = ((1-pred) * target).sum() precision = tp / (tp + fp + 1e-7) recall = tp / (tp + fn + 1e-7) dice = 2 * tp / (2 * tp + fp + fn + 1e-7) return precision, recall, dice5.2 模型量化与加速
为了临床部署,可以使用TorchScript进行模型优化:
# 模型量化 quantized_model = torch.quantization.quantize_dynamic( model, {nn.Conv2d, nn.Linear}, dtype=torch.qint8 ) # 转换为TorchScript traced_script = torch.jit.trace(quantized_model, torch.rand(1, 3, 256, 256)) traced_script.save("unet3plus_quantized.pt")在实际项目中,UNet3+相比传统UNet在边缘细节保持和小目标分割上有明显优势。特别是在处理CT/MRI多模态数据时,全尺度跳跃连接能更好地融合不同层次的特征信息。
