06-正则化技术¶

import matplotlib.pyplot as plt

def plot_overfitting(train_losses, val_losses):
    """可视化过拟合"""
    plt.figure(figsize=(10, 6))
    plt.plot(train_losses, label='训练损失', color='blue')
    plt.plot(val_losses, label='验证损失', color='red')
    plt.xlabel('Epoch')
    plt.ylabel('Loss')
    plt.title('过拟合示意图')
    plt.legend()
    plt.grid(True)
    plt.show()

import torch
import torch.nn as nn

# 方法1：使用 PyTorch 的 weight_decay 参数
model = nn.Sequential(
    nn.Linear(784, 256),
    nn.ReLU(),
    nn.Linear(256, 128),
    nn.ReLU(),
    nn.Linear(128, 10)
)

# weight_decay 就是 L2 正则化系数 λ
optimizer = torch.optim.SGD(model.parameters(), lr=0.01, weight_decay=1e-4)

# 方法2：手动实现 L2 正则化
def l2_regularization(model, lambda_l2=1e-4):
    """手动计算 L2 正则化项"""
    l2_reg = torch.tensor(0.0)
    for param in model.parameters():
        l2_reg += torch.norm(param, p=2) ** 2
    return lambda_l2 * l2_reg

# 训练循环中使用
criterion = nn.CrossEntropyLoss()
for epoch in range(num_epochs):
    for inputs, targets in dataloader:
        outputs = model(inputs)
        loss = criterion(outputs, targets) + l2_regularization(model)
        optimizer.zero_grad()  # 清零梯度，防止梯度累积
        loss.backward()  # 反向传播计算梯度
        optimizer.step()  # 根据梯度更新模型参数

def l1_regularization(model, lambda_l1=1e-5):
    """手动计算 L1 正则化项"""
    l1_reg = torch.tensor(0.0)
    for param in model.parameters():
        l1_reg += torch.norm(param, p=1)
    return lambda_l1 * l1_reg

# Elastic Net：L1 + L2 结合
def elastic_net_regularization(model, lambda_l1=1e-5, lambda_l2=1e-4):
    """Elastic Net 正则化"""
    l1_reg = torch.tensor(0.0)
    l2_reg = torch.tensor(0.0)
    for param in model.parameters():
        l1_reg += torch.norm(param, p=1)
        l2_reg += torch.norm(param, p=2) ** 2
    return lambda_l1 * l1_reg + lambda_l2 * l2_reg

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

# ===== 手动实现 Dropout =====
class ManualDropout(nn.Module):
    def __init__(self, p=0.5):  # __init__构造方法，创建对象时自动调用
        """
        Args:
            p: 丢弃概率（注意：PyTorch 的 Dropout 参数 p 是丢弃概率，不是保留概率）
        """
        super().__init__()  # super()调用父类方法
        self.p = p

    def forward(self, x):
        if self.training and self.p > 0:
            # 生成 Bernoulli 掩码（保留概率为 1-p）
            keep_prob = 1 - self.p
            mask = torch.bernoulli(torch.full_like(x, keep_prob))
            # Inverted Dropout：训练时缩放
            return x * mask / keep_prob
        return x

# ===== 使用 PyTorch 内置 Dropout =====
class DropoutNet(nn.Module):
    def __init__(self, input_dim=784, hidden_dim=256, num_classes=10, dropout_rate=0.5):
        super().__init__()
        self.fc1 = nn.Linear(input_dim, hidden_dim)
        self.dropout1 = nn.Dropout(p=dropout_rate)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.dropout2 = nn.Dropout(p=dropout_rate)
        self.fc3 = nn.Linear(hidden_dim, num_classes)

    def forward(self, x):
        x = x.view(x.size(0), -1)
        x = F.relu(self.fc1(x))
        x = self.dropout1(x)
        x = F.relu(self.fc2(x))
        x = self.dropout2(x)
        x = self.fc3(x)
        return x

# 使用示例
model = DropoutNet()

# 训练时：开启 Dropout
model.train()
output_train = model(torch.randn(32, 784))

# 评估时：关闭 Dropout
model.eval()
output_eval = model(torch.randn(32, 784))

