从MaskFormer到MpFormer：手把手教你用PyTorch复现这三个图像分割模型（附代码）

从MaskFormer到MpFormer：PyTorch实战图像分割模型全解析

在计算机视觉领域，图像分割一直是核心任务之一。近年来，基于Transformer的分割模型如MaskFormer系列展现出强大的性能。本文将带您从零开始，用PyTorch实现MaskFormer、Mask2Former和MpFormer三大模型，深入剖析代码细节与实现技巧。

1. 环境配置与数据准备

1.1 基础环境搭建

首先确保您的环境满足以下要求：

conda create -n segmentation python=3.8 conda activate segmentation pip install torch==1.12.1+cu113 torchvision==0.13.1+cu113 -f https://download.pytorch.org/whl/torch_stable.html pip install mmcv-full==1.6.1 -f https://download.openmmlab.com/mmcv/dist/cu113/torch1.12/index.html

关键依赖版本对照表：

1.2 数据集处理

以COCO数据集为例，我们需要实现特殊的数据加载方式：

from torch.utils.data import Dataset import pycocotools.mask as mask_utils class COCOPanopticDataset(Dataset): def __init__(self, root, annFile, transforms=None): self.coco = COCO(annFile) self.root = root self.transforms = transforms self.ids = list(sorted(self.coco.imgs.keys())) def __getitem__(self, idx): img_id = self.ids[idx] ann_ids = self.coco.getAnnIds(imgIds=img_id) anns = self.coco.loadAnns(ann_ids) # 处理mask和类别标签 masks = [self.coco.annToMask(ann) for ann in anns] masks = np.stack(masks, axis=0) labels = [ann['category_id'] for ann in anns] # 应用数据增强 if self.transforms: augmented = self.transforms(image=img, masks=masks) img = augmented['image'] masks = augmented['masks'] return img, masks, labels

注意：COCO数据集需要特殊处理全景标注格式，建议使用官方提供的pycocotools工具包

2. MaskFormer核心实现

2.1 模型架构分解

MaskFormer由三个关键模块组成：

1. 像素级模块：基于FPN的轻量级像素解码器
2. Transformer模块：标准解码器结构
3. 分割模块：分类头与mask预测头

class MaskFormer(nn.Module): def __init__(self, backbone, transformer, num_queries=100): super().__init__() self.backbone = backbone # 通常是ResNet或SwinTransformer self.pixel_decoder = FPN(backbone.out_channels) self.transformer = transformer self.query_embed = nn.Embedding(num_queries, transformer.d_model) self.class_embed = nn.Linear(transformer.d_model, num_classes + 1) self.mask_embed = MLP(transformer.d_model, transformer.d_model, 256, 3) def forward(self, x): # 特征提取 features = self.backbone(x) pixel_embeddings = self.pixel_decoder(features) # Transformer处理 query_pos = self.query_embed.weight hs = self.transformer(pixel_embeddings.flatten(2).permute(2,0,1), query_pos.unsqueeze(1).repeat(1,x.size(0),1)) # 预测输出 outputs_class = self.class_embed(hs) mask_embeds = self.mask_embed(hs) outputs_mask = torch.einsum("bqc,bchw->bqhw", mask_embeds, pixel_embeddings) return {"pred_logits": outputs_class[-1], "pred_masks": outputs_mask[-1]}

2.2 损失函数实现

MaskFormer使用二分图匹配损失，关键实现如下：

class SetCriterion(nn.Module): def __init__(self, num_classes, matcher): super().__init__() self.num_classes = num_classes self.matcher = matcher self.loss_fns = { "labels": nn.CrossEntropyLoss(), "masks": nn.BCEWithLogitsLoss() } def forward(self, outputs, targets): # 执行二分图匹配 indices = self.matcher(outputs, targets) # 计算分类损失 src_logits = outputs["pred_logits"] idx = self._get_src_permutation_idx(indices) target_classes = torch.cat([t["labels"][J] for t, (_, J) in zip(targets, indices)]) loss_ce = self.loss_fns["labels"](src_logits[idx], target_classes) # 计算mask损失 src_masks = outputs["pred_masks"] target_masks = torch.cat([t["masks"][J] for t, (_, J) in zip(targets, indices)]) loss_mask = self.loss_fns["masks"](src_masks[idx], target_masks) return {"loss_ce": loss_ce, "loss_mask": loss_mask}

