图像分割技术：像素级的精准识别（superior哥深度学习系列第12期）

12_图像分割技术：像素级的精准识别

superior哥深度学习系列第十二篇
从像素到语义，从分割到理解——探索计算机视觉的精细化世界

🎯 前言：当AI学会"精雕细琢"

各位小伙伴们，欢迎来到superior哥深度学习系列的第十二篇！前面我们学习了图像分类和目标检测，今天我们要进入一个更加精细的领域——图像分割。

如果说图像分类是让AI知道"这是什么"，目标检测是让AI知道"什么在哪里"，那么图像分割就是让AI知道"每个像素属于什么"。这是计算机视觉中最精细的任务之一，需要AI对图像进行像素级的理解和标注。

想象一下，当你看到一张街景图片时，你不仅能识别出汽车、行人、建筑物，还能准确地指出每个像素属于哪个物体。这就是图像分割要解决的问题！

📊 知识架构图

图像分割技术
├── 分割类型
│   ├── 语义分割 (Semantic Segmentation)
│   ├── 实例分割 (Instance Segmentation)  
│   └── 全景分割 (Panoptic Segmentation)
├── 经典算法
│   ├── FCN (全卷积网络)
│   ├── U-Net (医学图像分割)
│   ├── DeepLab (空洞卷积)
│   └── Mask R-CNN (实例分割)
├── 关键技术
│   ├── 编码器-解码器结构
│   ├── 跳跃连接 (Skip Connection)
│   ├── 空洞卷积 (Dilated Convolution)
│   └── 特征金字塔网络 (FPN)
├── 损失函数与评估
│   ├── 交叉熵损失
│   ├── Dice损失
│   ├── IoU指标
│   └── 像素准确率
└── 实战应用
    ├── 医学图像分割
    ├── 自动驾驶
    ├── 图像编辑
    └── 机器人视觉

🧠 第一章：图像分割基础理论

1.1 什么是图像分割？

图像分割是将图像分成若干个语义区域的过程，使得每个区域内的像素具有相似的特征（如颜色、纹理、亮度等）。在深度学习时代，图像分割主要分为三种类型：

🎨 语义分割 (Semantic Segmentation)

• 目标：为每个像素分配一个类别标签
• 特点：同一类别的不同实例不区分
• 例子：将所有的"人"像素都标记为"人"类别

🎭 实例分割 (Instance Segmentation)

• 目标：区分同一类别的不同实例
• 特点：不仅分类，还要区分个体
• 例子：将图片中的"人1"、“人2”、"人3"分别标记

🌈 全景分割 (Panoptic Segmentation)

• 目标：结合语义分割和实例分割
• 特点：对"物体"类别进行实例分割，对"背景"类别进行语义分割
• 例子：区分不同的车辆，但不区分不同的天空区域

1.2 图像分割的挑战

import torch
import torch.nn as nn
import torch.nn.functional as F
import torchvision
import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
import cv2

# 展示图像分割的挑战
def visualize_segmentation_challenges():
    """
    可视化图像分割面临的主要挑战
    """
    # 创建示例图像
    fig, axes = plt.subplots(2, 3, figsize=(15, 10))
    
    challenges = [
        "边界模糊", "尺度变化", "遮挡问题",
        "类别不平衡", "细节丢失", "复杂背景"
    ]
    
    for i, (ax, challenge) in enumerate(zip(axes.flat, challenges)):
        # 这里用随机图像代替实际挑战图像
        img = np.random.rand(100, 100, 3)
        ax.imshow(img)
        ax.set_title(f"挑战{i+1}: {challenge}", fontsize=12)
        ax.axis('off')
    
    plt.tight_layout()
    plt.savefig('segmentation_challenges.png', dpi=150, bbox_inches='tight')
    plt.show()

# 运行可视化
visualize_segmentation_challenges()

🔧 第二章：语义分割核心算法

2.1 FCN：全卷积网络的开创性工作

FCN (Fully Convolutional Network) 是深度学习时代语义分割的开山之作，它将传统的CNN分类网络改造成端到端的分割网络。

import torch
import torch.nn as nn
import torch.nn.functional as F

class FCN(nn.Module):
    """
    FCN-32s/16s/8s 实现
    基于VGG16backbone的全卷积网络
    """
    def __init__(self, num_classes=21, backbone='vgg16', pretrained=True):
        super(FCN, self).__init__()
        self.num_classes = num_classes
        
        # 使用VGG16作为backbone
        if backbone == 'vgg16':
            vgg = torchvision.models.vgg16(pretrained=pretrained)
            features = list(vgg.features.children())
            
            # 编码器部分 - 提取特征
            self.stage1 = nn.Sequential(*features[:10])   # pool1
            self.stage2 = nn.Sequential(*features[10:17]) # pool2  
            self.stage3 = nn.Sequential(*features[17:24]) # pool3
            self.stage4 = nn.Sequential(*features[24:31]) # pool4
            self.stage5 = nn.Sequential(*features[31:])   # pool5
            
            # 分类器改为全卷积
            self.classifier = nn.Sequential(
                nn.Conv2d(512, 4096, kernel_size=7, padding=3),
                nn.ReLU(inplace=True),
                nn.Dropout2d(0.5),
                nn.Conv2d(4096, 4096, kernel_size=1),
                nn.ReLU(inplace=True),
                nn.Dropout2d(0.5),
                nn.Conv2d(4096, num_classes, kernel_size=1)
            )
            
            # 上采样层
            self.upsample_32s = nn.ConvTranspose2d(
                num_classes, num_classes, kernel_size=64, stride=32, padding=16
            )
            self.upsample_16s = nn.ConvTranspose2d(
                num_classes, num_classes, kernel_size=32, stride=16, padding=8
            )
            self.upsample_8s = nn.ConvTranspose2d(
                num_classes, num_classes, kernel_size=16, stride=8, padding=4
            )
            
            # 跳跃连接的1x1卷积
            self.score_pool4 = nn.Conv2d(512, num_classes, kernel_size=1)
            self.score_pool3 = nn.Conv2d(256, num_classes, kernel_size=1)
    
    def forward(self, x, mode='fcn32s'):
        """
        前向传播
        mode: 'fcn32s', 'fcn16s', 'fcn8s'
        """
        input_size = x.size()[2:]
        
        # 编码器
        pool1 = self.stage1(x)    # 1/2
        pool2 = self.stage2(pool1) # 1/4  
        pool3 = self.stage3(pool2) # 1/8
        pool4 = self.stage4(pool3) # 1/16
        pool5 = self.stage5(pool4) # 1/32
        
        # 分类器
        score = self.classifier(pool5)
        
        if mode == 'fcn32s':
            # FCN-32s: 直接32倍上采样
            output = self.upsample_32s(score)
            
        elif mode == 'fcn16s':
            # FCN-16s: 融合pool4特征
            score_pool4 = self.score_pool4(pool4)
            score = F.interpolate(score, size=score_pool4.size()[2:], 
                                mode='bilinear', align_corners=False)
            score = score + score_pool4
            output = self.upsample_16s(score)
            
        elif mode == 'fcn8s':
            # FCN-8s: 融合pool4和pool3特征
            score_pool4 = self.score_pool4(pool4)
            score_pool3 = self.score_pool3(pool3)
            
            # 先融合pool4
            score = F.interpolate(score, size=score_pool4.size()[2:], 
                                mode='bilinear', align_corners=False)
            score = score + score_pool4
            
            # 再融合pool3
            score = F.interpolate(score, size=score_pool3.size()[2:], 
                                mode='bilinear', align_corners=False)
            score = score + score_pool3
            
            output = self.upsample_8s(score)
        
        # 调整到输入尺寸
        output = F.interpolate(output, size=input_size, 
                             mode='bilinear', align_corners=False)
        
        return output

# FCN训练类
class FCNTrainer:
    def __init__(self, model, device='cuda'):
        self.model = model
        self.device = device
        self.model.to(device)
        
        # 损失函数和优化器
        self.criterion = nn.CrossEntropyLoss(ignore_index=255)
        self.optimizer = torch.optim.SGD(
            model.parameters(), lr=1e-3, momentum=0.9, weight_decay=5e-4
        )
        self.scheduler = torch.optim.lr_scheduler.StepLR(
            self.optimizer, step_size=30, gamma=0.1
        )
    
    def train_epoch(self, dataloader):
        """训练一个epoch"""
        self.model.train()
        total_loss = 0
        
        for batch_idx, (images, targets) in enumerate(dataloader):
            images = images.to(self.device)
            targets = targets.to(self.device).long()
            
            # 前向传播
            outputs = self.model(images, mode='fcn8s')
            loss = self.criterion(outputs, targets)
            
            # 反向传播
            self.optimizer.zero_grad()
            loss.backward()
            self.optimizer.step()
            
            total_loss += loss.item()
            
            if batch_idx % 50 == 0:
                print(f'Batch {batch_idx}, Loss: {loss.item():.4f}')
        
        return total_loss / len(dataloader)
    
    def evaluate(self, dataloader):
        """评估模型"""
        self.model.eval()
        total_loss = 0
        correct_pixels = 0
        total_pixels = 0
        
        with torch.no_grad():
            for images, targets in dataloader:
                images = images.to(self.device)
                targets = targets.to(self.device).long()
                
                outputs = self.model(images, mode='fcn8s')
                loss = self.criterion(outputs, targets)
                total_loss += loss.item()
                
                # 计算像素准确率
                pred = outputs.argmax(dim=1)
                mask = targets != 255  # 忽略标签
                correct_pixels += (pred[mask] == targets[mask]).sum().item()
                total_pixels += mask.sum().item()
        
        accuracy = correct_pixels / total_pixels
        return total_loss / len(dataloader), accuracy

