CNN：卷积神经网络架构详解（基础篇）

import torch
import torch.nn as nn

# 全连接层处理 32x32 RGB 图像的参数对比
fc_params = 32 * 32 * 3 * 128  # 输入层到 128 个神经元
print(f"全连接层参数量: {fc_params:,}")  # 393,216

# 等价的 3x3 卷积层
conv_params = 3 * 3 * 3 * 128 + 128  # 128 个 3x3 卷积核 + bias
print(f"卷积层参数量: {conv_params:,}")  # 3,584

# 可视化展平操作破坏空间信息
import numpy as np

# 3x3 单通道图像，相邻像素为 1
img = np.array([
    [1, 2, 3],
    [4, 5, 6],
    [7, 8, 9]
])

# 展平后，原本相邻的 3 和 4 在向量中相距很远
flattened = img.flatten()
print(f"原始 2D:\n{img}")
print(f"展平后: {flattened}")
# 像素 3(index=2) 和像素 4(index=3) 虽然相邻，但语义关联被削弱

import torch
import torch.nn as nn

# 2D 卷积层参数解析
conv = nn.Conv2d(
    in_channels=3,      # RGB 三通道
    out_channels=16,    # 16 个卷积核
    kernel_size=3,      # 3x3 卷积核
    stride=1,           # 步长 1
    padding=1,          # 填充 1（保持尺寸）
    bias=True
)

# 参数量 = 16 * (3*3*3 + 1) = 448
total_params = sum(p.numel() for p in conv.parameters())
print(f"参数量: {total_params}")

# 输入输出尺寸验证
x = torch.randn(1, 3, 32, 32)  # batch=1, C=3, H=32, W=32
out = conv(x)
print(f"输出形状: {out.shape}")  # torch.Size([1, 16, 32, 32])

# 不同步长和填充对输出尺寸的影响
def conv_output_size(input_size, kernel_size, stride=1, padding=0):
    """计算卷积输出尺寸公式"""
    return (input_size + 2 * padding - kernel_size) // stride + 1

print("输入 32x32, 核 3x3:")
print(f"  stride=1, padding=0 → {conv_output_size(32, 3, 1, 0)}x{conv_output_size(32, 3, 1, 0)}")
print(f"  stride=1, padding=1 → {conv_output_size(32, 3, 1, 1)}x{conv_output_size(32, 3, 1, 1)}")
print(f"  stride=2, padding=1 → {conv_output_size(32, 3, 2, 1)}x{conv_output_size(32, 3, 2, 1)}")
# 输出: 30x30, 32x32, 16x16

import torch
import torch.nn as nn

# 最大池化 vs 平均池化对比
x = torch.tensor([[[[1, 2, 3, 4],
                    [5, 6, 7, 8],
                    [9, 10, 11, 12],
                    [13, 14, 15, 16]]]]).float()  # (1,1,4,4)

max_pool = nn.MaxPool2d(kernel_size=2, stride=2)
avg_pool = nn.AvgPool2d(kernel_size=2, stride=2)

print("最大池化结果:\n", max_pool(x))
# tensor([[[[ 6,  8],
#           [14, 16]]]])

print("平均池化结果:\n", avg_pool(x))
# tensor([[[[ 3.50,  5.50],
#           [11.50, 13.50]]]])

# 全局平均池化 (GAP) 替代全连接层
import torch.nn as nn

class GAPClassifier(nn.Module):
    """用全局平均池化代替 Flatten + FC"""
    def __init__(self, num_classes=10):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 64, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(64, 128, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
        )
        # GAP: (B, 128, 8, 8) -> (B, 128)
        self.gap = nn.AdaptiveAvgPool2d(1)
        self.classifier = nn.Linear(128, num_classes)

    def forward(self, x):
        x = self.features(x)
        x = self.gap(x).flatten(1)  # 全局平均
        return self.classifier(x)

import torch.nn as nn

# LeNet-5 简化版 PyTorch 实现
class LeNet5(nn.Module):
    def __init__(self, num_classes=10):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(1, 6, 5),      # 输入 1 通道, 6 个 5x5 核
            nn.Tanh(),
            nn.AvgPool2d(2),         # 池化 2x2
            nn.Conv2d(6, 16, 5),     # 16 个 5x5 核
            nn.Tanh(),
            nn.AvgPool2d(2),
        )
        self.classifier = nn.Sequential(
            nn.Linear(16 * 5 * 5, 120),
            nn.Tanh(),
            nn.Linear(120, 84),
            nn.Tanh(),
            nn.Linear(84, num_classes),
        )

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

# ResNet 基本残差块
import torch.nn as nn
import torch

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

    def __init__(self, in_channels, out_channels, stride=1):
        super().__init__()
        self.conv1 = nn.Conv2d(in_channels, out_channels, 3,
                               stride=stride, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_channels)
        self.conv2 = nn.Conv2d(out_channels, out_channels, 3,
                               stride=1, padding=1, bias=False)
        self.bn2 = 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):
        out = torch.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        out += self.shortcut(x)  # 残差连接：核心！
        return torch.relu(out)

def count_conv_params(in_ch, out_ch, kernel_size=3, bias=True):
    """计算卷积层参数量"""
    params = out_ch * in_ch * kernel_size * kernel_size
    if bias:
        params += out_ch
    return params

def count_conv_flops(in_ch, out_ch, kernel_size, in_h, in_w,
                     stride=1, padding=0, bias=True):
    """估算卷积层 FLOPs"""
    out_h = (in_h + 2 * padding - kernel_size) // stride + 1
    out_w = (in_w + 2 * padding - kernel_size) // stride + 1
    # 每个输出位置: kernel_h * kernel_w * in_ch 次乘 + 等量加
    flops_per_pos = 2 * kernel_size * kernel_size * in_ch
    return flops_per_pos * out_h * out_w * out_ch