# ===== 2D Dropout（用于 CNN）=====
class CNNWithDropout(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = nn.Conv2d(3, 32, 3, padding=1)
        self.conv2 = nn.Conv2d(32, 64, 3, padding=1)
        self.pool = nn.MaxPool2d(2, 2)
        # Dropout2d 会丢弃整个特征图通道
        self.dropout2d = nn.Dropout2d(p=0.25)
        self.fc1 = nn.Linear(64 * 8 * 8, 256)
        self.dropout = nn.Dropout(p=0.5)
        self.fc2 = nn.Linear(256, 10)

    def forward(self, x):
        x = self.pool(F.relu(self.conv1(x)))
        x = self.dropout2d(x)
        x = self.pool(F.relu(self.conv2(x)))
        x = self.dropout2d(x)
        x = x.view(x.size(0), -1)
        x = F.relu(self.fc1(x))
        x = self.dropout(x)
        x = self.fc2(x)
        return x

import torch
import torch.nn as nn

# ===== 手动实现 Batch Normalization =====
class ManualBatchNorm1d(nn.Module):
    def __init__(self, num_features, eps=1e-5, momentum=0.1):
        super().__init__()
        self.num_features = num_features
        self.eps = eps
        self.momentum = momentum

        # 可学习参数
        self.gamma = nn.Parameter(torch.ones(num_features))
        self.beta = nn.Parameter(torch.zeros(num_features))

        # 移动平均统计量（不参与梯度计算）
        self.register_buffer('running_mean', torch.zeros(num_features))
        self.register_buffer('running_var', torch.ones(num_features))

    def forward(self, x):
        if self.training:
            # 训练模式：使用 batch 统计量
            mean = x.mean(dim=0)
            var = x.var(dim=0, unbiased=False)

            # 更新移动平均
            with torch.no_grad():
                self.running_mean = (1 - self.momentum) * self.running_mean + self.momentum * mean
                self.running_var = (1 - self.momentum) * self.running_var + self.momentum * var
        else:
            # 推理模式：使用移动平均
            mean = self.running_mean
            var = self.running_var

        # 标准化
        x_hat = (x - mean) / torch.sqrt(var + self.eps)
        # 缩放和平移
        out = self.gamma * x_hat + self.beta
        return out

# ===== 2D Batch Normalization（用于 CNN）手动实现 =====
class ManualBatchNorm2d(nn.Module):
    def __init__(self, num_features, eps=1e-5, momentum=0.1):
        super().__init__()
        self.num_features = num_features
        self.eps = eps
        self.momentum = momentum

        self.gamma = nn.Parameter(torch.ones(1, num_features, 1, 1))
        self.beta = nn.Parameter(torch.zeros(1, num_features, 1, 1))

        self.register_buffer('running_mean', torch.zeros(1, num_features, 1, 1))
        self.register_buffer('running_var', torch.ones(1, num_features, 1, 1))

    def forward(self, x):
        # x shape: (N, C, H, W)
        if self.training:
            mean = x.mean(dim=(0, 2, 3), keepdim=True)
            var = x.var(dim=(0, 2, 3), keepdim=True, unbiased=False)
            with torch.no_grad():
                self.running_mean = (1 - self.momentum) * self.running_mean + self.momentum * mean
                self.running_var = (1 - self.momentum) * self.running_var + self.momentum * var
        else:
            mean = self.running_mean
            var = self.running_var

        x_hat = (x - mean) / torch.sqrt(var + self.eps)
        return self.gamma * x_hat + self.beta

# ===== 使用 BN 的完整网络 =====
class BNNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(784, 256),
            nn.BatchNorm1d(256),
            nn.ReLU(),
            nn.Linear(256, 128),
            nn.BatchNorm1d(128),
            nn.ReLU(),
            nn.Linear(128, 10)
        )

    def forward(self, x):
        return self.layers(x.view(x.size(0), -1))

# 测试手动实现
manual_bn = ManualBatchNorm1d(256)
pytorch_bn = nn.BatchNorm1d(256)

x = torch.randn(32, 256)
manual_bn.train()
pytorch_bn.train()
print("手动实现输出 shape:", manual_bn(x).shape)
print("PyTorch 实现输出 shape:", pytorch_bn(x).shape)