# 示例使用
def demo_fcn():
    """FCN模型演示"""
    # 创建模型
    model = FCN(num_classes=21)  # PASCAL VOC有21个类别
    
    # 创建模拟数据
    batch_size = 2
    images = torch.randn(batch_size, 3, 512, 512)
    
    # 测试不同模式
    modes = ['fcn32s', 'fcn16s', 'fcn8s']
    for mode in modes:
        output = model(images, mode=mode)
        print(f"{mode} output shape: {output.shape}")
    
    # 可视化网络结构
    print("\nFCN网络结构:")
    print(f"参数量: {sum(p.numel() for p in model.parameters()):,}")

# 运行演示
demo_fcn()

2.2 U-Net：医学图像分割的经典架构

U-Net因其独特的U型结构和跳跃连接设计，在医学图像分割领域取得了巨大成功。

class UNet(nn.Module):
    """
    U-Net实现 - 医学图像分割的经典网络
    特点：对称的编码器-解码器结构 + 跳跃连接
    """
    def __init__(self, in_channels=3, num_classes=1, base_channels=64):
        super(UNet, self).__init__()
        
        # 编码器路径 (Contracting Path)
        self.encoder1 = self.conv_block(in_channels, base_channels)
        self.encoder2 = self.conv_block(base_channels, base_channels * 2)
        self.encoder3 = self.conv_block(base_channels * 2, base_channels * 4)
        self.encoder4 = self.conv_block(base_channels * 4, base_channels * 8)
        
        # 底部连接
        self.bottleneck = self.conv_block(base_channels * 8, base_channels * 16)
        
        # 解码器路径 (Expansive Path)
        self.upconv4 = nn.ConvTranspose2d(base_channels * 16, base_channels * 8, 
                                         kernel_size=2, stride=2)
        self.decoder4 = self.conv_block(base_channels * 16, base_channels * 8)
        
        self.upconv3 = nn.ConvTranspose2d(base_channels * 8, base_channels * 4, 
                                         kernel_size=2, stride=2)
        self.decoder3 = self.conv_block(base_channels * 8, base_channels * 4)
        
        self.upconv2 = nn.ConvTranspose2d(base_channels * 4, base_channels * 2, 
                                         kernel_size=2, stride=2)
        self.decoder2 = self.conv_block(base_channels * 4, base_channels * 2)
        
        self.upconv1 = nn.ConvTranspose2d(base_channels * 2, base_channels, 
                                         kernel_size=2, stride=2)
        self.decoder1 = self.conv_block(base_channels * 2, base_channels)
        
        # 输出层
        self.final_conv = nn.Conv2d(base_channels, num_classes, kernel_size=1)
        
        # 池化层
        self.maxpool = nn.MaxPool2d(kernel_size=2, stride=2)
        
    def conv_block(self, in_channels, out_channels):
        """基本卷积块"""
        return nn.Sequential(
            nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1, bias=False),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True),
            nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1, bias=False),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True)
        )
    
    def forward(self, x):
        # 编码器
        enc1 = self.encoder1(x)          # [B, 64, H, W]
        enc2 = self.encoder2(self.maxpool(enc1))  # [B, 128, H/2, W/2]
        enc3 = self.encoder3(self.maxpool(enc2))  # [B, 256, H/4, W/4]
        enc4 = self.encoder4(self.maxpool(enc3))  # [B, 512, H/8, W/8]
        
        # 底部
        bottleneck = self.bottleneck(self.maxpool(enc4))  # [B, 1024, H/16, W/16]
        
        # 解码器 + 跳跃连接
        dec4 = self.upconv4(bottleneck)  # [B, 512, H/8, W/8]
        dec4 = torch.cat([dec4, enc4], dim=1)  # [B, 1024, H/8, W/8]
        dec4 = self.decoder4(dec4)       # [B, 512, H/8, W/8]
        
        dec3 = self.upconv3(dec4)        # [B, 256, H/4, W/4]
        dec3 = torch.cat([dec3, enc3], dim=1)  # [B, 512, H/4, W/4]
        dec3 = self.decoder3(dec3)       # [B, 256, H/4, W/4]
        
        dec2 = self.upconv2(dec3)        # [B, 128, H/2, W/2]
        dec2 = torch.cat([dec2, enc2], dim=1)  # [B, 256, H/2, W/2]
        dec2 = self.decoder2(dec2)       # [B, 128, H/2, W/2]
        
        dec1 = self.upconv1(dec2)        # [B, 64, H, W]
        dec1 = torch.cat([dec1, enc1], dim=1)  # [B, 128, H, W]
        dec1 = self.decoder1(dec1)       # [B, 64, H, W]
        
        # 输出
        output = self.final_conv(dec1)   # [B, num_classes, H, W]
        return output

# 改进版U-Net：U-Net++
class UNetPlusPlus(nn.Module):
    """
    U-Net++ (嵌套U-Net)
    通过密集跳跃连接提升分割精度
    """
    def __init__(self, in_channels=3, num_classes=1, deep_supervision=False):
        super(UNetPlusPlus, self).__init__()
        
        self.deep_supervision = deep_supervision
        base_ch = 32
        
        # 编码器
        self.conv0_0 = self.conv_block(in_channels, base_ch)
        self.conv1_0 = self.conv_block(base_ch, base_ch*2)
        self.conv2_0 = self.conv_block(base_ch*2, base_ch*4)
        self.conv3_0 = self.conv_block(base_ch*4, base_ch*8)
        self.conv4_0 = self.conv_block(base_ch*8, base_ch*16)
        
        # 嵌套连接
        self.conv0_1 = self.conv_block(base_ch + base_ch*2, base_ch)
        self.conv1_1 = self.conv_block(base_ch*2 + base_ch*4, base_ch*2)
        self.conv2_1 = self.conv_block(base_ch*4 + base_ch*8, base_ch*4)
        self.conv3_1 = self.conv_block(base_ch*8 + base_ch*16, base_ch*8)
        
        self.conv0_2 = self.conv_block(base_ch*2 + base_ch*2, base_ch)
        self.conv1_2 = self.conv_block(base_ch*4 + base_ch*4, base_ch*2)
        self.conv2_2 = self.conv_block(base_ch*8 + base_ch*8, base_ch*4)
        
        self.conv0_3 = self.conv_block(base_ch*3 + base_ch*2, base_ch)
        self.conv1_3 = self.conv_block(base_ch*6 + base_ch*4, base_ch*2)
        
        self.conv0_4 = self.conv_block(base_ch*4 + base_ch*2, base_ch)
        
        # 上采样
        self.up = nn.Upsample(scale_factor=2, mode='bilinear', align_corners=True)
        
        # 输出层
        if self.deep_supervision:
            self.final1 = nn.Conv2d(base_ch, num_classes, kernel_size=1)
            self.final2 = nn.Conv2d(base_ch, num_classes, kernel_size=1)
            self.final3 = nn.Conv2d(base_ch, num_classes, kernel_size=1)
            self.final4 = nn.Conv2d(base_ch, num_classes, kernel_size=1)
        else:
            self.final = nn.Conv2d(base_ch, num_classes, kernel_size=1)
        
        self.maxpool = nn.MaxPool2d(kernel_size=2, stride=2)
    
    def conv_block(self, in_channels, out_channels):
        return 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)
        )
    
    def forward(self, x):
        x0_0 = self.conv0_0(x)
        x1_0 = self.conv1_0(self.maxpool(x0_0))
        x0_1 = self.conv0_1(torch.cat([x0_0, self.up(x1_0)], 1))
        
        x2_0 = self.conv2_0(self.maxpool(x1_0))
        x1_1 = self.conv1_1(torch.cat([x1_0, self.up(x2_0)], 1))
        x0_2 = self.conv0_2(torch.cat([x0_0, x0_1, self.up(x1_1)], 1))
        
        x3_0 = self.conv3_0(self.maxpool(x2_0))
        x2_1 = self.conv2_1(torch.cat([x2_0, self.up(x3_0)], 1))
        x1_2 = self.conv1_2(torch.cat([x1_0, x1_1, self.up(x2_1)], 1))
        x0_3 = self.conv0_3(torch.cat([x0_0, x0_1, x0_2, self.up(x1_2)], 1))
        
        x4_0 = self.conv4_0(self.maxpool(x3_0))
        x3_1 = self.conv3_1(torch.cat([x3_0, self.up(x4_0)], 1))
        x2_2 = self.conv2_2(torch.cat([x2_0, x2_1, self.up(x3_1)], 1))
        x1_3 = self.conv1_3(torch.cat([x1_0, x1_1, x1_2, self.up(x2_2)], 1))
        x0_4 = self.conv0_4(torch.cat([x0_0, x0_1, x0_2, x0_3, self.up(x1_3)], 1))
        
        if self.deep_supervision:
            output1 = self.final1(x0_1)
            output2 = self.final2(x0_2)
            output3 = self.final3(x0_3)
            output4 = self.final4(x0_4)
            return [output1, output2, output3, output4]
        else:
            output = self.final(x0_4)
            return output