# VGG-16 第一层: 3->64, 3x3, 224x224
params = count_conv_params(3, 64, 3)
flops = count_conv_flops(3, 64, 3, 224, 224, 1, 1)
print(f"VGG第一层: {params:,} params, {flops/1e6:.1f} MFLOPs")
# 1,792 params, 115.4 MFLOPs

# 使用 thop 库自动计算模型 FLOPs
# pip install thop
from thop import profile, clever_format
import torchvision.models as models
import torch

# 分析 ResNet-18
model = models.resnet18()
dummy_input = torch.randn(1, 3, 224, 224)
macs, params = profile(model, inputs=(dummy_input,))

# MACs 是乘加操作数, FLOPs = 2 * MACs
flops = 2 * macs
print(f"ResNet-18 FLOPs: {clever_format(flops, '%.2f')}")
print(f"ResNet-18 Params: {clever_format(params, '%.2f')}")
# 典型输出: 1.82 GFLOPs, 11.69 M params

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

# 策略1: 特征提取（冻结所有层）
def feature_extraction(num_classes=10):
    model = models.resnet18(weights=models.ResNet18_Weights.IMAGENET1K_V1)
    # 冻结所有预训练参数
    for param in model.parameters():
        param.requires_grad = False
    # 替换分类头
    model.fc = nn.Linear(model.fc.in_features, num_classes)
    return model  # 只有 fc 层可训练

# 策略2: 部分微调（冻结前两层，微调后两层）
def partial_finetune(num_classes=10):
    model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V2)
    frozen_count = 0
    for name, param in model.named_parameters():
        if 'layer3' not in name and 'layer4' not in name and 'fc' not in name:
            param.requires_grad = False
            frozen_count += 1
    model.fc = nn.Linear(model.fc.in_features, num_classes)
    print(f"冻结了 {frozen_count} 个参数组")
    return model

# 分层学习率设置
import torch.optim as optim

def setup_discriminative_lr(model, base_lr=1e-5, head_lr=1e-3):
    """为不同层设置不同学习率"""
    backbone_params = []
    head_params = []
    for name, param in model.named_parameters():
        if not param.requires_grad:
            continue
        if 'fc' in name or 'classifier' in name:
            head_params.append(param)
        else:
            backbone_params.append(param)

    optimizer = optim.Adam([
        {'params': backbone_params, 'lr': base_lr},
        {'params': head_params, 'lr': head_lr},
    ])
    return optimizer

# 带 warmup 的学习率调度
from torch.optim.lr_scheduler import CosineAnnealingLR

scheduler = CosineAnnealingLR(optimizer, T_max=50, eta_min=1e-7)
# 余弦退火：学习率从初始值平滑衰减到最小值

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

# 1. 数据准备
transform_train = transforms.Compose([
    transforms.RandomCrop(32, padding=4),   # 数据增强
    transforms.RandomHorizontalFlip(),       # 水平翻转
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465),
                         (0.2470, 0.2435, 0.2616)),  # CIFAR-10 均值方差
])

transform_test = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465),
                         (0.2470, 0.2435, 0.2616)),
])

trainset = datasets.CIFAR10(root='./data', train=True,
                            download=True, transform=transform_train)
testset = datasets.CIFAR10(root='./data', train=False,
                           download=True, transform=transform_test)
trainloader = DataLoader(trainset, batch_size=128, shuffle=True, num_workers=2)
testloader = DataLoader(testset, batch_size=128, shuffle=False, num_workers=2)

# 2. 模型定义
class SimpleCNN(nn.Module):
    def __init__(self, num_classes=10):
        super().__init__()
        self.features = nn.Sequential(
            # Block 1: 3 -> 64
            nn.Conv2d(3, 64, 3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.Conv2d(64, 64, 3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),  # 32 -> 16

            # Block 2: 64 -> 128
            nn.Conv2d(64, 128, 3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.Conv2d(128, 128, 3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),  # 16 -> 8

            # Block 3: 128 -> 256
            nn.Conv2d(128, 256, 3, padding=1),
            nn.BatchNorm2d(256),
            nn.ReLU(inplace=True),
            nn.Conv2d(256, 256, 3, padding=1),
            nn.BatchNorm2d(256),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),  # 8 -> 4
        )
        self.classifier = nn.Sequential(
            nn.Dropout(0.5),
            nn.Linear(256 * 4 * 4, 512),
            nn.ReLU(inplace=True),
            nn.Dropout(0.3),
            nn.Linear(512, num_classes),
        )

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

# 3. 训练循环
def train_one_epoch(model, loader, criterion, optimizer, device):
    model.train()
    total_loss = correct = total = 0
    for images, labels in loader:
        images, labels = images.to(device), labels.to(device)
        optimizer.zero_grad()
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

        total_loss += loss.item() * images.size(0)
        _, predicted = outputs.max(1)
        correct += predicted.eq(labels).sum().item()
        total += labels.size(0)
    return total_loss / total, correct / total

# 训练主流程
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = SimpleCNN().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.1,
                      momentum=0.9, weight_decay=5e-4)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100)

for epoch in range(100):
    train_loss, train_acc = train_one_epoch(
        model, trainloader, criterion, optimizer, device)
    scheduler.step()
    if (epoch + 1) % 10 == 0:
        print(f"Epoch {epoch+1}: loss={train_loss:.4f}, acc={train_acc:.4f}")