CNN 卷积神经网络（进阶）：从 LeNet 到 ResNet

import numpy as np

def conv2d_manual(image, kernel, stride=1, padding=0):
    """手动实现二维卷积（互相关）操作，理解底层原理"""
    # 填充
    if padding > 0:
        image = np.pad(image, padding, mode='constant', constant_values=0)

    h_in, w_in = image.shape
    k_h, k_w = kernel.shape
    h_out = (h_in - k_h) // stride + 1
    w_out = (w_in - k_w) // stride + 1

    output = np.zeros((h_out, w_out))

    # 滑动窗口计算
    for i in range(h_out):
        for j in range(w_out):
            region = image[i*stride:i*stride+k_h, j*stride:j*stride+k_w]
            output[i, j] = np.sum(region * kernel)

    return output

# 创建一个 5×5 的模拟图像（像素值 0-255）
img = np.array([
    [0, 0, 0, 0, 0],
    [0, 255, 255, 255, 0],
    [0, 255, 255, 255, 0],
    [0, 255, 255, 255, 0],
    [0, 0, 0, 0, 0]
], dtype=float)

# 垂直边缘检测卷积核
edge_kernel = np.array([
    [-1, 0, 1],
    [-1, 0, 1],
    [-1, 0, 1]
], dtype=float)

result = conv2d_manual(img, edge_kernel, stride=1, padding=0)
print("输入图像形状:", img.shape)
print("输出特征图形状:", result.shape)
print("卷积结果:")
print(result.astype(int))

# 多通道卷积：模拟 RGB 三通道输入
def conv2d_multi_channel(image_3d, kernels_3d, bias=None):
    """多通道卷积：每个卷积核也是 3D 的
    
    Args:
        image_3d: (H, W, C_in) 输入，如 RGB 图像
        kernels_3d: (C_out, K, K, C_in) 卷积核组
        bias: (C_out,) 偏置
    """
    H, W, C_in = image_3d.shape
    C_out, K, _, _ = kernels_3d.shape
    H_out = H - K + 1
    W_out = W - K + 1

    output = np.zeros((H_out, W_out, C_out))

    for c_out in range(C_out):
        kernel = kernels_3d[c_out]  # (K, K, C_in)
        for i in range(H_out):
            for j in range(W_out):
                region = image_3d[i:i+K, j:j+K, :]  # (K, K, C_in)
                output[i, j, c_out] = np.sum(region * kernel)
        if bias is not None:
            output[:, :, c_out] += bias[c_out]

    return output

# 模拟 6×6 RGB 图像
np.random.seed(42)
image_rgb = np.random.randint(0, 256, (6, 6, 3)).astype(float)

# 2 个 3×3×3 的随机卷积核
kernels = np.random.randn(2, 3, 3, 3).astype(float)
bias = np.array([0.1, -0.1])

output = conv2d_multi_channel(image_rgb, kernels, bias)
print(f"输入: {image_rgb.shape} → 输出: {output.shape}")
print(f"通道 0 的最大激活值: {output[:,:,0].max():.2f}")
print(f"通道 1 的最大激活值: {output[:,:,1].max():.2f}")

import numpy as np

def max_pool2d(feature_map, pool_size=2, stride=2):
    """最大池化"""
    H, W = feature_map.shape
    H_out = (H - pool_size) // stride + 1
    W_out = (W - pool_size) // stride + 1
    output = np.zeros((H_out, W_out))

    for i in range(H_out):
        for j in range(W_out):
            region = feature_map[i*stride:i*stride+pool_size,
                                 j*stride:j*stride+pool_size]
            output[i, j] = np.max(region)
    return output

def avg_pool2d(feature_map, pool_size=2, stride=2):
    """平均池化"""
    H, W = feature_map.shape
    H_out = (H - pool_size) // stride + 1
    W_out = (W - pool_size) // stride + 1
    output = np.zeros((H_out, W_out))

    for i in range(H_out):
        for j in range(W_out):
            region = feature_map[i*stride:i*stride+pool_size,
                                 j*stride:j*stride+pool_size]
            output[i, j] = np.mean(region)
    return output