# PyTorch 中 Layer Normalization
class TransformerBlock(nn.Module):
    def __init__(self, d_model=512, nhead=8, dim_ff=2048, dropout=0.1):
        super().__init__()
        self.self_attn = nn.MultiheadAttention(d_model, nhead, dropout=dropout)
        self.ffn = nn.Sequential(
            nn.Linear(d_model, dim_ff),
            nn.ReLU(),
            nn.Linear(dim_ff, d_model)
        )
        # Transformer 中使用 LayerNorm
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):
        # Pre-Norm 结构
        attn_out, _ = self.self_attn(self.norm1(x), self.norm1(x), self.norm1(x))
        x = x + self.dropout(attn_out)
        ffn_out = self.ffn(self.norm2(x))
        x = x + self.dropout(ffn_out)
        return x

# Instance Normalization：对每个样本的每个通道做归一化（风格迁移中常用）
instance_norm = nn.InstanceNorm2d(64)

# Group Normalization：将通道分组做归一化
group_norm = nn.GroupNorm(num_groups=32, num_channels=64)

# RMS Normalization：只使用均方根进行归一化（LLaMA 中使用）
class RMSNorm(nn.Module):
    def __init__(self, dim, eps=1e-6):
        super().__init__()
        self.eps = eps
        self.weight = nn.Parameter(torch.ones(dim))

    def forward(self, x):
        rms = torch.sqrt(torch.mean(x ** 2, dim=-1, keepdim=True) + self.eps)
        return x / rms * self.weight

import torch.nn as nn
import torch.nn.init as init

class InitializedNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(784, 256)
        self.fc2 = nn.Linear(256, 128)
        self.fc3 = nn.Linear(128, 10)

        # 应用初始化
        self._initialize_weights()

    def _initialize_weights(self):
        for m in self.modules():
            if isinstance(m, nn.Linear):  # isinstance检查对象类型
                # Xavier 初始化（适用于 Sigmoid/Tanh）
                # init.xavier_uniform_(m.weight)
                # init.xavier_normal_(m.weight)

                # Kaiming/He 初始化（适用于 ReLU）
                init.kaiming_normal_(m.weight, mode='fan_in', nonlinearity='relu')

                if m.bias is not None:
                    init.zeros_(m.bias)

            elif isinstance(m, nn.Conv2d):
                init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu')
                if m.bias is not None:
                    init.zeros_(m.bias)

            elif isinstance(m, nn.BatchNorm2d):
                init.ones_(m.weight)
                init.zeros_(m.bias)

# 验证初始化效果
def check_activation_stats(model, x):
    """检查各层激活值的统计量"""
    activations = []
    hooks = []

    def hook_fn(module, input, output):
        activations.append(output.detach())  # detach()从计算图分离，不参与梯度计算

    for layer in model.children():
        hooks.append(layer.register_forward_hook(hook_fn))

    with torch.no_grad():
        model(x)

    for i, act in enumerate(activations):  # enumerate同时获取索引和元素
        print(f"Layer {i}: mean={act.mean():.4f}, std={act.std():.4f}, "
              f"dead_ratio={(act == 0).float().mean():.4f}")  # 链式调用，连续执行多个方法

    for h in hooks:
        h.remove()

import torchvision.transforms as transforms

# ===== 基础数据增强 =====
basic_transform = transforms.Compose([
    transforms.RandomHorizontalFlip(p=0.5),
    transforms.RandomVerticalFlip(p=0.5),
    transforms.RandomRotation(degrees=15),
    transforms.RandomCrop(32, padding=4),
    transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2, hue=0.1),
    transforms.RandomGrayscale(p=0.1),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])

# ===== 高级数据增强 =====
advanced_transform = transforms.Compose([
    transforms.RandomResizedCrop(224, scale=(0.08, 1.0)),
    transforms.RandomHorizontalFlip(),
    transforms.RandomApply([
        transforms.ColorJitter(0.4, 0.4, 0.4, 0.1)
    ], p=0.8),
    transforms.RandomGrayscale(p=0.2),
    transforms.RandomApply([
        transforms.GaussianBlur(kernel_size=23, sigma=(0.1, 2.0))
    ], p=0.5),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
    transforms.RandomErasing(p=0.25),
])

import numpy as np

