08-卷积神经网络¶

import torch
import torch.nn.functional as F

# 示例：手动边缘检测
# 水平边缘检测核
kernel_horizontal = torch.tensor([
    [-1, -1, -1],
    [ 0,  0,  0],
    [ 1,  1,  1]
], dtype=torch.float32).reshape(1, 1, 3, 3)  # reshape重塑张量形状

# 垂直边缘检测核
kernel_vertical = torch.tensor([
    [-1, 0, 1],
    [-1, 0, 1],
    [-1, 0, 1]
], dtype=torch.float32).reshape(1, 1, 3, 3)

# 对一张灰度图应用卷积
# input shape: (batch_size, channels, height, width)
image = torch.randn(1, 1, 28, 28)
edges_h = F.conv2d(image, kernel_horizontal, padding=1)
edges_v = F.conv2d(image, kernel_vertical, padding=1)

import torch.nn as nn

# 最大池化层
max_pool = nn.MaxPool2d(kernel_size=2, stride=2)  # 输出尺寸减半

# 全局平均池化
gap = nn.AdaptiveAvgPool2d((1, 1))  # 输出固定为 1x1
# 使用示例
x = torch.randn(8, 512, 7, 7)  # (batch, channels, h, w)
out = gap(x)  # (8, 512, 1, 1)
out = out.view(out.size(0), -1)  # (8, 512) → 送入分类器

class LeNet5(nn.Module):  # 继承nn.Module定义神经网络层
    def __init__(self, num_classes=10):  # __init__构造方法，创建对象时自动调用
        super().__init__()  # super()调用父类方法
        self.features = nn.Sequential(
            nn.Conv2d(1, 6, kernel_size=5),
            nn.Sigmoid(),
            nn.AvgPool2d(kernel_size=2, stride=2),
            nn.Conv2d(6, 16, kernel_size=5),
            nn.Sigmoid(),
            nn.AvgPool2d(kernel_size=2, stride=2),
        )
        self.classifier = nn.Sequential(
            nn.Linear(16 * 5 * 5, 120),
            nn.Sigmoid(),
            nn.Linear(120, 84),
            nn.Sigmoid(),
            nn.Linear(84, num_classes),
        )

    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), -1)
        return self.classifier(x)

class BasicBlock(nn.Module):
    """ResNet-18/34 使用的基本残差块"""
    expansion = 1

    def __init__(self, in_channels, out_channels, stride=1, downsample=None):
        super().__init__()
        self.conv1 = nn.Conv2d(in_channels, out_channels, 3, stride, 1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_channels)
        self.conv2 = nn.Conv2d(out_channels, out_channels, 3, 1, 1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_channels)
        self.downsample = downsample  # 当维度不匹配时用1×1卷积调整
        self.relu = nn.ReLU(inplace=True)

    def forward(self, x):
        identity = x
        out = self.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        if self.downsample is not None:
            identity = self.downsample(x)
        out += identity  # 残差连接
        return self.relu(out)

class Bottleneck(nn.Module):
    """ResNet-50/101/152 使用的瓶颈残差块"""
    expansion = 4

    def __init__(self, in_channels, mid_channels, stride=1, downsample=None):
        super().__init__()
        out_channels = mid_channels * self.expansion
        self.conv1 = nn.Conv2d(in_channels, mid_channels, 1, bias=False)  # 1×1降维
        self.bn1 = nn.BatchNorm2d(mid_channels)
        self.conv2 = nn.Conv2d(mid_channels, mid_channels, 3, stride, 1, bias=False)  # 3×3卷积
        self.bn2 = nn.BatchNorm2d(mid_channels)
        self.conv3 = nn.Conv2d(mid_channels, out_channels, 1, bias=False)  # 1×1升维
        self.bn3 = nn.BatchNorm2d(out_channels)
        self.downsample = downsample
        self.relu = nn.ReLU(inplace=True)

    def forward(self, x):
        identity = x
        out = self.relu(self.bn1(self.conv1(x)))
        out = self.relu(self.bn2(self.conv2(out)))
        out = self.bn3(self.conv3(out))
        if self.downsample is not None:
            identity = self.downsample(x)
        out += identity
        return self.relu(out)