# 演示
fm = np.array([
    [1, 3, 2, 4],
    [5, 6, 1, 2],
    [3, 2, 8, 7],
    [4, 1, 5, 3]
], dtype=float)

print("输入特征图:")
print(fm.astype(int))
print("\n最大池化 (2×2):")
print(max_pool2d(fm).astype(int))
print("\n平均池化 (2×2):")
print(np.round(avg_pool2d(fm), 1))

# 全局平均池化（GAP）：替代全连接层的现代做法
def global_average_pooling(feature_map_3d):
    """对每个通道做全局平均，将 (H, W, C) 变为 (C,)
    
    这是现代 CNN（如 ResNet）中替代 Flatten + Dense 的标准做法。
    它大幅减少参数量，同时保留每个通道的全局激活信息。
    """
    H, W, C = feature_map_3d.shape
    output = np.zeros(C)
    for c in range(C):
        output[c] = np.mean(feature_map_3d[:, :, c])
    return output

# 模拟 ResNet 最后的特征图：7×7×512
np.random.seed(42)
feature_map = np.random.randn(7, 7, 512).astype(float)

# 传统做法：Flatten → FC
flattened = feature_map.flatten()  # 7*7*512 = 25088 维
print(f"Flatten 后维度: {flattened.shape[0]}")
print(f"如果接 1000 类 FC 层: 参数量 = {flattened.shape[0] * 1000:,}")

# 现代做法：GAP → FC
gap_output = global_average_pooling(feature_map)
print(f"GAP 后维度: {gap_output.shape[0]}")
print(f"如果接 1000 类 FC 层: 参数量 = {gap_output.shape[0] * 1000:,}")
print(f"参数节省: {1 - gap_output.shape[0]/flattened.shape[0]:.1%}")

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

class BasicBlock(nn.Module):
    """ResNet 基本残差块（两层卷积）"""
    def __init__(self, in_channels, out_channels, stride=1):
        super().__init__()
        self.conv1 = nn.Conv2d(in_channels, out_channels,
                               kernel_size=3, stride=stride,
                               padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_channels)
        self.conv2 = nn.Conv2d(out_channels, out_channels,
                               kernel_size=3, stride=1,
                               padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_channels)

        # 如果输入输出维度不同，需要用 1×1 卷积对齐
        self.shortcut = nn.Sequential()
        if stride != 1 or in_channels != out_channels:
            self.shortcut = nn.Sequential(
                nn.Conv2d(in_channels, out_channels, kernel_size=1,
                          stride=stride, bias=False),
                nn.BatchNorm2d(out_channels)
            )

    def forward(self, x):
        identity = self.shortcut(x)  # 捷径路径
        out = F.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        out = out + identity  # 残差连接：F(x) + x
        out = F.relu(out)
        return out

# 验证残差块
block = BasicBlock(64, 64)
x = torch.randn(1, 64, 32, 32)
y = block(x)
print(f"输入: {x.shape} → 输出: {y.shape}")
print(f"残差连接确保梯度可以直接回传")