# U-Net训练工具
class UNetTrainer:
    def __init__(self, model, device='cuda'):
        self.model = model.to(device)
        self.device = device
        
        # 二分类分割的损失函数
        self.criterion = nn.BCEWithLogitsLoss()
        self.optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
        self.scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
            self.optimizer, mode='min', patience=5, factor=0.5
        )
    
    def dice_loss(self, pred, target, smooth=1.):
        """Dice损失函数"""
        pred = torch.sigmoid(pred)
        intersection = (pred * target).sum()
        dice = (2. * intersection + smooth) / (pred.sum() + target.sum() + smooth)
        return 1 - dice
    
    def combined_loss(self, pred, target):
        """组合损失：BCE + Dice"""
        bce = self.criterion(pred, target)
        dice = self.dice_loss(pred, target)
        return 0.5 * bce + 0.5 * dice
    
    def train_epoch(self, dataloader):
        self.model.train()
        total_loss = 0
        
        for images, masks in dataloader:
            images = images.to(self.device)
            masks = masks.to(self.device).float()
            
            self.optimizer.zero_grad()
            outputs = self.model(images)
            
            loss = self.combined_loss(outputs, masks)
            loss.backward()
            self.optimizer.step()
            
            total_loss += loss.item()
        
        return total_loss / len(dataloader)
    
    def validate(self, dataloader):
        self.model.eval()
        total_loss = 0
        dice_scores = []
        
        with torch.no_grad():
            for images, masks in dataloader:
                images = images.to self.device)
                masks = masks.to(self.device).float()
                
                outputs = self.model(images)
                loss = self.combined_loss(outputs, masks)
                total_loss += loss.item()
                
                # 计算Dice分数
                pred = torch.sigmoid(outputs) > 0.5
                dice = self.dice_coefficient(pred, masks)
                dice_scores.append(dice.item())
        
        return total_loss / len(dataloader), np.mean(dice_scores)
    
    def dice_coefficient(self, pred, target, smooth=1.):
        """计算Dice系数"""
        intersection = (pred * target).sum()
        dice = (2. * intersection + smooth) / (pred.sum() + target.sum() + smooth)
        return dice

# 演示U-Net
def demo_unet():
    """U-Net演示"""
    # 创建模型
    unet = UNet(in_channels=3, num_classes=1)
    unet_pp = UNetPlusPlus(in_channels=3, num_classes=1, deep_supervision=False)
    
    # 测试输入
    x = torch.randn(2, 3, 256, 256)
    
    # 前向传播
    with torch.no_grad():
        output1 = unet(x)
        output2 = unet_pp(x)
    
    print(f"输入尺寸: {x.shape}")
    print(f"U-Net输出: {output1.shape}")
    print(f"U-Net++输出: {output2.shape}")
    
    # 参数统计
    print(f"\nU-Net参数量: {sum(p.numel() for p in unet.parameters()):,}")
    print(f"U-Net++参数量: {sum(p.numel() for p in unet_pp.parameters()):,}")

demo_unet()

2.3 DeepLab：空洞卷积的威力

DeepLab系列通过空洞卷积(Atrous Convolution)技术，在不增加参数的情况下扩大感受野，是语义分割领域的重要贡献。

class AtrousConv2d(nn.Module):
    """空洞卷积模块"""
    def __init__(self, in_channels, out_channels, kernel_size, dilation=1):
        super(AtrousConv2d, self).__init__()
        self.conv = nn.Conv2d(
            in_channels, out_channels, kernel_size,
            padding=dilation, dilation=dilation, bias=False
        )
        self.bn = nn.BatchNorm2d(out_channels)
        self.relu = nn.ReLU(inplace=True)
    
    def forward(self, x):
        return self.relu(self.bn(self.conv(x)))

class ASPP(nn.Module):
    """
    Atrous Spatial Pyramid Pooling
    空洞空间金字塔池化 - DeepLab的核心模块
    """
    def __init__(self, in_channels, out_channels=256):
        super(ASPP, self).__init__()
        
        # 不同膨胀率的空洞卷积
        self.conv1 = nn.Sequential(
            nn.Conv2d(in_channels, out_channels, 1, bias=False),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True)
        )
        
        self.conv2 = AtrousConv2d(in_channels, out_channels, 3, dilation=6)
        self.conv3 = AtrousConv2d(in_channels, out_channels, 3, dilation=12)
        self.conv4 = AtrousConv2d(in_channels, out_channels, 3, dilation=18)
        
        # 全局平均池化分支
        self.global_avg_pool = nn.Sequential(
            nn.AdaptiveAvgPool2d(1),
            nn.Conv2d(in_channels, out_channels, 1, bias=False),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True)
        )
        
        # 融合后的卷积
        self.conv_concat = nn.Sequential(
            nn.Conv2d(out_channels * 5, out_channels, 1, bias=False),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True),
            nn.Dropout(0.5)
        )
    
    def forward(self, x):
        size = x.size()[2:]
        
        # 5个分支
        feat1 = self.conv1(x)
        feat2 = self.conv2(x)
        feat3 = self.conv3(x)
        feat4 = self.conv4(x)
        feat5 = F.interpolate(
            self.global_avg_pool(x), size=size, 
            mode='bilinear', align_corners=False
        )
        
        # 拼接和融合
        concat = torch.cat([feat1, feat2, feat3, feat4, feat5], dim=1)
        output = self.conv_concat(concat)
        
        return output

class DeepLabV3Plus(nn.Module):
    """
    DeepLab v3+ 实现
    结合编码器-解码器结构和ASPP模块
    """
    def __init__(self, num_classes=21, backbone='resnet50', pretrained=True):
        super(DeepLabV3Plus, self).__init__()
        
        # ResNet backbone
        if backbone == 'resnet50':
            resnet = torchvision.models.resnet50(pretrained=pretrained)
        elif backbone == 'resnet101':
            resnet = torchvision.models.resnet101(pretrained=pretrained)
        
        # 编码器部分
        self.layer0 = nn.Sequential(
            resnet.conv1, resnet.bn1, resnet.relu, resnet.maxpool
        )
        self.layer1 = resnet.layer1
        self.layer2 = resnet.layer2
        self.layer3 = resnet.layer3
        self.layer4 = resnet.layer4
        
        # 修改layer3和layer4的步长，使用空洞卷积
        self._modify_resnet_stride()
        
        # ASPP模块
        self.aspp = ASPP(2048, 256)
        
        # 解码器
        self.decoder = nn.Sequential(
            nn.Conv2d(256, 48, 1, bias=False),
            nn.BatchNorm2d(48),
            nn.ReLU(inplace=True)
        )
        
        # 最终分类层
        self.classifier = nn.Sequential(
            nn.Conv2d(256 + 48, 256, 3, padding=1, bias=False),
            nn.BatchNorm2d(256),
            nn.ReLU(inplace=True),
            nn.Dropout(0.5),
            nn.Conv2d(256, 256, 3, padding=1, bias=False),
            nn.BatchNorm2d(256),
            nn.ReLU(inplace=True),
            nn.Dropout(0.1),
            nn.Conv2d(256, num_classes, 1)
        )
    
    def _modify_resnet_stride(self):
        """修改ResNet的步长和膨胀率"""
        # layer3的最后一个block
        self.layer3[0].conv2.stride = (1, 1)
        self.layer3[0].downsample[0].stride = (1, 1)
        
        # layer4使用空洞卷积
        for block in self.layer4:
            block.conv2.dilation = (2, 2)
            block.conv2.padding = (2, 2)
    
    def forward(self, x):
        input_size = x.size()[2:]
        
        # 编码器
        x = self.layer0(x)    # 1/4
        low_level = x         # 保存低级特征
        
        x = self.layer1(x)    # 1/4
        x = self.layer2(x)    # 1/8
        x = self.layer3(x)    # 1/8 (修改后)
        x = self.layer4(x)    # 1/8 (修改后)
        
        # ASPP
        x = self.aspp(x)      # [B, 256, H/8, W/8]
        
        # 上采样到1/4
        x = F.interpolate(x, size=low_level.size()[2:], 
                         mode='bilinear', align_corners=False)
        
        # 处理低级特征
        low_level = self.decoder(low_level)  # [B, 48, H/4, W/4]
        
        # 特征融合
        x = torch.cat([x, low_level], dim=1)  # [B, 304, H/4, W/4]
        
        # 分类
        x = self.classifier(x)                # [B, num_classes, H/4, W/4]
        
        # 最终上采样
        x = F.interpolate(x, size=input_size, 
                         mode='bilinear', align_corners=False)
        
        return x