3. Mask2Former进阶实现

3.1 Masked Attention机制

Mask2Former的核心改进在于masked attention的实现：

class MaskedAttention(nn.Module): def __init__(self, dim, num_heads=8): super().__init__() self.num_heads = num_heads self.scale = (dim // num_heads) ** -0.5 self.qkv = nn.Linear(dim, dim * 3) self.proj = nn.Linear(dim, dim) def forward(self, x, mask=None): B, N, C = x.shape qkv = self.qkv(x).reshape(B, N, 3, self.num_heads, C // self.num_heads) q, k, v = qkv.unbind(2) attn = (q @ k.transpose(-2, -1)) * self.scale if mask is not None: attn = attn.masked_fill(mask.unsqueeze(1).unsqueeze(2) == 0, float('-inf')) attn = attn.softmax(dim=-1) x = (attn @ v).transpose(1, 2).reshape(B, N, C) return self.proj(x)

3.2 多尺度特征处理

高分辨率特征金字塔的实现技巧：

class MultiScaleFeatureFusion(nn.Module): def __init__(self, in_channels, out_channels): super().__init__() self.lateral_convs = nn.ModuleList() self.fpn_convs = nn.ModuleList() for i in range(len(in_channels)): self.lateral_convs.append( nn.Conv2d(in_channels[i], out_channels, 1)) self.fpn_convs.append( nn.Sequential( nn.Conv2d(out_channels, out_channels, 3, padding=1), nn.GroupNorm(32, out_channels), nn.ReLU(inplace=True) )) def forward(self, features): laterals = [conv(f) for conv, f in zip(self.lateral_convs, features)] # 自顶向下路径 used_backbone_levels = len(laterals) for i in range(used_backbone_levels - 1, 0, -1): laterals[i - 1] += F.interpolate( laterals[i], scale_factor=2, mode="nearest") # 特征融合 outs = [self.fpn_convs[i](laterals[i]) for i in range(used_backbone_levels)] return outs

4. MpFormer创新实现

4.1 GT Mask注入机制

MpFormer在训练时引入GT mask作为额外监督：

class MPFormerDecoderLayer(nn.Module): def __init__(self, d_model, nhead, dim_feedforward=2048): super().__init__() self.self_attn = nn.MultiheadAttention(d_model, nhead) self.multihead_attn = nn.MultiheadAttention(d_model, nhead) self.linear1 = nn.Linear(d_model, dim_feedforward) self.linear2 = nn.Linear(dim_feedforward, d_model) self.norm1 = nn.LayerNorm(d_model) self.norm2 = nn.LayerNorm(d_model) self.norm3 = nn.LayerNorm(d_model) self.dropout = nn.Dropout(0.1) def forward(self, tgt, memory, tgt_mask=None, memory_mask=None, tgt_key_padding_mask=None, memory_key_padding_mask=None): # 自注意力 tgt2 = self.self_attn(tgt, tgt, tgt, attn_mask=tgt_mask, key_padding_mask=tgt_key_padding_mask)[0] tgt = tgt + self.dropout(tgt2) tgt = self.norm1(tgt) # 交叉注意力（注入GT mask） tgt2 = self.multihead_attn(tgt, memory, memory, attn_mask=memory_mask, key_padding_mask=memory_key_padding_mask)[0] tgt = tgt + self.dropout(tgt2) tgt = self.norm2(tgt) # FFN tgt2 = self.linear2(self.dropout(F.relu(self.linear1(tgt)))) tgt = tgt + self.dropout(tgt2) tgt = self.norm3(tgt) return tgt

4.2 噪声注入策略

MpFormer通过添加噪声增强鲁棒性：

def add_noise_to_mask(mask, noise_type="point", noise_ratio=0.1): """ 为GT mask添加噪声 Args: mask: 原始mask [H, W] noise_type: point | flip noise_ratio: 噪声比例 """ if noise_type == "point": # 随机选择部分点置反 h, w = mask.shape num_noise = int(h * w * noise_ratio) coords = torch.randint(0, h*w, (num_noise,)) noisy_mask = mask.flatten() noisy_mask[coords] = 1 - noisy_mask[coords] return noisy_mask.reshape(h, w) elif noise_type == "flip": # 随机翻转类别 if random.random() < noise_ratio: return 1 - mask return mask else: return mask

5. 训练与优化技巧

5.1 训练脚本配置

完整的训练流程实现：

def train_one_epoch(model, criterion, data_loader, optimizer, device, epoch): model.train() metric_logger = MetricLogger(delimiter=" ") header = f"Epoch: [{epoch}]" for images, targets in metric_logger.log_every(data_loader, 10, header): images = images.to(device) targets = [{k: v.to(device) for k, v in t.items()} for t in targets] # 前向传播 outputs = model(images) # 计算损失 loss_dict = criterion(outputs, targets) losses = sum(loss_dict.values()) # 反向传播 optimizer.zero_grad() losses.backward() optimizer.step() # 记录指标 metric_logger.update(loss=losses.item(), **loss_dict) return metric_logger def main(): # 初始化模型 backbone = build_backbone(config) transformer = build_transformer(config) model = MaskFormer(backbone, transformer, config.num_queries) # 数据加载 dataset = COCOPanopticDataset(config.data_path, config.ann_path) data_loader = DataLoader(dataset, batch_size=4, shuffle=True) # 优化器配置 param_dicts = [ {"params": [p for n, p in model.named_parameters() if "backbone" not in n and p.requires_grad]}, {"params": [p for n, p in model.named_parameters() if "backbone" in n and p.requires_grad], "lr": config.lr_backbone} ] optimizer = torch.optim.AdamW(param_dicts, lr=config.lr, weight_decay=config.weight_decay) # 训练循环 for epoch in range(config.epochs): train_one_epoch(model, criterion, data_loader, optimizer, device, epoch)

5.2 关键超参数设置

不同模型的推荐配置：

6. 推理与可视化

6.1 推理脚本实现

@torch.no_grad() def inference(model, image_path, device): # 图像预处理 image = Image.open(image_path).convert("RGB") transform = T.Compose([ T.Resize(800), T.ToTensor(), T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) ]) img_tensor = transform(image).unsqueeze(0).to(device) # 模型推理 outputs = model(img_tensor) # 后处理 prob = outputs["pred_logits"].softmax(-1)[..., :-1] masks = outputs["pred_masks"].sigmoid() # 获取最终预测 scores, labels = prob.max(-1) keep = scores > 0.5 return masks[keep], labels[keep] def visualize(image_path, masks, labels): image = cv2.imread(image_path) image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) for mask, label in zip(masks, labels): color = np.random.randint(0, 255, size=3) mask = mask.cpu().numpy() image[mask > 0.5] = image[mask > 0.5] * 0.5 + color * 0.5 plt.imshow(image) plt.show()

6.2 性能优化技巧

提升推理速度的实用方法：

1. 半精度推理：

model.half() # 转换为半精度 with torch.cuda.amp.autocast(): outputs = model(img_tensor.half())

2. TensorRT加速：

# 转换模型为ONNX格式 torch.onnx.export(model, img_tensor, "model.onnx", input_names=["input"], output_names=["output"]) # 使用TensorRT优化 trt_model = torch2trt(model, [img_tensor], fp16_mode=True)

3. 多尺度测试技巧：

def multi_scale_inference(model, image, scales=[0.5, 1.0, 1.5]): results = [] for scale in scales: h, w = image.shape[-2:] resized_img = F.interpolate(image, scale_factor=scale, mode="bilinear") outputs = model(resized_img) outputs["pred_masks"] = F.interpolate( outputs["pred_masks"], size=(h, w), mode="bilinear") results.append(outputs) # 融合多尺度结果 final_output = { "pred_logits": torch.mean(torch.stack([r["pred_logits"] for r in results]), 0), "pred_masks": torch.mean(torch.stack([r["pred_masks"] for r in results]), 0) } return final_output