class Bottleneck(nn.Module):
    """ResNet 瓶颈块（三层卷积）—— 50 层以上使用
    
    1×1 Conv 降维 → 3×3 Conv 处理 → 1×1 Conv 升维
    这种设计在保持精度的同时大幅减少计算量
    """
    expansion = 4  # 输出通道是中间通道的 4 倍

    def __init__(self, in_channels, mid_channels, stride=1):
        super().__init__()
        out_channels = mid_channels * self.expansion

        self.conv1 = nn.Conv2d(in_channels, mid_channels, 1, bias=False)
        self.bn1 = nn.BatchNorm2d(mid_channels)
        self.conv2 = nn.Conv2d(mid_channels, mid_channels, 3,
                               stride=stride, padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(mid_channels)
        self.conv3 = nn.Conv2d(mid_channels, out_channels, 1, bias=False)
        self.bn3 = nn.BatchNorm2d(out_channels)

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

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

# 计算参数量对比
def count_params(module):
    return sum(p.numel() for p in module.parameters())

basic = BasicBlock(256, 256)
bottleneck = Bottleneck(256, 64)
print(f"BasicBlock 参数量: {count_params(basic):,}")
print(f"Bottleneck 参数量: {count_params(bottleneck):,}")
print(f"Bottleneck 输出通道: {256 * Bottleneck.expansion}")

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

# ========================================
# 1. 定义 CNN 模型
# ========================================
class SimpleCNN(nn.Module):
    """类 VGG 的小型 CNN，适合 CIFAR-10"""
    def __init__(self, num_classes=10):
        super().__init__()
        self.features = nn.Sequential(
            # 第一个卷积块
            nn.Conv2d(3, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.Conv2d(64, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2),  # 32->16

            # 第二个卷积块
            nn.Conv2d(64, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.Conv2d(128, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2),  # 16->8

            # 第三个卷积块
            nn.Conv2d(128, 256, kernel_size=3, padding=1),
            nn.BatchNorm2d(256),
            nn.ReLU(inplace=True),
            nn.Conv2d(256, 256, kernel_size=3, padding=1),
            nn.BatchNorm2d(256),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2),  # 8->4
        )

        # 分类头：全局平均池化 + 全连接
        self.classifier = nn.Sequential(
            nn.AdaptiveAvgPool2d((1, 1)),  # (256, 4, 4) -> (256, 1, 1)
            nn.Flatten(),
            nn.Linear(256, num_classes)
        )

    def forward(self, x):
        x = self.features(x)
        x = self.classifier(x)
        return x

model = SimpleCNN()
total_params = sum(p.numel() for p in model.parameters())
print(f"模型总参数量: {total_params:,}")
print(f"模型架构:")
print(model)

# ========================================
# 2. 训练与评估完整流程
# ========================================

def train_model(model, train_loader, val_loader, epochs=20, lr=0.001, device='cpu'):
    """完整的 CNN 训练循环"""
    model.to(device)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=lr)
    scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs)

    best_val_acc = 0.0

    for epoch in range(epochs):
        # 训练阶段
        model.train()
        train_loss = 0.0
        train_correct = 0
        train_total = 0

        for images, labels in train_loader:
            images, labels = images.to(device), labels.to(device)

            optimizer.zero_grad()
            outputs = model(images)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            train_loss += loss.item()
            _, predicted = outputs.max(1)
            train_total += labels.size(0)
            train_correct += predicted.eq(labels).sum().item()

        # 验证阶段
        model.eval()
        val_correct = 0
        val_total = 0
        with torch.no_grad():
            for images, labels in val_loader:
                images, labels = images.to(device), labels.to(device)
                outputs = model(images)
                _, predicted = outputs.max(1)
                val_total += labels.size(0)
                val_correct += predicted.eq(labels).sum().item()

        train_acc = train_correct / train_total
        val_acc = val_correct / val_total
        scheduler.step()

        if val_acc > best_val_acc:
            best_val_acc = val_acc
            torch.save(model.state_dict(), 'best_cnn.pt')

        print(f"Epoch {epoch+1:3d} | "
              f"Loss: {train_loss/len(train_loader):.4f} | "
              f"Train Acc: {train_acc:.4f} | "
              f"Val Acc: {val_acc:.4f} | "
              f"Best: {best_val_acc:.4f}")

    return model

# 数据加载
transform_train = transforms.Compose([
    transforms.RandomCrop(32, padding=4),
    transforms.RandomHorizontalFlip(),
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465),
                         (0.2023, 0.1994, 0.2010)),
])

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