# DeepLab训练器
class DeepLabTrainer:
    def __init__(self, model, device='cuda'):
        self.model = model.to(device)
        self.device = device
        
        # 损失函数 - 处理类别不平衡
        class_weights = torch.ones(21)  # PASCAL VOC
        class_weights[0] = 0.1  # 背景类权重降低
        self.criterion = nn.CrossEntropyLoss(
            weight=class_weights.to(device), ignore_index=255
        )
        
        # 优化器 - 不同层使用不同学习率
        backbone_params = []
        classifier_params = []
        
        for name, param in model.named_parameters():
            if 'classifier' in name or 'aspp' in name or 'decoder' in name:
                classifier_params.append(param)
            else:
                backbone_params.append(param)
        
        self.optimizer = torch.optim.SGD([
            {'params': backbone_params, 'lr': 1e-4},
            {'params': classifier_params, 'lr': 1e-3}
        ], momentum=0.9, weight_decay=1e-4)
        
        self.scheduler = torch.optim.lr_scheduler.PolynomialLR(
            self.optimizer, total_iters=100, power=0.9
        )
    
    def calculate_miou(self, pred, target, num_classes):
        """计算平均IoU"""
        pred = pred.argmax(dim=1)
        ious = []
        
        for c in range(num_classes):
            pred_c = (pred == c)
            target_c = (target == c)
            
            intersection = (pred_c & target_c).sum().float()
            union = (pred_c | target_c).sum().float()
            
            if union > 0:
                iou = intersection / union
                ious.append(iou.item())
        
        return np.mean(ious) if ious else 0.0
    
    def train_epoch(self, dataloader, num_classes=21):
        self.model.train()
        total_loss = 0
        total_miou = 0
        
        for batch_idx, (images, targets) in enumerate(dataloader):
            images = images.to(self.device)
            targets = targets.to(self.device).long()
            
            # 前向传播
            outputs = self.model(images)
            loss = self.criterion(outputs, targets)
            
            # 反向传播
            self.optimizer.zero_grad()
            loss.backward()
            self.optimizer.step()
            
            # 计算指标
            total_loss += loss.item()
            miou = self.calculate_miou(outputs, targets, num_classes)
            total_miou += miou
            
            if batch_idx % 20 == 0:
                print(f'Batch {batch_idx}: Loss={loss.item():.4f}, mIoU={miou:.4f}')
        
        self.scheduler.step()
        return total_loss / len(dataloader), total_miou / len(dataloader)

# 可视化空洞卷积效果
def visualize_atrous_convolution():
    """可视化空洞卷积的感受野"""
    import matplotlib.patches as patches
    
    fig, axes = plt.subplots(1, 4, figsize=(16, 4))
    dilations = [1, 2, 4, 8]
    
    for i, (ax, dilation) in enumerate(zip(axes, dilations)):
        # 创建网格
        ax.set_xlim(0, 10)
        ax.set_ylim(0, 10)
        ax.set_aspect('equal')
        
        # 绘制感受野
        center = 5
        kernel_size = 3
        effective_size = kernel_size + (kernel_size - 1) * (dilation - 1)
        
        # 绘制有效点
        for y in range(kernel_size):
            for x in range(kernel_size):
                pos_x = center - kernel_size//2 + x * dilation
                pos_y = center - kernel_size//2 + y * dilation
                
                if 0 <= pos_x < 10 and 0 <= pos_y < 10:
                    circle = patches.Circle((pos_x, pos_y), 0.2, 
                                          color='red', alpha=0.7)
                    ax.add_patch(circle)
        
        # 绘制中心点
        circle = patches.Circle((center, center), 0.2, color='blue')
        ax.add_patch(circle)
        
        ax.set_title(f'Dilation = {dilation}\nReceptive Field = {effective_size}x{effective_size}')
        ax.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.savefig('atrous_convolution_visualization.png', dpi=150, bbox_inches='tight')
    plt.show()

# 演示DeepLab
def demo_deeplab():
    """DeepLab演示"""
    model = DeepLabV3Plus(num_classes=21, backbone='resnet50')
    
    # 测试输入
    x = torch.randn(2, 3, 513, 513)  # DeepLab常用尺寸
    
    with torch.no_grad():
        output = model(x)
    
    print(f"输入尺寸: {x.shape}")
    print(f"输出尺寸: {output.shape}")
    print(f"参数量: {sum(p.numel() for p in model.parameters()):,}")

demo_deeplab()

print(“✅ 第二章：语义分割核心算法 - 已完成！”)

🎭 第三章：实例分割技术

实例分割不仅要识别像素的类别，还要区分同一类别的不同个体。这是比语义分割更具挑战性的任务，需要同时进行目标检测和精确的像素级分割。

3.1 实例分割概述

实例分割的核心挑战：

• 检测与分割的统一：需要先定位对象，再进行精确分割
• 实例区分：同一类别的不同实例需要分别标记
• 精确边界：要求像素级的精确分割边界
• 多尺度处理：不同大小的对象都要准确分割

3.2 Mask R-CNN：经典的实例分割框架

Mask R-CNN在Faster R-CNN的基础上增加了掩码分支，实现了端到端的实例分割。

import torch
import torch.nn as nn
import torch.nn.functional as F
from torchvision.models import resnet50
import numpy as np
import matplotlib.pyplot as plt
from typing import Dict, List, Tuple