# 1×1卷积：将512通道降到64通道
reduce = nn.Conv2d(512, 64, kernel_size=1)
# 参数量仅：512 × 64 + 64 = 32,832

# 转置卷积：将特征图放大2倍
upsample = nn.ConvTranspose2d(256, 128, kernel_size=4, stride=2, padding=1)
# 输入 (256, 8, 8) → 输出 (128, 16, 16)

# 空洞卷积：3×3核，空洞率2，等效感受野5×5
dilated_conv = nn.Conv2d(64, 64, kernel_size=3, dilation=2, padding=2)

class DepthwiseSeparableConv(nn.Module):
    """MobileNet中使用的深度可分离卷积"""
    def __init__(self, in_channels, out_channels, stride=1):
        super().__init__()
        self.depthwise = nn.Conv2d(
            in_channels, in_channels, kernel_size=3,
            stride=stride, padding=1, groups=in_channels, bias=False
        )
        self.bn1 = nn.BatchNorm2d(in_channels)
        self.pointwise = nn.Conv2d(in_channels, out_channels, kernel_size=1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_channels)
        self.relu = nn.ReLU6(inplace=True)

    def forward(self, x):
        x = self.relu(self.bn1(self.depthwise(x)))
        x = self.relu(self.bn2(self.pointwise(x)))
        return x

import torchvision.transforms as T

basic_transforms = T.Compose([
    T.RandomHorizontalFlip(p=0.5),           # 水平翻转
    T.RandomVerticalFlip(p=0.1),             # 垂直翻转（某些场景）
    T.RandomRotation(degrees=15),            # 随机旋转
    T.RandomCrop(32, padding=4),             # 随机裁剪
    T.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2, hue=0.1),  # 颜色抖动
    T.RandomGrayscale(p=0.1),               # 随机灰度
    T.RandomErasing(p=0.1),                  # 随机擦除（Cutout变体）
    T.ToTensor(),
    T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])

def mixup_data(x, y, alpha=0.2):
    lam = torch.distributions.Beta(alpha, alpha).sample()  # 链式调用，连续执行多个方法
    batch_size = x.size(0)
    index = torch.randperm(batch_size).to(x.device)
    mixed_x = lam * x + (1 - lam) * x[index]
    y_a, y_b = y, y[index]
    return mixed_x, y_a, y_b, lam

def mixup_criterion(criterion, pred, y_a, y_b, lam):
    return lam * criterion(pred, y_a) + (1 - lam) * criterion(pred, y_b)

# RandAugment（torchvision内置）
from torchvision.transforms import RandAugment

train_transforms = T.Compose([
    T.RandomResizedCrop(224),
    T.RandomHorizontalFlip(),
    RandAugment(num_ops=2, magnitude=9),
    T.ToTensor(),
    T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])

import torchvision.models as models

# 加载预训练的ResNet-50
model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V2)

# 冻结所有层
for param in model.parameters():
    param.requires_grad = False

# 替换分类头
num_classes = 10
model.fc = nn.Linear(model.fc.in_features, num_classes)
# 只有model.fc的参数会被更新

# 策略一：全部微调，低学习率
for param in model.parameters():
    param.requires_grad = True

optimizer = torch.optim.AdamW([
    {'params': model.fc.parameters(), 'lr': 1e-3},       # 分类头：较大学习率
    {'params': model.layer4.parameters(), 'lr': 1e-4},    # 深层：中等学习率
    {'params': model.layer3.parameters(), 'lr': 1e-5},    # 浅层：小学习率
], weight_decay=0.01)

# 策略二：逐步解冻（Gradual Unfreezing）
# 先训练分类头 → 解冻最后几层 → 解冻更多层

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as T
from torch.utils.data import DataLoader

# ==================== 1. 超参数 ====================
BATCH_SIZE = 128
EPOCHS = 100
LR = 0.1
WEIGHT_DECAY = 5e-4
DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

# ==================== 2. 数据准备 ====================
train_transform = T.Compose([
    T.RandomCrop(32, padding=4),
    T.RandomHorizontalFlip(),
    T.ToTensor(),
    T.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)),
])