def mixup_data(x, y, alpha=1.0):
    """Mixup 数据增强"""
    if alpha > 0:
        lam = np.random.beta(alpha, alpha)
    else:
        lam = 1

    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):
    """Mixup 损失函数"""
    return lam * criterion(pred, y_a) + (1 - lam) * criterion(pred, y_b)

def cutmix_data(x, y, alpha=1.0):
    """CutMix 数据增强"""
    lam = np.random.beta(alpha, alpha)
    batch_size = x.size(0)
    index = torch.randperm(batch_size).to(x.device)

    # 生成随机框
    W, H = x.size(2), x.size(3)
    cut_ratio = np.sqrt(1 - lam)
    cut_w = int(W * cut_ratio)
    cut_h = int(H * cut_ratio)

    cx = np.random.randint(W)
    cy = np.random.randint(H)

    x1 = np.clip(cx - cut_w // 2, 0, W)
    y1 = np.clip(cy - cut_h // 2, 0, H)
    x2 = np.clip(cx + cut_w // 2, 0, W)
    y2 = np.clip(cy + cut_h // 2, 0, H)

    mixed_x = x.clone()
    mixed_x[:, :, x1:x2, y1:y2] = x[index, :, x1:x2, y1:y2]

    # 调整 lambda
    lam = 1 - ((x2 - x1) * (y2 - y1) / (W * H))
    y_a, y_b = y, y[index]
    return mixed_x, y_a, y_b, lam

class EarlyStopping:
    """早停机制"""
    def __init__(self, patience=10, min_delta=0.0, mode='min', verbose=True):
        """
        Args:
            patience: 在多少个 epoch 内验证指标未改善时停止
            min_delta: 被认为是改善的最小变化量
            mode: 'min' 表示指标越小越好（如 loss），'max' 表示越大越好（如 accuracy）
        """
        self.patience = patience
        self.min_delta = min_delta
        self.mode = mode
        self.verbose = verbose
        self.counter = 0
        self.best_score = None
        self.early_stop = False
        self.best_model_state = None

    def __call__(self, score, model):  # __call__使实例可像函数一样被调用
        import copy
        if self.best_score is None:
            self.best_score = score
            self.best_model_state = copy.deepcopy(model.state_dict())
        elif self._is_improvement(score):
            self.best_score = score
            self.best_model_state = copy.deepcopy(model.state_dict())
            self.counter = 0
            if self.verbose:
                print(f"EarlyStopping: 指标改善到 {score:.6f}")
        else:
            self.counter += 1
            if self.verbose:
                print(f"EarlyStopping: {self.counter}/{self.patience}")
            if self.counter >= self.patience:
                self.early_stop = True

    def _is_improvement(self, score):
        if self.mode == 'min':
            return score < self.best_score - self.min_delta
        else:
            return score > self.best_score + self.min_delta

    def load_best_model(self, model):
        model.load_state_dict(self.best_model_state)

# ===== 使用示例 =====
early_stopping = EarlyStopping(patience=10, mode='min')

for epoch in range(max_epochs):
    train_loss = train_one_epoch(model, train_loader, optimizer, criterion)
    val_loss = evaluate(model, val_loader, criterion)

    early_stopping(val_loss, model)
    if early_stopping.early_stop:
        print(f"Early stopping at epoch {epoch}")
        break

# 加载最佳模型
early_stopping.load_best_model(model)

import torch.nn.utils as utils

# ===== 按范数裁剪（推荐）=====
max_grad_norm = 1.0

loss.backward()
# 裁剪所有参数的梯度
total_norm = utils.clip_grad_norm_(model.parameters(), max_norm=max_grad_norm)
print(f"梯度范数: {total_norm:.4f}")
optimizer.step()

# ===== 按值裁剪 =====
max_grad_value = 0.5

loss.backward()
utils.clip_grad_value_(model.parameters(), clip_value=max_grad_value)
optimizer.step()

# ===== 监控梯度范数 =====
def monitor_gradients(model):
    """监控模型中各层的梯度范数"""
    total_norm = 0
    for name, param in model.named_parameters():
        if param.grad is not None:
            param_norm = param.grad.data.norm(2)
            total_norm += param_norm.item() ** 2  # .item()将单元素张量转为Python数值
            print(f"{name}: grad_norm = {param_norm:.6f}")
    total_norm = total_norm ** 0.5
    print(f"Total gradient norm: {total_norm:.6f}")
    return total_norm