class MaskRCNN(nn.Module):
    """
    Mask R-CNN实例分割模型
    在Faster R-CNN基础上增加掩码预测分支
    """
    def __init__(self, num_classes=81, backbone='resnet50'):
        super().__init__()
        self.num_classes = num_classes
        
        # 骨干网络 (共享特征提取器)
        if backbone == 'resnet50':
            resnet = resnet50(pretrained=True)
            self.backbone = nn.Sequential(*list(resnet.children())[:-2])
            backbone_dim = 2048
        
        # RPN (区域提议网络)
        self.rpn = self._build_rpn(backbone_dim)
        
        # ROI Align (替代ROI Pooling以保持像素对齐)
        self.roi_align = self._build_roi_align()
        
        # 检测头 (分类 + 边界框回归)
        self.detection_head = self._build_detection_head(backbone_dim)
        
        # 掩码头 (像素级分割)
        self.mask_head = self._build_mask_head(backbone_dim)
        
    def _build_rpn(self, in_channels):
        """构建区域提议网络"""
        return nn.Sequential(
            nn.Conv2d(in_channels, 512, 3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(512, 512, 3, padding=1),
            nn.ReLU(inplace=True),
            # 分类分支 (前景/背景)
            nn.Conv2d(512, 3, 1),  # 3个anchor比例
            # 回归分支 (边界框坐标)
            nn.Conv2d(512, 12, 1)  # 3个anchor × 4个坐标
        )
    
    def _build_roi_align(self):
        """构建ROI Align层"""
        # 简化实现，实际使用torchvision.ops.roi_align
        class SimpleROIAlign(nn.Module):
            def __init__(self, output_size=7):
                super().__init__()
                self.output_size = output_size
                
            def forward(self, features, rois):
                # 简化的ROI对齐实现
                batch_size, channels, height, width = features.shape
                num_rois = rois.shape[0]
                
                # 创建输出张量
                output = torch.zeros(
                    num_rois, channels, self.output_size, self.output_size,
                    device=features.device, dtype=features.dtype
                )
                
                # 对每个ROI进行特征提取 (简化版本)
                for i, roi in enumerate(rois):
                    # roi格式: [batch_idx, x1, y1, x2, y2]
                    batch_idx = int(roi[0])
                    x1, y1, x2, y2 = roi[1:].int()
                    
                    # 提取ROI区域特征
                    roi_features = features[batch_idx, :, y1:y2, x1:x2]
                    
                    # 调整尺寸到固定大小
                    if roi_features.numel() > 0:
                        roi_features = F.interpolate(
                            roi_features.unsqueeze(0),
                            size=(self.output_size, self.output_size),
                            mode='bilinear', align_corners=False
                        ).squeeze(0)
                        output[i] = roi_features
                
                return output
        
        return SimpleROIAlign(output_size=7)
    
    def _build_detection_head(self, in_channels):
        """构建检测头"""
        return nn.Sequential(
            nn.Linear(in_channels * 7 * 7, 1024),
            nn.ReLU(inplace=True),
            nn.Linear(1024, 1024),
            nn.ReLU(inplace=True),
            # 分类分支
            nn.Linear(1024, self.num_classes),
            # 回归分支
            nn.Linear(1024, self.num_classes * 4)
        )
    
    def _build_mask_head(self, in_channels):
        """构建掩码预测头"""
        return nn.Sequential(
            # 上采样卷积层
            nn.ConvTranspose2d(in_channels, 256, 2, stride=2),
            nn.ReLU(inplace=True),
            nn.Conv2d(256, 256, 3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(256, 256, 3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(256, 256, 3, padding=1),
            nn.ReLU(inplace=True),
            # 最终掩码预测
            nn.ConvTranspose2d(256, self.num_classes, 2, stride=2),
            nn.Sigmoid()  # 输出概率掩码
        )
    
    def forward(self, images, targets=None):
        """
        前向传播
        Args:
            images: 输入图像 [B, 3, H, W]
            targets: 训练时的标注信息
        Returns:
            训练时返回损失字典，推理时返回预测结果
        """
        # 1. 特征提取
        features = self.backbone(images)
        
        # 2. RPN提议生成
        proposals = self._generate_proposals(features)
        
        # 3. ROI特征提取
        roi_features = self.roi_align(features, proposals)
        
        # 4. 检测预测
        detection_outputs = self._forward_detection_head(roi_features)
        
        # 5. 掩码预测
        mask_outputs = self._forward_mask_head(roi_features)
        
        if self.training and targets is not None:
            # 训练模式：计算损失
            return self._compute_losses(detection_outputs, mask_outputs, targets)
        else:
            # 推理模式：后处理并返回结果
            return self._postprocess_detections(detection_outputs, mask_outputs)
    
    def _generate_proposals(self, features):
        """生成候选区域"""
        # 简化的提议生成 (实际实现需要anchor生成和NMS)
        batch_size, _, height, width = features.shape
        num_proposals = 100  # 每张图片的提议数量
        
        proposals = []
        for b in range(batch_size):
            # 生成随机提议 (实际应该基于RPN输出)
            props = torch.rand(num_proposals, 5, device=features.device)
            props[:, 0] = b  # 批次索引
            props[:, 1:] *= torch.tensor([width, height, width, height], 
                                       device=features.device)  # 坐标缩放
            proposals.append(props)
        
        return torch.cat(proposals, dim=0)
    
    def _forward_detection_head(self, roi_features):
        """检测头前向传播"""
        # 展平ROI特征
        flattened = roi_features.view(roi_features.size(0), -1)
        
        # 检测预测
        outputs = self.detection_head(flattened)
        
        return {
            'class_logits': outputs,  # 分类logits
            'bbox_regression': outputs  # 边界框回归 (简化版本)
        }
    
    def _forward_mask_head(self, roi_features):
        """掩码头前向传播"""
        mask_logits = self.mask_head(roi_features)
        return {'mask_logits': mask_logits}
    
    def _compute_losses(self, detection_outputs, mask_outputs, targets):
        """计算训练损失"""
        # 简化的损失计算
        total_loss = torch.tensor(0.0, device=detection_outputs['class_logits'].device)
        
        return {
            'total_loss': total_loss,
            'class_loss': total_loss * 0.3,
            'bbox_loss': total_loss * 0.3,
            'mask_loss': total_loss * 0.4
        }
    
    def _postprocess_detections(self, detection_outputs, mask_outputs):
        """后处理检测结果"""
        # 简化的后处理
        return {
            'boxes': torch.zeros(10, 4),  # 检测框
            'labels': torch.zeros(10, dtype=torch.long),  # 类别标签
            'scores': torch.zeros(10),  # 置信度
            'masks': torch.zeros(10, 28, 28)  # 掩码
        }

# Mask R-CNN训练函数
def train_mask_rcnn():
    """Mask R-CNN训练示例"""
    model = MaskRCNN(num_classes=81)
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
    
    # 模拟训练数据
    images = torch.randn(2, 3, 800, 800)
    targets = [
        {
            'boxes': torch.tensor([[100, 100, 200, 200], [300, 300, 400, 400]]),
            'labels': torch.tensor([1, 2]),
            'masks': torch.randint(0, 2, (2, 800, 800)).float()
        }
    ] * 2
    
    model.train()
    
    # 前向传播
    losses = model(images, targets)
    total_loss = losses['total_loss']
    
    # 反向传播
    optimizer.zero_grad()
    total_loss.backward()
    optimizer.step()
    
    print(f"训练损失: {total_loss.item():.4f}")
    print(f"分类损失: {losses['class_loss'].item():.4f}")
    print(f"边界框损失: {losses['bbox_loss'].item():.4f}")
    print(f"掩码损失: {losses['mask_loss'].item():.4f}")

# Mask R-CNN推理函数
def infer_mask_rcnn():
    """Mask R-CNN推理示例"""
    model = MaskRCNN(num_classes=81)
    model.eval()
    
    # 模拟推理数据
    images = torch.randn(1, 3, 800, 800)
    
    with torch.no_grad():
        predictions = model(images)
    
    print("推理结果:")
    print(f"检测框数量: {len(predictions['boxes'])}")
    print(f"检测框形状: {predictions['boxes'].shape}")
    print(f"标签形状: {predictions['labels'].shape}")
    print(f"置信度形状: {predictions['scores'].shape}")
    print(f"掩码形状: {predictions['masks'].shape}")

# 实例分割结果可视化
def visualize_instance_segmentation():
    """可视化实例分割结果"""
    # 创建模拟的实例分割结果
    fig, axes = plt.subplots(2, 3, figsize=(15, 10))
    
    # 原图
    original_img = np.random.rand(256, 256, 3)
    axes[0, 0].imshow(original_img)
    axes[0, 0].set_title('原始图像')
    axes[0, 0].axis('off')
    
    # 检测框
    axes[0, 1].imshow(original_img)
    # 添加检测框 (模拟)
    from matplotlib.patches import Rectangle
    rect1 = Rectangle((50, 50), 100, 80, linewidth=2, 
                     edgecolor='red', facecolor='none')
    rect2 = Rectangle((150, 120), 90, 70, linewidth=2, 
                     edgecolor='blue', facecolor='none')
    axes[0, 1].add_patch(rect1)
    axes[0, 1].add_patch(rect2)
    axes[0, 1].set_title('检测框')
    axes[0, 1].axis('off')
    
    # 实例掩码
    mask1 = np.zeros((256, 256))
    mask1[50:130, 50:150] = 1
    mask2 = np.zeros((256, 256))
    mask2[120:190, 150:240] = 2
    
    combined_mask = mask1 + mask2
    axes[0, 2].imshow(combined_mask, cmap='tab10')
    axes[0, 2].set_title('实例掩码')
    axes[0, 2].axis('off')
    
    # 单个实例
    axes[1, 0].imshow(mask1, cmap='Reds')
    axes[1, 0].set_title('实例1掩码')
    axes[1, 0].axis('off')
    
    axes[1, 1].imshow(mask2 > 0, cmap='Blues')
    axes[1, 1].set_title('实例2掩码')
    axes[1, 1].axis('off')
    
    # 融合结果
    fusion = original_img.copy()
    mask_colored = np.zeros_like(fusion)
    mask_colored[mask1 > 0] = [1, 0, 0]  # 红色
    mask_colored[mask2 > 0] = [0, 0, 1]  # 蓝色
    
    fusion_result = 0.7 * fusion + 0.3 * mask_colored
    axes[1, 2].imshow(fusion_result)
    axes[1, 2].set_title('融合结果')
    axes[1, 2].axis('off')
    
    plt.tight_layout()
    plt.savefig('instance_segmentation_demo.png', dpi=150, bbox_inches='tight')
    plt.show()

# 演示函数
def demo_mask_rcnn():
    """Mask R-CNN完整演示"""
    print("=== Mask R-CNN实例分割演示 ===\n")
    
    # 模型构建
    model = MaskRCNN(num_classes=81)
    total_params = sum(p.numel() for p in model.parameters())
    print(f"模型参数量: {total_params:,}")
    
    # 训练演示
    print("\n1. 训练演示:")
    train_mask_rcnn()
    
    # 推理演示
    print("\n2. 推理演示:")
    infer_mask_rcnn()
    
    # 可视化演示
    print("\n3. 可视化演示:")
    visualize_instance_segmentation()

# 运行演示
demo_mask_rcnn()

3.3 YOLACT：实时实例分割

YOLACT (You Only Look At CoefficienTs) 是一种快速的实例分割方法，通过原型掩码和系数预测实现高效分割。

class YOLACT(nn.Module):
    """
    YOLACT实时实例分割模型
    通过原型掩码和系数组合实现快速分割
    """
    def __init__(self, num_classes=81, num_prototypes=32):
        super().__init__()
        self.num_classes = num_classes
        self.num_prototypes = num_prototypes
        
        # 骨干网络
        self.backbone = self._build_backbone()
        
        # FPN特征金字塔
        self.fpn = self._build_fpn()
        
        # 原型网络 (生成原型掩码)
        self.protonet = self._build_protonet()
        
        # 预测头 (分类 + 边界框 + 掩码系数)
        self.prediction_head = self._build_prediction_head()
    
    def _build_backbone(self):
        """构建骨干网络"""
        resnet = resnet50(pretrained=True)
        return nn.Sequential(*list(resnet.children())[:-2])
    
    def _build_fpn(self):
        """构建特征金字塔网络"""
        return nn.ModuleDict({
            'lateral_conv1': nn.Conv2d(2048, 256, 1),
            'lateral_conv2': nn.Conv2d(1024, 256, 1),
            'lateral_conv3': nn.Conv2d(512, 256, 1),
            'output_conv1': nn.Conv2d(256, 256, 3, padding=1),
            'output_conv2': nn.Conv2d(256, 256, 3, padding=1),
            'output_conv3': nn.Conv2d(256, 256, 3, padding=1),
        })
    
    def _build_protonet(self):
        """构建原型网络"""
        return nn.Sequential(
            nn.Conv2d(256, 256, 3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(256, 256, 3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(256, 256, 3, padding=1),
            nn.ReLU(inplace=True),
            # 上采样到原图1/4尺寸
            nn.Upsample(scale_factor=2, mode='bilinear', align_corners=False),
            nn.Conv2d(256, 256, 3, padding=1),
            nn.ReLU(inplace=True),
            # 生成原型掩码
            nn.Conv2d(256, self.num_prototypes, 1),
            nn.ReLU(inplace=True)  # 原型激活值非负
        )
    
    def _build_prediction_head(self):
        """构建预测头"""
        return nn.ModuleDict({
            'class_conv': nn.Conv2d(256, self.num_classes, 3, padding=1),
            'box_conv': nn.Conv2d(256, 4, 3, padding=1),
            'coeff_conv': nn.Conv2d(256, self.num_prototypes, 3, padding=1),
        })
    
    def forward(self, x):
        """
        前向传播
        Args:
            x: 输入图像 [B, 3, H, W]
        Returns:
            包含分类、边界框、掩码系数和原型的字典
        """
        # 1. 骨干网络特征提取
        backbone_features = self.backbone(x)
        
        # 2. FPN特征金字塔 (简化版本)
        fpn_features = self._forward_fpn(backbone_features)
        
        # 3. 原型掩码生成
        prototypes = self.protonet(fpn_features)
        
        # 4. 预测头输出
        class_pred = self.prediction_head['class_conv'](fpn_features)
        box_pred = self.prediction_head['box_conv'](fpn_features)
        coeff_pred = self.prediction_head['coeff_conv'](fpn_features)
        
        return {
            'class_pred': class_pred,      # [B, num_classes, H, W]
            'box_pred': box_pred,          # [B, 4, H, W]
            'coeff_pred': coeff_pred,      # [B, num_prototypes, H, W]
            'prototypes': prototypes       # [B, num_prototypes, H/4, W/4]
        }
    
    def _forward_fpn(self, backbone_features):
        """FPN前向传播 (简化版本)"""
        # 这里简化为直接使用backbone输出
        return backbone_features
    
    def assemble_masks(self, prototypes, coefficients, predictions):
        """
        组装最终掩码
        Args:
            prototypes: 原型掩码 [B, num_prototypes, H, W]
            coefficients: 掩码系数 [N, num_prototypes]
            predictions: 预测结果
        Returns:
            最终掩码 [N, H, W]
        """
        # 线性组合原型掩码
        # masks = coefficients @ prototypes
        prototypes_flat = prototypes.view(prototypes.size(0), 
                                        prototypes.size(1), -1)
        masks = torch.matmul(coefficients, prototypes_flat)
        
        # 重塑为图像形状
        masks = masks.view(masks.size(0), prototypes.size(2), prototypes.size(3))
        
        # Sigmoid激活得到掩码概率
        masks = torch.sigmoid(masks)
        
        return masks

# YOLACT可视化函数
def visualize_yolact_process():
    """可视化YOLACT工作流程"""
    fig, axes = plt.subplots(2, 4, figsize=(16, 8))
    
    # 模拟数据
    batch_size, h, w = 1, 64, 64
    num_prototypes = 4
    
    # 1. 输入图像
    input_img = np.random.rand(h, w, 3)
    axes[0, 0].imshow(input_img)
    axes[0, 0].set_title('输入图像')
    axes[0, 0].axis('off')
    
    # 2. 原型掩码
    prototypes = np.random.rand(num_prototypes, h, w)
    for i in range(num_prototypes):
        if i < 2:
            axes[0, i+1].imshow(prototypes[i], cmap='viridis')
            axes[0, i+1].set_title(f'原型掩码 {i+1}')
            axes[0, i+1].axis('off')
    
    axes[0, 3].imshow(prototypes[2], cmap='viridis')
    axes[0, 3].set_title('原型掩码 3')
    axes[0, 3].axis('off')
    
    # 3. 掩码系数
    coeffs = np.array([0.8, 0.3, -0.5, 0.2])  # 示例系数
    axes[1, 0].bar(range(num_prototypes), coeffs, color=['red', 'green', 'blue', 'orange'])
    axes[1, 0].set_title('掩码系数')
    axes[1, 0].set_xlabel('原型索引')
    axes[1, 0].set_ylabel('系数值')
    axes[1, 0].grid(True, alpha=0.3)
    
    # 4. 线性组合过程
    combined = np.zeros((h, w))
    for i, coeff in enumerate(coeffs):
        combined += coeff * prototypes[i]
    
    axes[1, 1].imshow(combined, cmap='RdBu')
    axes[1, 1].set_title('线性组合结果')
    axes[1, 1].axis('off')
    
    # 5. Sigmoid激活
    final_mask = 1 / (1 + np.exp(-combined))
    axes[1, 2].imshow(final_mask, cmap='gray')
    axes[1, 2].set_title('Sigmoid激活')
    axes[1, 2].axis('off')
    
    # 6. 最终结果
    result = input_img.copy()
    mask_colored = np.zeros_like(result)
    mask_colored[mask1 > 0] = [1, 0, 0]  # 红色
    mask_colored[mask2 > 0] = [0, 0, 1]  # 蓝色
    
    final_result = 0.7 * result + 0.3 * mask_colored
    axes[1, 3].imshow(final_result)
    axes[1, 3].set_title('最终分割结果')
    axes[1, 3].axis('off')
    
    plt.tight_layout()
    plt.savefig('yolact_process.png', dpi=150, bbox_inches='tight')
    plt.show()

# 演示YOLACT
def demo_yolact():
    """YOLACT演示"""
    model = YOLACT(num_classes=81, num_prototypes=32)
    
    # 测试输入
    x = torch.randn(2, 3, 550, 550)  # YOLACT常用输入尺寸
    
    with torch.no_grad():
        outputs = model(x)
    
    print("YOLACT输出:")
    for key, value in outputs.items():
        print(f"{key}: {value.shape}")
    
    # 可视化工作流程
    visualize_yolact_process()
    
    print(f"模型参数量: {sum(p.numel() for p in model.parameters()):,}")

demo_yolact()

3.4 实例分割性能优化技巧

# 实例分割优化技巧集合
class InstanceSegmentationOptimizer:
    """实例分割模型优化技巧"""
    
    @staticmethod
    def focal_loss(predictions, targets, alpha=0.25, gamma=2.0):
        """
        Focal Loss处理类别不平衡
        专门为实例分割中的前景/背景不平衡设计
        """
        ce_loss = F.cross_entropy(predictions, targets, reduction='none')
        pt = torch.exp(-ce_loss)
        focal_loss = alpha * (1 - pt) ** gamma * ce_loss
        return focal_loss.mean()
    
    @staticmethod
    def soft_nms(boxes, scores, masks, sigma=0.5, thresh=0.3):
        """
        Soft NMS避免硬性抑制重叠检测
        在密集场景中保留更多有效实例
        """
        # 简化的Soft NMS实现
        indices = torch.argsort(scores, descending=True)
        keep = []
        
        while len(indices) > 0:
            current = indices[0]
            keep.append(current)
            
            if len(indices) == 1:
                break
                
            # 计算IoU
            current_box = boxes[current].unsqueeze(0)
            other_boxes = boxes[indices[1:]]
            ious = InstanceSegmentationOptimizer.box_iou(current_box, other_boxes)
            
            # Soft NMS权重衰减
            weights = torch.exp(-(ious.squeeze() ** 2) / sigma)
            scores[indices[1:]] *= weights
            
            # 移除低分数检测
            valid_mask = scores[indices[1:]] > thresh
            indices = indices[1:][valid_mask]
        
        return torch.tensor(keep)
    
    @staticmethod
    def box_iou(box1, box2):
        """计算边界框IoU"""
        # 简化实现
        inter_area = torch.ones(box1.size(0), box2.size(0))
        union_area = torch.ones(box1.size(0), box2.size(0))
        return inter_area / union_area
    
    @staticmethod
    def multi_scale_training(model, images, targets, scales=[0.8, 1.0, 1.2]):
        """
        多尺度训练增强模型鲁棒性
        """
        total_loss = 0
        
        for scale in scales:
            # 缩放图像
            h, w = images.shape[-2:]
            new_h, new_w = int(h * scale), int(w * scale)
            
            scaled_images = F.interpolate(
                images, size=(new_h, new_w), 
                mode='bilinear', align_corners=False
            )
            
            # 相应调整标注
            scaled_targets = []
            for target in targets:
                scaled_target = target.copy()
                if 'boxes' in target:
                    scaled_target['boxes'] *= scale
                scaled_targets.append(scaled_target)
            
            # 前向传播
            outputs = model(scaled_images, scaled_targets)
            scale_loss = outputs['total_loss']
            
            total_loss += scale_loss / len(scales)
        
        return total_loss
    
    @staticmethod
    def feature_pyramid_attention(features):
        """
        特征金字塔注意力机制
        增强不同尺度特征的信息流动
        """
        # 全局平均池化获取上下文
        global_context = F.adaptive_avg_pool2d(features, 1)
        
        # 通道注意力
        channel_attention = torch.sigmoid(
            F.conv2d(global_context, weight=torch.ones(features.size(1), features.size(1), 1, 1))
        )
        
        # 空间注意力
        spatial_attention = torch.sigmoid(
            F.conv2d(features.mean(dim=1, keepdim=True), 
                    weight=torch.ones(1, 1, 7, 7), padding=3)
        )
        
        # 特征增强
        enhanced_features = features * channel_attention * spatial_attention
        
        return enhanced_features

# 性能评估工具
class InstanceSegmentationEvaluator:
    """实例分割评估工具"""
    
    def __init__(self, num_classes):
        self.num_classes = num_classes
        self.reset()
    
    def reset(self):
        """重置PQ指标"""
        self.predictions = []
        self.targets = []
    
    def add_batch(self, predictions, targets):
        """添加一个batch的预测和真值"""
        self.predictions.extend(predictions)
        self.targets.extend(targets)
    
    def compute_ap(self, iou_threshold=0.5):
        """计算平均精度AP"""
        # 简化的AP计算
        total_ap = 0
        valid_classes = 0
        
        for class_id in range(self.num_classes):
            class_predictions = [p for p in self.predictions if p['class'] == class_id]
            class_targets = [t for t in self.targets if t['class'] == class_id]
            
            if len(class_targets) == 0:
                continue
            
            # 计算该类别的AP
            class_ap = self._compute_class_ap(class_predictions, class_targets, iou_threshold)
            total_ap += class_ap
            valid_classes += 1
        
        return total_ap / valid_classes if valid_classes > 0 else 0
    
    def _compute_class_ap(self, predictions, targets, iou_threshold):
        """计算单个类别的AP"""
        # 简化实现
        return 0.5  # 占位符
    
    def compute_metrics(self):
        """计算完整评估指标"""
        ap_50 = self.compute_ap(0.5)
        ap_75 = self.compute_ap(0.75)
        
        # AP在不同IoU阈值下的平均
        ap_5095 = np.mean([self.compute_ap(t) for t in np.arange(0.5, 1.0, 0.05)])
        
        return {
            'AP@0.5': ap_50,
            'AP@0.75': ap_75,
            'AP@0.5:0.95': ap_5095
        }

# 可视化实例分割结果
def visualize_instance_segmentation_results(image, results, class_names):
    """可视化实例分割结果"""
    fig, ax = plt.subplots(figsize=(8, 8))
    
    # 原图
    ax.imshow(image.permute(1, 2, 0))
    
    # 检测框
    for result in results:
        if result['score'] < 0.5:
            continue
        
        # 画框
        x1, y1, x2, y2 = result['box']
        width, height = x2 - x1, y2 - y1
        rect = plt.Rectangle((x1, y1), width, height, 
                             linewidth=2, edgecolor='red', facecolor='none')
        ax.add_patch(rect)
        
        # 标签
        class_id = result['class']
        class_name = class_names[class_id] if class_id < len(class_names) else f'Class {class_id}'
        ax.text(x1, y1, f'{class_name}: {result["score"]:.2f}', 
                fontsize=12, color='red', bbox=dict(facecolor='yellow', alpha=0.5))
    
    ax.set_title('实例分割结果')
    ax.axis('off')
    plt.show()

# 演示实例分割可视化
def demo_instance_segmentation_visualization():
    """演示实例分割结果可视化"""
    # 模拟数据
    image = torch.randn(3, 256, 256)
    results = [
        {'box': [30, 30, 100, 100], 'class': 0, 'score': 0.95},
        {'box': [60, 60, 120, 120], 'class': 1, 'score': 0.85},
        {'box': [90, 90, 150, 150], 'class': 0, 'score': 0.90}
    ]
    class_names = ['背景', '物体']
    
    visualize_instance_segmentation_results(image, results, class_names)

# 运行演示
demo_instance_segmentation_visualization()

print(“✅ 第三章：实例分割技术 - 已完成！”)

🌐 第四章：全景分割技术

全景分割（Panoptic Segmentation）统一了语义分割和实例分割，为图像中的每个像素分配语义标签和实例ID，实现了场景的完整理解。

4.1 全景分割概述

全景分割的核心思想是将所有像素分为两类：

• Thing类别：可数的对象（如人、车、动物），需要实例分割
• Stuff类别：不可数的区域（如天空、道路、草地），只需语义分割

4.2 Panoptic FPN：统一的全景分割框架

Panoptic FPN是首个端到端的全景分割框架，它在Mask R-CNN的基础上增加了语义分割分支。

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
import matplotlib.pyplot as plt
from collections import defaultdict

class PanopticFPN(nn.Module):
    """Panoptic FPN全景分割网络"""
    
    def __init__(self, num_classes=134, num_thing_classes=80):
        super().__init__()
        self.num_classes = num_classes
        self.num_thing_classes = num_thing_classes  # 可数对象类别数
        self.num_stuff_classes = num_classes - num_thing_classes  # 不可数区域类别数
        
        # 骨干网络 + FPN
        self.backbone_fpn = self._build_backbone_fpn()
        
        # 实例分割分支（复用Mask R-CNN）
        self.instance_head = self._build_instance_head()
        
        # 语义分割分支
        self.semantic_head = self._build_semantic_head()
        
    def _build_backbone_fpn(self):
        """构建骨干网络和FPN"""
        # 使用ResNet50 + FPN
        import torchvision.models as models
        backbone = models.resnet50(pretrained=True)
        backbone = nn.Sequential(*list(backbone.children())[:-2])
        
        # FPN
        fpn = FeaturePyramidNetwork([256, 512, 1024, 2048], 256)
        
        return nn.Sequential(backbone, fpn)
    
    def _build_instance_head(self):
        """构建实例分割头"""
        return InstanceHead(256, self.num_thing_classes)
    
    def _build_semantic_head(self):
        """构建语义分割头"""
        return SemanticHead(256, self.num_stuff_classes)
    
    def forward(self, images, targets=None):
        # 特征提取
        features = self.backbone_fpn(images)
        
        # 实例分割预测
        instance_results = self.instance_head(features, targets)
        
        # 语义分割预测
        semantic_results = self.semantic_head(features)
        
        if self.training and targets is not None:
            # 训练模式：返回损失
            return {
                **instance_results['losses'],
                'semantic_loss': semantic_results['loss']
            }
        else:
            # 推理模式：融合结果
            panoptic_results = self.panoptic_fusion(
                instance_results['predictions'],
                semantic_results['predictions']
            )
            return panoptic_results

class SemanticHead(nn.Module):
    """语义分割头"""
    
    def __init__(self, in_channels, num_classes):
        super().__init__()
        self.num_classes = num_classes
        
        # 多尺度特征融合
        self.lateral_convs = nn.ModuleList([
            nn.Conv2d(in_channels, 128, 1) for _ in range(4)
        ])
        
        # 语义分割卷积层
        self.conv1 = nn.Conv2d(128 * 4, 256, 3, padding=1)
        self.conv2 = nn.Conv2d(256, 256, 3, padding=1)
        self.classifier = nn.Conv2d(256, num_classes, 1)
        
        # 上采样
        self.upsample = nn.Upsample(scale_factor=4, mode='bilinear', align_corners=False)
        
    def forward(self, features):
        # 多尺度特征融合
        target_size = features[0].shape[-2:]
        
        fused_features = []
        for i, (feature, lateral_conv) in enumerate(zip(features, self.lateral_convs)):
            # 调整特征图大小
            if i > 0:
                feature = F.interpolate(feature, size=target_size, mode='bilinear', align_corners=False)
            
            # 1x1卷积降维
            feature = lateral_conv(feature)
            fused_features.append(feature)
        
        # 拼接多尺度特征
        x = torch.cat(fused_features, dim=1)
        
        # 语义分割预测
        x = F.relu(self.conv1(x))
        x = F.relu(self.conv2(x))
        semantic_logits = self.classifier(x)
        
        # 上采样到原图尺寸
        semantic_logits = self.upsample(semantic_logits)
        
        if self.training:
            # 计算语义分割损失
            loss = self.compute_semantic_loss(semantic_logits, targets)
            return {'loss': loss}
        else:
            return {'predictions': semantic_logits}
    
    def compute_semantic_loss(self, logits, targets):
        """计算语义分割损失"""
        # 这里简化，实际需要根据targets计算损失
        return F.cross_entropy(logits, targets['semantic_masks'])

class InstanceHead(nn.Module):
    """实例分割头（基于Mask R-CNN）"""
    
    def __init__(self, in_channels, num_classes):
        super().__init__()
        # 这里可以复用Mask R-CNN的实现
        pass
    
    def forward(self, features, targets=None):
        # 实例分割的具体实现
        if self.training:
            return {'losses': {}}
        else:
            return {'predictions': {}}

class PanopticFusion:
    """全景分割融合模块"""
    
    def __init__(self, overlap_threshold=0.5, stuff_area_threshold=4096):
        self.overlap_threshold = overlap_threshold
        self.stuff_area_threshold = stuff_area_threshold
    
    def __call__(self, instance_results, semantic_results):
        """
        融合实例分割和语义分割结果
        
        Args:
            instance_results: 实例分割结果
            semantic_results: 语义分割结果
        
        Returns:
            panoptic_segmentation: 全景分割结果
        """
        panoptic_seg = torch.zeros_like(semantic_results, dtype=torch.int32)
        segments_info = []
        current_segment_id = 1
        
        # 处理实例分割结果（Thing类别）
        for mask, label, score in zip(
            instance_results['masks'],
            instance_results['labels'], 
            instance_results['scores']
        ):
            # 过滤低置信度预测
            if score < 0.5:
                continue
            
            mask_area = mask.sum().item()
            if mask_area == 0:
                continue
            
            # 检查与已有分割的重叠
            intersect = (panoptic_seg > 0) & mask
            intersect_area = intersect.sum().item()
            
            if intersect_area / mask_area < self.overlap_threshold:
                # 添加新的实例
                panoptic_seg[mask] = current_segment_id
                segments_info.append({
                    'id': current_segment_id,
                    'category_id': label.item(),
                    'area': mask_area,
                    'iscrowd': False
                })
                current_segment_id += 1
        
        # 处理语义分割结果（Stuff类别）
        semantic_pred = torch.argmax(semantic_results, dim=0)
        
        for label in torch.unique(semantic_pred):
            if label == 0:  # 忽略背景
                continue
                
            mask = (semantic_pred == label) & (panoptic_seg == 0)
            mask_area = mask.sum().item()
            
            if mask_area > self.stuff_area_threshold:
                panoptic_seg[mask] = current_segment_id
                segments_info.append({
                    'id': current_segment_id,
                    'category_id': label.item() + 80,  # Stuff类别ID偏移
                    'area': mask_area,
                    'iscrowd': False
                })
                current_segment_id += 1
        
        return {
            'panoptic_seg': panoptic_seg,
            'segments_info': segments_info
        }

# Panoptic FPN训练
def train_panoptic_fpn():
    """训练Panoptic FPN"""
    
    model = PanopticFPN(num_classes=134, num_thing_classes=80)
    
    # 优化器
    optimizer = torch.optim.SGD(
        model.parameters(), 
        lr=0.02, 
        momentum=0.9, 
        weight_decay=0.0001
    )
    
    # 学习率调度器
    scheduler = torch.optim.lr_scheduler.MultiStepLR(
        optimizer, 
        milestones=[16, 22], 
        gamma=0.1
    )
    
    model.train()
    for epoch in range(24):
        epoch_losses = defaultdict(float)
        
        for batch_idx, (images, targets) in enumerate(dataloader):
            optimizer.zero_grad()
            
            # 前向传播
            loss_dict = model(images, targets)
            
            # 计算总损失
            total_loss = sum(loss_dict.values())
            
            # 反向传播
            total_loss.backward()
            
            # 梯度裁剪
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            
            optimizer.step()
            
            # 记录损失
            for key, value in loss_dict.items():
                epoch_losses[key] += value.item()
            
            if batch_idx % 100 == 0:
                print(f'Epoch {epoch}, Batch {batch_idx}')
                for key, value in loss_dict.items():
                    print(f'  {key}: {value.item():.4f}')
        
        scheduler.step()
        
        # 打印epoch平均损失
        print(f'Epoch {epoch} completed:')
        for key, value in epoch_losses.items():
            print(f'  Average {key}: {value/len(dataloader):.4f}')

# 全景分割评估
class PanopticQuality:
    """全景分割质量评估"""
    
    def __init__(self, num_classes=134):
        self.num_classes = num_classes
        self.reset()
    
    def reset(self):
        self.pq_per_class = np.zeros(self.num_classes)
        self.sq_per_class = np.zeros(self.num_classes)
        self.rq_per_class = np.zeros(self.num_classes)
        self.num_samples = 0
    
    def update(self, pred_panoptic, pred_segments, gt_panoptic, gt_segments):
        """更新PQ指标"""
        self.num_samples += 1
        
        # 计算每个类别的指标
        for class_id in range(self.num_classes):
            pq, sq, rq = self._compute_pq_single_class(
                pred_panoptic, pred_segments,
                gt_panoptic, gt_segments,
                class_id
            )
            
            self.pq_per_class[class_id] += pq
            self.sq_per_class[class_id] += sq
            self.rq_per_class[class_id] += rq
    
    def _compute_pq_single_class(self, pred_pan, pred_segs, gt_pan, gt_segs, class_id):
        """计算单个类别的PQ"""
        # 获取该类别的预测和真值段
        pred_class_segments = [s for s in pred_segs if s['category_id'] == class_id]
        gt_class_segments = [s for s in gt_segs if s['category_id'] == class_id]
        
        if len(pred_class_segments) == 0 and len(gt_class_segments) == 0:
            return 1.0, 1.0, 1.0
        
        if len(pred_class_segments) == 0 or len(gt_class_segments) == 0:
            return 0.0, 0.0, 0.0
        
        # 计算IoU匹配
        ious = []
        matched_gt = set()
        
        for pred_seg in pred_class_segments:
            pred_mask = pred_pan == pred_seg['id']
            best_iou = 0
            best_gt_idx = -1
            
            for gt_idx, gt_seg in enumerate(gt_class_segments):
                if gt_idx in matched_gt:
                    continue
                
                gt_mask = gt_pan == gt_seg['id']
                
                # 计算IoU
                intersection = (pred_mask & gt_mask).sum().float()
                union = (pred_mask | gt_mask).sum().float()
                
                if union > 0:
                    iou = intersection / union
                    if iou > best_iou and iou > 0.5:
                        best_iou = iou
                        best_gt_idx = gt_idx
            
            if best_gt_idx >= 0:
                ious.append(best_iou)
                matched_gt.add(best_gt_idx)
        
        # 计算PQ组件
        if len(ious) == 0:
            return 0.0, 0.0, 0.0
        
        sq = np.mean(ious)  # Segmentation Quality
        rq = len(ious) / len(gt_class_segments)  # Recognition Quality
        pq = sq * rq  # Panoptic Quality
        
        return pq, sq, rq
    
    def compute(self):
        """计算最终PQ指标"""
        if self.num_samples == 0:
            return {'PQ': 0, 'SQ': 0, 'RQ': 0}
        
        # 平均每个类别的指标
        pq_per_class = self.pq_per_class / self.num_samples
        sq_per_class = self.sq_per_class / self.num_samples
        rq_per_class = self.rq_per_class / self.num_samples
        
        # 计算整体指标
        pq = np.mean(pq_per_class[pq_per_class > 0])
        sq = np.mean(sq_per_class[sq_per_class > 0])
        rq = np.mean(rq_per_class[rq_per_class > 0])
        
        return {
            'PQ': pq,
            'SQ': sq,
            'RQ': rq,
            'PQ_per_class': pq_per_class
        }

# 可视化全景分割结果
def visualize_panoptic_results(image, panoptic_result, class_names):
    """绘制全景分割结果"""
    fig, axes = plt.subplots(2, 2, figsize=(16, 12))
    
    # 原图
    axes[0, 0].imshow(image.permute(1, 2, 0))
    axes[0, 0].set_title('Original Image', fontsize=14)
    axes[0, 0].axis('off')
    
    # 全景分割结果
    panoptic_seg = panoptic_result['panoptic_seg']
    segments_info = panoptic_result['segments_info']
    
    # 创建彩色分割图
    colored_seg = np.zeros((*panoptic_seg.shape, 3), dtype=np.uint8)
    colors = plt.cm.Set3(np.linspace(0, 1, len(segments_info)))
    
    for i, segment in enumerate(segments_info):
        mask = panoptic_seg == segment['id']
        colored_seg[mask] = (colors[i][:3] * 255).astype(np.uint8)
    
    axes[0, 1].imshow(colored_seg)
    axes[0, 1].set_title('Panoptic Segmentation', fontsize=14)
    axes[0, 1].axis('off')
    
    # Thing vs Stuff 分析
    thing_mask = np.zeros_like(panoptic_seg, dtype=bool)
    stuff_mask = np.zeros_like(panoptic_seg, dtype=bool)
    
    for segment in segments_info:
        mask = panoptic_seg == segment['id']
        if segment['category_id'] < 80:  # Thing类别
            thing_mask |= mask
        else:  # Stuff类别
            stuff_mask |= mask
    
    thing_stuff_vis = np.zeros((*panoptic_seg.shape, 3))
    thing_stuff_vis[thing_mask] = [1, 0, 0]  # 红色表示Thing
    thing_stuff_vis[stuff_mask] = [0, 0, 1]   # 蓝色表示Stuff
    
    axes[1, 0].imshow(thing_stuff_vis)
    axes[1, 0].set_title('Thing (Red) vs Stuff (Blue)', fontsize=14)
    axes[1, 0].axis('off')
    
    # 叠加结果
    overlay = image.permute(1, 2, 0).clone().numpy()
    overlay = overlay * 0.6 + colored_seg.astype(float) / 255 * 0.4
    
    axes[1, 1].imshow(np.clip(overlay, 0, 1))
    axes[1, 1].set_title('Overlay Result', fontsize=14)
    axes[1, 1].axis('off')
    
    plt.tight_layout()
    plt.savefig('panoptic_segmentation_demo.png', dpi=150, bbox_inches='tight')
    plt.show()

# 演示Panoptic FPN
def demo_panoptic_fpn():
    """Panoptic FPN演示"""
    model = PanopticFPN(num_classes=134, num_thing_classes=80)
    
    # 测试输入
    x = torch.randn(2, 3, 512, 512)  # Panoptic FPN常用输入尺寸
    
    with torch.no_grad():
        output = model(x)
    
    print(f"输入尺寸: {x.shape}")
    print(f"输出尺寸: {output['panoptic_seg'].shape}")
    print(f"参数量: {sum(p.numel() for p in model.parameters()):,}")

# 运行演示
demo_panoptic_fpn()