test_transform = T.Compose([
    T.ToTensor(),
    T.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)),
])

train_dataset = torchvision.datasets.CIFAR10(
    root='./data', train=True, download=True, transform=train_transform)
test_dataset = torchvision.datasets.CIFAR10(
    root='./data', train=False, download=True, transform=test_transform)

train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True, num_workers=2)  # DataLoader批量加载数据，支持shuffle和多进程
test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE, shuffle=False, num_workers=2)

# ==================== 3. 模型定义 ====================
class ResidualBlock(nn.Module):
    def __init__(self, in_channels, out_channels, stride=1):
        super().__init__()
        self.conv1 = nn.Conv2d(in_channels, out_channels, 3, stride, 1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_channels)
        self.conv2 = nn.Conv2d(out_channels, out_channels, 3, 1, 1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_channels)
        self.relu = nn.ReLU(inplace=True)

        self.shortcut = nn.Sequential()
        if stride != 1 or in_channels != out_channels:
            self.shortcut = nn.Sequential(
                nn.Conv2d(in_channels, out_channels, 1, stride, bias=False),
                nn.BatchNorm2d(out_channels)
            )

    def forward(self, x):
        out = self.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        out += self.shortcut(x)
        return self.relu(out)

class ResNetCIFAR(nn.Module):
    """简化版ResNet，适用于CIFAR-10 (32×32图像)"""
    def __init__(self, num_classes=10):
        super().__init__()
        self.conv1 = nn.Conv2d(3, 64, 3, 1, 1, bias=False)
        self.bn1 = nn.BatchNorm2d(64)
        self.relu = nn.ReLU(inplace=True)

        self.layer1 = self._make_layer(64, 64, 2, stride=1)
        self.layer2 = self._make_layer(64, 128, 2, stride=2)
        self.layer3 = self._make_layer(128, 256, 2, stride=2)
        self.layer4 = self._make_layer(256, 512, 2, stride=2)

        self.avgpool = nn.AdaptiveAvgPool2d((1, 1))
        self.fc = nn.Linear(512, num_classes)

        # Kaiming初始化
        for m in self.modules():
            if isinstance(m, nn.Conv2d):  # isinstance检查对象类型
                nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu')
            elif isinstance(m, nn.BatchNorm2d):
                nn.init.constant_(m.weight, 1)
                nn.init.constant_(m.bias, 0)

    def _make_layer(self, in_channels, out_channels, num_blocks, stride):
        layers = [ResidualBlock(in_channels, out_channels, stride)]
        for _ in range(1, num_blocks):
            layers.append(ResidualBlock(out_channels, out_channels, 1))
        return nn.Sequential(*layers)

    def forward(self, x):
        x = self.relu(self.bn1(self.conv1(x)))
        x = self.layer1(x)
        x = self.layer2(x)
        x = self.layer3(x)
        x = self.layer4(x)
        x = self.avgpool(x)
        x = x.view(x.size(0), -1)
        return self.fc(x)

# ==================== 4. 训练 ====================
model = ResNetCIFAR(num_classes=10).to(DEVICE)
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=LR, momentum=0.9, weight_decay=WEIGHT_DECAY)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=EPOCHS)

best_acc = 0.0

for epoch in range(EPOCHS):
    # 训练
    model.train()
    train_loss = 0.0
    correct = 0
    total = 0

    for inputs, targets in train_loader:
        inputs, targets = inputs.to(DEVICE), targets.to(DEVICE)

        optimizer.zero_grad()  # 清零梯度，防止梯度累积
        outputs = model(inputs)
        loss = criterion(outputs, targets)
        loss.backward()  # 反向传播计算梯度
        optimizer.step()  # 根据梯度更新模型参数

        train_loss += loss.item()  # .item()将单元素张量转为Python数值
        _, predicted = outputs.max(1)
        total += targets.size(0)
        correct += predicted.eq(targets).sum().item()

    scheduler.step()
    train_acc = 100. * correct / total

    # 验证
    model.eval()
    correct = 0
    total = 0
    with torch.no_grad():  # 禁用梯度计算，节省内存
        for inputs, targets in test_loader:
            inputs, targets = inputs.to(DEVICE), targets.to(DEVICE)
            outputs = model(inputs)
            _, predicted = outputs.max(1)
            total += targets.size(0)
            correct += predicted.eq(targets).sum().item()

    test_acc = 100. * correct / total
    lr = optimizer.param_groups[0]['lr']

    if test_acc > best_acc:
        best_acc = test_acc
        torch.save(model.state_dict(), 'best_cifar10_resnet.pth')

    if (epoch + 1) % 10 == 0:
        print(f"Epoch [{epoch+1}/{EPOCHS}] LR: {lr:.6f} | "
              f"Train Acc: {train_acc:.2f}% | Test Acc: {test_acc:.2f}% | "
              f"Best: {best_acc:.2f}%")