print("训练流程已定义。在 GPU 上运行需要下载 CIFAR-10 数据集。")
print("运行命令: train_model(model, train_loader, test_loader, epochs=20)")

import torch
import torch.nn as nn
from torchvision import models

def setup_transfer_learning(model_name='resnet18', num_classes=10,
                            freeze_backbone=True, device='cpu'):
    """设置迁移学习模型"""
    # 加载预训练模型
    if model_name == 'resnet18':
        model = models.resnet18(weights=models.ResNet18_Weights.IMAGENET1K_V1)
    elif model_name == 'resnet50':
        model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1)
    elif model_name == 'efficientnet_b0':
        model = models.efficientnet_b0(
            weights=models.EfficientNet_B0_Weights.IMAGENET1K_V1)
    else:
        raise ValueError(f"未知模型: {model_name}")

    # 替换分类头
    if 'resnet' in model_name:
        n_features = model.fc.in_features
        model.fc = nn.Linear(n_features, num_classes)
    elif 'efficientnet' in model_name:
        n_features = model.classifier[1].in_features
        model.classifier[1] = nn.Linear(n_features, num_classes)

    # 冻结骨干网络
    if freeze_backbone:
        for name, param in model.named_parameters():
            if 'fc' not in name and 'classifier' not in name:
                param.requires_grad = False

    # 检查哪些层可训练
    trainable = [n for n, p in model.named_parameters() if p.requires_grad]
    frozen = [n for n, p in model.named_parameters() if not p.requires_grad]
    print(f"可训练参数: {len(trainable)} 组")
    print(f"冻结参数: {len(frozen)} 组")

    return model.to(device)

# 使用示例
model = setup_transfer_learning('resnet18', num_classes=10, freeze_backbone=True)
print("\nResNet18 迁移学习模型已就绪")

# 两阶段微调策略：第一阶段冻结，第二阶段解冻

def two_stage_finetune(model, train_loader, val_loader,
                       stage1_epochs=5, stage2_epochs=10, device='cpu'):
    """两阶段微调：先冻结骨干，再解冻全部微调"""
    model.to(device)
    criterion = nn.CrossEntropyLoss()

    # ===== 第一阶段：冻结骨干，只训练分类头 =====
    print("=== 第一阶段：冻结骨干网络 ===")
    optimizer = optim.Adam(filter(lambda p: p.requires_grad, model.parameters()),
                           lr=1e-3)

    for epoch in range(stage1_epochs):
        model.train()
        for images, labels in train_loader:
            images, labels = images.to(device), labels.to(device)
            optimizer.zero_grad()
            loss = criterion(model(images), labels)
            loss.backward()
            optimizer.step()

        # 简单评估
        model.eval()
        correct = total = 0
        with torch.no_grad():
            for images, labels in val_loader:
                images, labels = images.to(device), labels.to(device)
                pred = model(images).argmax(1)
                correct += (pred == labels).sum().item()
                total += labels.size(0)
        print(f"  Stage1 Epoch {epoch+1}: Val Acc = {correct/total:.4f}")

    # ===== 第二阶段：解冻全部层，小学习率微调 =====
    print("\n=== 第二阶段：解冻全部层微调 ===")
    for param in model.parameters():
        param.requires_grad = True

    optimizer = optim.Adam(model.parameters(), lr=1e-5)
    scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=stage2_epochs)

    for epoch in range(stage2_epochs):
        model.train()
        for images, labels in train_loader:
            images, labels = images.to(device), labels.to(device)
            optimizer.zero_grad()
            loss = criterion(model(images), labels)
            loss.backward()
            optimizer.step()
        scheduler.step()

        model.eval()
        correct = total = 0
        with torch.no_grad():
            for images, labels in val_loader:
                images, labels = images.to(device), labels.to(device)
                pred = model(images).argmax(1)
                correct += (pred == labels).sum().item()
                total += labels.size(0)
        print(f"  Stage2 Epoch {epoch+1}: Val Acc = {correct/total:.4f}")

    return model