从ViT到UNETR手把手教你用PyTorch和MONAI复现这个医学图像分割的Transformer模型医学图像分割一直是计算机视觉领域的重要研究方向尤其在临床诊断和治疗规划中发挥着关键作用。近年来随着Transformer架构在自然语言处理领域的巨大成功研究者们开始探索将其应用于视觉任务的可能性。UNETRUNEt-TRansformer作为这一探索的重要成果巧妙地将Transformer的全局建模能力与CNN的局部特征提取优势相结合为3D医学图像分割带来了新的突破。本文将带你从零开始使用PyTorch和MONAI框架完整复现UNETR模型。不同于简单的理论讲解我们将重点关注实际代码实现中的关键细节和常见问题确保你能真正掌握这一强大工具。1. 环境准备与数据加载在开始构建模型之前我们需要搭建合适的开发环境并准备数据。医学图像处理有其特殊性3D体积数据如CT、MRI的处理与传统2D图像有很大不同。首先安装必要的Python包pip install torch torchvision monai nibabel tqdm医学图像通常以NIfTI.nii或.nii.gz格式存储我们可以使用MONAI提供的NibabelReader来加载from monai.data import NibabelReader, Dataset, DataLoader from monai.transforms import Compose, LoadImage, AddChannel, ScaleIntensity, RandSpatialCrop, ToTensor transforms Compose([ LoadImage(image_onlyTrue, readerNibabelReader()), AddChannel(), ScaleIntensity(), RandSpatialCrop((96, 96, 96), random_sizeFalse), ToTensor() ]) dataset Dataset( data[{image: path/to/image.nii.gz, label: path/to/label.nii.gz}], transformtransforms ) dataloader DataLoader(dataset, batch_size4, shuffleTrue)注意医学图像数据通常较大建议使用SSD存储并确保有足够的内存。对于显存有限的GPU可以减小batch size或使用梯度累积技术。2. UNETR架构详解与实现UNETR的核心创新在于将Transformer作为编码器同时保留U-Net风格的解码器结构。让我们分模块实现这一架构。2.1 Patch Embedding层Transformer需要将3D体积数据转换为序列输入这一步至关重要import torch import torch.nn as nn class PatchEmbedding3D(nn.Module): def __init__(self, img_size96, patch_size16, in_chans1, embed_dim768): super().__init__() self.img_size (img_size, img_size, img_size) self.patch_size (patch_size, patch_size, patch_size) self.num_patches (img_size // patch_size) ** 3 self.projection nn.Conv3d( in_chans, embed_dim, kernel_sizepatch_size, stridepatch_size ) def forward(self, x): B, C, H, W, D x.shape assert H self.img_size[0] and W self.img_size[1] and D self.img_size[2], \ fInput image size ({H}*{W}*{D}) doesnt match model ({self.img_size[0]}*{self.img_size[1]}*{self.img_size[2]}). x self.projection(x).flatten(2).transpose(1, 2) return x2.2 Transformer编码器实现我们基于ViT架构实现3D版本的Transformer编码器class TransformerBlock(nn.Module): def __init__(self, embed_dim, num_heads, mlp_ratio4.0, dropout0.1): super().__init__() self.norm1 nn.LayerNorm(embed_dim) self.attn nn.MultiheadAttention(embed_dim, num_heads, dropoutdropout) self.norm2 nn.LayerNorm(embed_dim) self.mlp nn.Sequential( nn.Linear(embed_dim, int(embed_dim * mlp_ratio)), nn.GELU(), nn.Dropout(dropout), nn.Linear(int(embed_dim * mlp_ratio), embed_dim), nn.Dropout(dropout) ) def forward(self, x): x x self.attn(self.norm1(x), self.norm1(x), self.norm1(x))[0] x x self.mlp(self.norm2(x)) return x class TransformerEncoder(nn.Module): def __init__(self, embed_dim, depth, num_heads, mlp_ratio4.0, dropout0.1): super().__init__() self.blocks nn.ModuleList([ TransformerBlock(embed_dim, num_heads, mlp_ratio, dropout) for _ in range(depth) ]) def forward(self, x): for blk in self.blocks: x blk(x) return x2.3 CNN解码器与跳跃连接解码器负责将Transformer提取的特征上采样并融合class DecoderBlock(nn.Module): def __init__(self, in_channels, out_channels): super().__init__() self.conv1 nn.Conv3d(in_channels, out_channels, kernel_size3, padding1) self.norm1 nn.InstanceNorm3d(out_channels) self.conv2 nn.Conv3d(out_channels, out_channels, kernel_size3, padding1) self.norm2 nn.InstanceNorm3d(out_channels) self.up nn.ConvTranspose3d(in_channels, out_channels, kernel_size2, stride2) def forward(self, x, skipNone): x self.up(x) if skip is not None: x torch.cat([x, skip], dim1) x self.conv1(x) x self.norm1(x) x nn.ReLU()(x) x self.conv2(x) x self.norm2(x) x nn.ReLU()(x) return x class UNETRDecoder(nn.Module): def __init__(self, in_channels, out_channels, feature_sizes): super().__init__() self.decoder1 DecoderBlock(feature_sizes[0], feature_sizes[1]) self.decoder2 DecoderBlock(feature_sizes[1], feature_sizes[2]) self.decoder3 DecoderBlock(feature_sizes[2], feature_sizes[3]) self.final_conv nn.Conv3d(feature_sizes[3], out_channels, kernel_size1) def forward(self, features): z3, z6, z9, z12 features x self.decoder1(z12, z9) x self.decoder2(x, z6) x self.decoder3(x, z3) x self.final_conv(x) return x3. 完整UNETR模型集成现在我们将各个组件组合成完整的UNETR模型class UNETR(nn.Module): def __init__(self, img_size96, in_chans1, num_classes14, embed_dim768, patch_size16, depth12, num_heads12): super().__init__() self.patch_embed PatchEmbedding3D(img_size, patch_size, in_chans, embed_dim) self.pos_embed nn.Parameter(torch.zeros(1, (img_size//patch_size)**3, embed_dim)) self.transformer TransformerEncoder(embed_dim, depth, num_heads) # Feature projection layers for skip connections self.proj_z3 nn.Sequential( nn.Conv3d(embed_dim, embed_dim//2, kernel_size3, padding1), nn.InstanceNorm3d(embed_dim//2), nn.ReLU() ) self.proj_z6 nn.Sequential( nn.Conv3d(embed_dim, embed_dim//4, kernel_size3, padding1), nn.InstanceNorm3d(embed_dim//4), nn.ReLU() ) self.proj_z9 nn.Sequential( nn.Conv3d(embed_dim, embed_dim//8, kernel_size3, padding1), nn.InstanceNorm3d(embed_dim//8), nn.ReLU() ) self.decoder UNETRDecoder( in_channelsembed_dim, out_channelsnum_classes, feature_sizes[embed_dim//2, embed_dim//4, embed_dim//8, embed_dim//16] ) def forward(self, x): # Patch embedding x self.patch_embed(x) x x self.pos_embed # Transformer encoder x self.transformer(x) # Reshape and project features for skip connections B, N, C x.shape P self.patch_embed.num_patches ** (1/3) P int(P) z12 x.transpose(1, 2).view(B, C, P, P, P) z9 self.proj_z9(z12) z6 self.proj_z6(z12) z3 self.proj_z3(z12) # Decoder x self.decoder([z3, z6, z9, z12]) return x4. 训练策略与优化技巧训练3D医学图像分割模型需要特别注意数据增强和损失函数的选择4.1 数据增强策略使用MONAI提供的丰富医学图像增强方法from monai.transforms import ( RandRotate90, RandFlip, RandAdjustContrast, RandGaussianNoise, RandGaussianSmooth ) train_transforms Compose([ LoadImage(image_onlyTrue, readerNibabelReader()), AddChannel(), ScaleIntensity(), RandRotate90(prob0.5, spatial_axes(0, 1)), RandFlip(prob0.5, spatial_axis0), RandAdjustContrast(prob0.5, gamma(0.8, 1.2)), RandGaussianNoise(prob0.2, std0.01), RandGaussianSmooth(prob0.2, sigma_x(0.5, 1.0)), RandSpatialCrop((96, 96, 96), random_sizeFalse), ToTensor() ])4.2 损失函数与评估指标医学图像分割常用Dice损失和交叉熵损失的组合from monai.losses import DiceLoss, DiceCELoss loss_func DiceCELoss( to_onehot_yTrue, softmaxTrue, squared_predTrue, smooth_nr1e-5, smooth_dr1e-5 ) # 评估指标 from monai.metrics import DiceMetric, HausdorffDistanceMetric dice_metric DiceMetric(include_backgroundFalse, reductionmean) hd_metric HausdorffDistanceMetric(include_backgroundFalse, percentile95)4.3 训练循环实现完整的训练循环需要考虑混合精度训练和梯度累积from torch.cuda.amp import GradScaler, autocast def train_epoch(model, loader, optimizer, loss_func, device, grad_accum2): model.train() total_loss 0 scaler GradScaler() optimizer.zero_grad() for i, batch in enumerate(loader): inputs, labels batch[image].to(device), batch[label].to(device) with autocast(): outputs model(inputs) loss loss_func(outputs, labels) scaler.scale(loss).backward() if (i 1) % grad_accum 0: scaler.step(optimizer) scaler.update() optimizer.zero_grad() total_loss loss.item() return total_loss / len(loader)5. 模型部署与性能优化训练好的模型需要优化才能在临床环境中高效运行5.1 模型量化与剪枝# 动态量化 model torch.quantization.quantize_dynamic( model, {nn.Conv3d, nn.ConvTranspose3d}, dtypetorch.qint8 ) # 剪枝示例 from torch.nn.utils import prune parameters_to_prune [ (module, weight) for module in model.modules() if isinstance(module, nn.Conv3d) ] prune.global_unstructured( parameters_to_prune, pruning_methodprune.L1Unstructured, amount0.2 )5.2 ONNX导出与TensorRT优化# 导出ONNX模型 dummy_input torch.randn(1, 1, 96, 96, 96, devicecuda) torch.onnx.export( model, dummy_input, unetr.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch}, output: {0: batch}} ) # TensorRT优化 (需要安装torch-tensorrt) import torch_tensorrt trt_model torch_tensorrt.compile( model, inputs[torch_tensorrt.Input((1, 1, 96, 96, 96), dtypetorch.float32)], enabled_precisions{torch.float32} )在实际项目中我们发现将patch size从16减小到12可以在保持性能的同时显著降低显存占用。对于显存有限的GPU如24GB这是训练更大体积如128×128×128的关键技巧。