print(f"\n训练完成！最佳测试准确率: {best_acc:.2f}%")

import matplotlib.pyplot as plt

def visualize_feature_maps(model, image, layer_name='layer1'):
    """可视化中间层特征图"""
    activation = {}

    def hook_fn(module, input, output):
        activation[layer_name] = output.detach()  # detach()从计算图分离，不参与梯度计算

    # 注册hook
    layer = dict(model.named_modules())[layer_name]
    handle = layer.register_forward_hook(hook_fn)

    # 前向传播
    model.eval()
    with torch.no_grad():
        _ = model(image.unsqueeze(0).to(DEVICE))  # unsqueeze增加一个维度

    handle.remove()

    # 可视化前16个通道
    feat = activation[layer_name].cpu().squeeze(0)
    fig, axes = plt.subplots(4, 4, figsize=(10, 10))
    for i, ax in enumerate(axes.flat):  # enumerate同时获取索引和元素
        if i < feat.size(0):
            ax.imshow(feat[i], cmap='viridis')
        ax.axis('off')
    plt.suptitle(f'Feature Maps: {layer_name}')
    plt.tight_layout()
    plt.show()

import numpy as np
import cv2

class GradCAM:
    """Grad-CAM可视化"""
    def __init__(self, model, target_layer):
        self.model = model
        self.gradients = None
        self.activations = None

        target_layer.register_forward_hook(self._save_activation)
        target_layer.register_full_backward_hook(self._save_gradient)

    def _save_activation(self, module, input, output):
        self.activations = output.detach()

    def _save_gradient(self, module, grad_input, grad_output):
        self.gradients = grad_output[0].detach()

    def generate(self, input_image, target_class=None):
        self.model.eval()
        output = self.model(input_image)

        if target_class is None:
            target_class = output.argmax(dim=1).item()

        self.model.zero_grad()
        output[0, target_class].backward()

        # 全局平均池化梯度作为权重
        weights = self.gradients.mean(dim=[2, 3], keepdim=True)  # (1, C, 1, 1)
        cam = (weights * self.activations).sum(dim=1, keepdim=True)  # (1, 1, H, W)
        cam = torch.relu(cam)
        cam = cam.squeeze().cpu().numpy()

        # 归一化到 [0,1]
        cam = (cam - cam.min()) / (cam.max() - cam.min() + 1e-8)
        return cam

# 使用示例
# grad_cam = GradCAM(model, model.layer4[-1])
# cam = grad_cam.generate(image.unsqueeze(0).to(DEVICE))
# 将cam上采样到原图大小，叠加显示

def visualize_kernels(model):
    """可视化第一层卷积核"""
    kernels = model.conv1.weight.data.cpu()
    n = min(kernels.size(0), 64)
    fig, axes = plt.subplots(8, 8, figsize=(12, 12))
    for i, ax in enumerate(axes.flat):
        if i < n:
            kernel = kernels[i]
            # 归一化到 [0,1] 用于显示
            kernel = (kernel - kernel.min()) / (kernel.max() - kernel.min())
            if kernel.size(0) == 3:
                ax.imshow(kernel.permute(1, 2, 0))  # RGB
            else:
                ax.imshow(kernel[0], cmap='gray')    # 灰度
        ax.axis('off')
    plt.suptitle('First Layer Convolution Kernels')
    plt.tight_layout()
    plt.show()