深度学习学习教程，从入门到精通，自编码器 — 完整知识点与代码案例(14）

分享之家

454人浏览 · 2026-09-12 14:36:45

分享之家 · 2026-09-12 14:36:45 发布

自编码器 — 完整知识点与代码案例

一、欠完备自编码器 (Undercomplete Autoencoder)

1.1 核心知识点

自编码器（Autoencoder） 是一种神经网络，目标是学习输入数据的压缩表示。它由两部分组成：

编码器（Encoder）：将输入 $x$ 映射到隐藏表示 $h = f (x)$
解码器（Decoder）：将隐藏表示映射回重构 $x^=g(h)\hat{x} = g(h)$

欠完备（Undercomplete） 指隐藏层维度 $h$ 小于输入维度 $x$ ，迫使网络学习数据中最重要的特征。损失函数为：

$L (x, g (f (x)))$

常用 MSE 或交叉熵。当 $h$ 维度远小于 $x$ 维度时，自编码器被迫学习数据的低维流形结构，效果类似于 PCA。

1.2 案例代码：基础欠完备自编码器（MNIST）

import torch                          # PyTorch 深度学习框架
import torch.nn as nn                 # 神经网络模块，包含各种层和损失函数
import torch.optim as optim           # 优化器模块，包含SGD、Adam等
import torchvision                    # 计算机视觉工具库
import torchvision.transforms as transforms  # 数据预处理和增强工具
import matplotlib.pyplot as plt       # 绘图库，用于可视化结果

# ======================== 超参数设置 ========================
batch_size = 128          # 每批次处理的样本数，影响训练速度和内存占用
learning_rate = 1e-3      # 学习率，控制参数更新步长
num_epochs = 20           # 训练总轮数，遍历整个数据集的次数
encoding_dim = 32         # 编码维度（隐藏层大小），必须小于输入维度784

# ======================== 数据预处理 ========================
transform = transforms.Compose([
    transforms.ToTensor(),            # 将PIL图像转为Tensor，像素值从[0,255]变为[0,1]
])

# 下载并加载MNIST训练集
# MNIST包含60000张28x28的手写数字灰度图
train_dataset = torchvision.datasets.MNIST(
    root='./data',                    # 数据存储路径
    train=True,                       # 指定为训练集
    transform=transform,              # 应用预处理变换
    download=True                     # 如果本地没有则自动下载
)

# 创建数据加载器，支持自动批处理、打乱和多进程加载
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset,            # 传入数据集
    batch_size=batch_size,            # 每批次样本数
    shuffle=True,                     # 每个epoch打乱数据顺序，增加随机性
    num_workers=2                     # 用于数据加载的子进程数
)

# ======================== 定义欠完备自编码器 ========================
class UndercompleteAutoencoder(nn.Module):
    """
    欠完备自编码器：
    - 输入层: 784 (28x28 像素展平)
    - 编码层: 32 维 (远小于784，强制压缩)
    - 解码层: 784 维 (重构回原始维度)
    """
    def __init__(self, input_dim=784, encoding_dim=32):
        super(UndercompleteAutoencoder, self).__init__()
        
        # ---------- 编码器部分 ----------
        # 全连接层：将784维输入压缩到32维隐藏表示
        # 权重矩阵形状：[784, 32]，偏置向量形状：[32]
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 256),       # 第一层：784 → 256，初步降维
            nn.ReLU(inplace=True),           # ReLU激活：f(x)=max(0,x)，引入非线性
            nn.Linear(256, encoding_dim),    # 第二层：256 → 32，进一步压缩
            nn.ReLU(inplace=True)            # ReLU激活，确保编码非负
        )
        
        # ---------- 解码器部分 ----------
        # 全连接层：将32维隐藏表示还原到784维
        self.decoder = nn.Sequential(
            nn.Linear(encoding_dim, 256),    # 第一层：32 → 256，逐步扩展
            nn.ReLU(inplace=True),           # ReLU激活
            nn.Linear(256, input_dim),       # 第二层：256 → 784，重构输出
            nn.Sigmoid()                     # Sigmoid激活：输出值限制在[0,1]，匹配像素范围
        )
    
    def forward(self, x):
        """
        前向传播：定义数据从输入到输出的流动路径
        参数x: 输入张量，形状 [batch_size, 784]
        返回: 重构张量，形状 [batch_size, 784]
        """
        encoded = self.encoder(x)            # 编码：输入 → 压缩表示
        decoded = self.decoder(encoded)      # 解码：压缩表示 → 重构
        return decoded                       # 返回重构结果
    
    def encode(self, x):
        """仅执行编码操作，用于获取数据的压缩表示"""
        return self.encoder(x)

# ======================== 模型初始化 ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')  # 自动选择GPU或CPU
model = UndercompleteAutoencoder(input_dim=784, encoding_dim=encoding_dim).to(device)  # 实例化并移至设备

# 均方误差损失：L = (1/n) * Σ(x_i - x̂_i)²
criterion = nn.MSELoss()

# Adam优化器：自适应学习率，momentum=0.9，适合稀疏梯度
optimizer = optim.Adam(model.parameters(), lr=learning_rate, weight_decay=1e-5)

# ======================== 训练循环 ========================
losses = []  # 记录每个epoch的平均损失

for epoch in range(num_epochs):
    epoch_loss = 0.0  # 累积当前epoch的损失
    for batch_idx, (data, _) in enumerate(train_loader):
        # data形状: [batch_size, 1, 28, 28]
        # 注意：MNIST标签(_)在此处不需要，自编码器是无监督学习
        
        data = data.view(data.size(0), -1).to(device)  # 展平：[B,1,28,28] → [B,784]，并移至设备
        
        # ---- 前向传播 ----
        reconstructed = model(data)          # 通过模型得到重构结果
        loss = criterion(reconstructed, data)  # 计算重构误差（MSE损失）
        
        # ---- 反向传播与优化 ----
        optimizer.zero_grad()                # 清除上一步的梯度（PyTorch默认累积梯度）
        loss.backward()                      # 反向传播：计算损失对所有参数的梯度
        optimizer.step()                     # 更新参数：θ = θ - lr * gradient
        
        epoch_loss += loss.item()            # 累加损失值（.item()从Tensor提取标量）
    
    # 计算并记录平均损失
    avg_loss = epoch_loss / len(train_loader)
    losses.append(avg_loss)
    print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {avg_loss:.6f}')

# ======================== 可视化结果 ========================
# 选取测试样本进行重构展示
test_data = next(iter(train_loader))[0][:8]       # 取前8张图
test_flat = test_data.view(8, -1).to(device)       # 展平
with torch.no_grad():                              # 禁用梯度计算，节省内存
    reconstructed = model(test_flat)               # 执行重构

fig, axes = plt.subplots(2, 8, figsize=(16, 4))    # 创建2行8列子图
for i in range(8):
    axes[0, i].imshow(test_data[i].squeeze(), cmap='gray')        # 第一行：原始图像
    axes[0, i].axis('off')
    axes[1, i].imshow(reconstructed[i].cpu().view(28,28), cmap='gray')  # 第二行：重构图像
    axes[1, i].axis('off')
plt.suptitle('Top: Original | Bottom: Reconstructed', fontsize=14)
plt.tight_layout()
plt.show()

二、正则自编码器 (Regularized Autoencoder)

2.1 核心知识点

正则自编码器通过在损失函数中添加正则化项来约束编码器的行为，即使编码维度不小于输入维度也能学到有意义的表示。常见正则化策略：

正则化方法	损失函数	效果
稀疏正则	$\lambda \|h\|_1$	编码向量稀疏，只有少量非零
权重衰减	$\lambda \|\theta\|_2^2$	防止权重过大，类似L2正则
变分正则	$L + K L (q ∣ p)$	学习概率分布（VAE的基础）

关键区别：

欠完备：通过架构约束（维度小）
正则化：通过损失函数约束（即使维度大也能学到好的表示）

2.2 案例代码：稀疏自编码器

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt

# ======================== 超参数设置 ========================
batch_size = 128              # 批大小
num_epochs = 30               # 稀疏自编码器通常需要更多轮次
learning_rate = 1e-3          # 学习率
sparse_weight = 1e-3          # 稀疏正则化系数λ，控制稀疏惩罚强度
target_sparsity = 0.05        # 目标激活率ρ：每个神经元平均只有5%时间被激活
input_dim = 784               # 输入维度 28×28
hidden_dim = 256              # 隐藏层维度可以很大（甚至是过完备的）

# ======================== 数据加载 ========================
transform = transforms.Compose([transforms.ToTensor()])

train_dataset = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform, download=True
)
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset, batch_size=batch_size, shuffle=True, num_workers=2
)

# ======================== 稀疏自编码器定义 ========================
class SparseAutoencoder(nn.Module):
    """
    稀疏自编码器：
    - 隐藏层维度可以大于或等于输入维度（过完备）
    - 通过稀疏正则化强制编码向量稀疏
    - KL散度作为稀疏惩罚项
    """
    def __init__(self, input_dim=784, hidden_dim=256):
        super(SparseAutoencoder, self).__init__()
        
        # 编码器：两层网络，输出使用Sigmoid确保值在[0,1]之间
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),   # 784 → 256
            nn.ReLU(inplace=True),              # ReLU激活
            nn.Linear(hidden_dim, hidden_dim),  # 256 → 256（可以过完备）
            nn.Sigmoid()                        # Sigmoid：输出在(0,1)之间，方便计算KL散度
        )
        
        # 解码器：对称结构
        self.decoder = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),  # 256 → 256
            nn.ReLU(inplace=True),              # ReLU激活
            nn.Linear(hidden_dim, input_dim),   # 256 → 784
            nn.Sigmoid()                        # Sigmoid：输出像素值在[0,1]
        )
    
    def forward(self, x):
        """前向传播，同时返回编码用于计算稀疏惩罚"""
        h = self.encoder(x)                     # 编码：得到隐藏表示
        reconstructed = self.decoder(h)         # 解码：得到重构
        return reconstructed, h                 # 同时返回编码h，用于计算稀疏正则项

# ======================== KL散度稀疏惩罚计算 ========================
def kl_divergence_sparsity(encoded, target_sparsity, epsilon=1e-10):
    """
    计算KL散度稀疏惩罚：
    KL(ρ || ρ̂_j) = ρ·log(ρ/ρ̂_j) + (1-ρ)·log((1-ρ)/(1-ρ̂_j))
    
    其中：
    - ρ = target_sparsity（目标平均激活率，如0.05）
    - ρ̂_j = 实际平均激活率（对batch维度取均值）
    
    参数:
        encoded: 编码器输出，形状 [batch_size, hidden_dim]
        target_sparsity: 目标稀疏度ρ
        epsilon: 数值稳定的小常数，防止log(0)
    返回:
        标量KL散度值
    """
    # 计算每个神经元在当前batch上的平均激活率
    # dim=0 对batch维度取均值，得到形状 [hidden_dim]
    rho_hat = torch.mean(encoded, dim=0)
    
    # 将值限制在[epsilon, 1-epsilon]范围内，避免log(0)或除零
    rho_hat = torch.clamp(rho_hat, min=epsilon, max=1 - epsilon)
    
    # KL散度公式：鼓励ρ̂接近ρ
    # 当ρ̂ = ρ时，KL = 0（最小值）
    kl_div = target_sparsity * torch.log(target_sparsity / rho_hat) + \
             (1 - target_sparsity) * torch.log((1 - target_sparsity) / (1 - rho_hat))
    
    return torch.sum(kl_div)                    # 对所有隐藏神经元求和

# ======================== 模型与优化器 ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = SparseAutoencoder(input_dim, hidden_dim).to(device)
criterion = nn.MSELoss()                        # 重构损失：均方误差
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# ======================== 训练循环 ========================
total_losses = []      # 记录总损失
recon_losses = []      # 记录重构损失
sparse_losses = []     # 记录稀疏惩罚损失

for epoch in range(num_epochs):
    epoch_total = 0.0
    epoch_recon = 0.0
    epoch_sparse = 0.0
    
    for batch_idx, (data, _) in enumerate(train_loader):
        data = data.view(data.size(0), -1).to(device)  # 展平：[B,1,28,28] → [B,784]
        
        # ---- 前向传播 ----
        reconstructed, encoded = model(data)    # 获取重构和编码
        
        # ---- 计算损失 ----
        recon_loss = criterion(reconstructed, data)             # 重构损失
        kl_loss = kl_divergence_sparsity(encoded, target_sparsity)  # KL稀疏惩罚
        total_loss = recon_loss + sparse_weight * kl_loss       # 总损失 = 重构 + λ·稀疏
        
        # ---- 反向传播与优化 ----
        optimizer.zero_grad()                  # 清除旧梯度
        total_loss.backward()                  # 计算总损失的梯度
        optimizer.step()                       # 更新参数
        
        # 累加损失
        epoch_total += total_loss.item()
        epoch_recon += recon_loss.item()
        epoch_sparse += kl_loss.item()
    
    # 记录平均损失
    n = len(train_loader)
    total_losses.append(epoch_total / n)
    recon_losses.append(epoch_recon / n)
    sparse_losses.append(epoch_sparse / n)
    
    if (epoch + 1) % 5 == 0:                   # 每5个epoch打印一次
        print(f'Epoch [{epoch+1}/{num_epochs}] '
              f'Total: {total_losses[-1]:.6f}, '
              f'Recon: {recon_losses[-1]:.6f}, '
              f'Sparse: {sparse_losses[-1]:.4f}')

# ======================== 可视化损失曲线 ========================
plt.figure(figsize=(10, 4))
plt.plot(total_losses, label='Total Loss', linewidth=2)       # 总损失曲线
plt.plot(recon_losses, label='Reconstruction Loss', linewidth=2)  # 重构损失曲线
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('Sparse Autoencoder Training Loss')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

# ======================== 可视化稀疏激活 ========================
model.eval()                                    # 切换到评估模式
with torch.no_grad():
    sample = next(iter(train_loader))[0][:100]  # 取100个样本
    sample = sample.view(100, -1).to(device)
    _, h = model(sample)                        # 获取编码
    activations = h.cpu().numpy()               # 转为numpy

plt.figure(figsize=(12, 4))
plt.imshow(activations.T, cmap='hot', aspect='auto')  # 热力图：横轴=样本，纵轴=神经元
plt.xlabel('Sample Index')
plt.ylabel('Neuron Index')
plt.title('Sparse Activation Patterns (darker = more zeros)')
plt.colorbar(label='Activation')
plt.show()

三、表示能力、层的大小和深度 (Representational Capacity, Layer Size and Depth)

3.1 核心知识点

表示能力指的是自编码器能学习到的函数族的丰富程度。

关键原则：

设计选择	效果	注意事项
增加隐藏层宽度	能表示更复杂的映射	过宽可能导致恒等映射（不压缩信息）
增加网络深度	每层学到更抽象的特征	深层更高效： $O(2^n)$ 的指数表达能力
线性激活	退化为PCA	无法捕获非线性结构
非线性激活	能表示任意复杂函数	理论上万能近似定理适用

深层 vs 浅层：

具有 $k$ 个隐藏层的深度自编码器比仅用1层的浅层自编码器能用更少的参数达到相同重构质量
深度网络学到的是层次化表示：低层→边缘/纹理，高层→语义/结构

3.2 案例代码：不同深度与宽度对比

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np

# ======================== 超参数 ========================
batch_size = 128
num_epochs = 15
learning_rate = 1e-3
input_dim = 784
encoding_dim = 64                # 最终编码维度（所有模型相同以便比较）

# ======================== 数据加载 ========================
transform = transforms.Compose([transforms.ToTensor()])
train_dataset = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform, download=True
)
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset, batch_size=batch_size, shuffle=True, num_workers=2
)

# ======================== 模型1：浅而宽的自编码器 ========================
class ShallowWideAutoencoder(nn.Module):
    """
    浅而宽：1个隐藏层，宽度1024
    优点：简单，训练快
    缺点：参数多，可能过拟合，特征不够抽象
    """
    def __init__(self, input_dim=784, encoding_dim=64):
        super(ShallowWideAutoencoder, self).__init__()
        
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 1024),        # 单层宽隐藏层：784 → 1024（过完备）
            nn.ReLU(inplace=True),             # ReLU激活
            nn.Linear(1024, encoding_dim),     # 压缩到编码维度：1024 → 64
            nn.ReLU(inplace=True)
        )
        
        self.decoder = nn.Sequential(
            nn.Linear(encoding_dim, 1024),     # 64 → 1024
            nn.ReLU(inplace=True),
            nn.Linear(1024, input_dim),        # 1024 → 784
            nn.Sigmoid()                       # 输出[0,1]
        )
    
    def forward(self, x):
        return self.decoder(self.encoder(x))   # 编码后解码

# ======================== 模型2：深而窄的自编码器 ========================
class DeepNarrowAutoencoder(nn.Module):
    """
    深而窄：4个隐藏层，逐步压缩
    784 → 256 → 128 → 64 → encoding_dim
    优点：参数高效，层次化特征
    缺点：训练难度大，可能梯度消失
    """
    def __init__(self, input_dim=784, encoding_dim=64):
        super(DeepNarrowAutoencoder, self).__init__()
        
        # 编码器：逐层压缩，每层维度减半
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 256),         # 第1层：784 → 256
            nn.BatchNorm1d(256),               # 批归一化：稳定训练，加速收敛
            nn.ReLU(inplace=True),             # ReLU激活
            nn.Linear(256, 128),               # 第2层：256 → 128
            nn.BatchNorm1d(128),               # 批归一化
            nn.ReLU(inplace=True),
            nn.Linear(128, 64),                # 第3层：128 → 64
            nn.BatchNorm1d(64),
            nn.ReLU(inplace=True),
            nn.Linear(64, encoding_dim),       # 第4层：64 → encoding_dim
            nn.ReLU(inplace=True)
        )
        
        # 解码器：对称扩展
        self.decoder = nn.Sequential(
            nn.Linear(encoding_dim, 64),       # encoding_dim → 64
            nn.BatchNorm1d(64),
            nn.ReLU(inplace=True),
            nn.Linear(64, 128),                # 64 → 128
            nn.BatchNorm1d(128),
            nn.ReLU(inplace=True),
            nn.Linear(128, 256),               # 128 → 256
            nn.BatchNorm1d(256),
            nn.ReLU(inplace=True),
            nn.Linear(256, input_dim),         # 256 → 784
            nn.Sigmoid()
        )
    
    def forward(self, x):
        return self.decoder(self.encoder(x))

# ======================== 模型3：卷积自编码器（利用空间结构） ========================
class ConvAutoencoder(nn.Module):
    """
    卷积自编码器：利用图像的空间局部性
    比全连接更高效地处理2D图像数据
    编码路径：逐步减小空间尺寸，增加通道数
    解码路径：逐步增大空间尺寸，减少通道数
    """
    def __init__(self):
        super(ConvAutoencoder, self).__init__()
        
        # 编码器：卷积层逐步提取空间特征
        self.encoder = nn.Sequential(
            # 输入: [B, 1, 28, 28]
            nn.Conv2d(1, 16, kernel_size=3, stride=2, padding=1),
            # 卷积：1通道→16通道，stride=2使尺寸减半
            # 输出: [B, 16, 14, 14]
            nn.BatchNorm2d(16),                 # 通道维度批归一化
            nn.ReLU(inplace=True),
            
            nn.Conv2d(16, 32, kernel_size=3, stride=2, padding=1),
            # 卷积：16→32通道，尺寸再减半
            # 输出: [B, 32, 7, 7]
            nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
            
            nn.Conv2d(32, 64, kernel_size=7),
            # 卷积：32→64通道，7x7核匹配特征图尺寸
            # 输出: [B, 64, 1, 1] → 压缩为64维向量
        )
        
        # 解码器：转置卷积逐步恢复空间尺寸
        self.decoder = nn.Sequential(
            nn.ConvTranspose2d(64, 32, kernel_size=7),
            # 转置卷积（反卷积）：64→32通道，恢复尺寸
            # 输出: [B, 32, 7, 7]
            nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
            
            nn.ConvTranspose2d(32, 16, kernel_size=3, stride=2, padding=1, output_padding=1),
            # 转置卷积：32→16通道，stride=2使尺寸翻倍
            # 输出: [B, 16, 14, 14]
            nn.BatchNorm2d(16),
            nn.ReLU(inplace=True),
            
            nn.ConvTranspose2d(16, 1, kernel_size=3, stride=2, padding=1, output_padding=1),
            # 转置卷积：16→1通道，恢复原始尺寸
            # 输出: [B, 1, 28, 28]
            nn.Sigmoid()                        # 像素值[0,1]
        )
    
    def forward(self, x):
        z = self.encoder(x)                     # 编码：[B,1,28,28] → [B,64,1,1]
        reconstructed = self.decoder(z)         # 解码：[B,64,1,1] → [B,1,28,28]
        return reconstructed

# ======================== 训练与比较函数 ========================
def train_model(model, train_loader, num_epochs, learning_rate, device):
    """通用训练函数，返回每个epoch的损失列表"""
    model = model.to(device)
    criterion = nn.MSELoss()                    # 重构损失
    optimizer = optim.Adam(model.parameters(), lr=learning_rate)
    losses = []
    
    for epoch in range(num_epochs):
        epoch_loss = 0.0
        for data, _ in train_loader:
            if isinstance(model, ConvAutoencoder):
                # 卷积模型：保留空间维度，不展平
                data = data.to(device)
            else:
                # 全连接模型：需要展平
                data = data.view(data.size(0), -1).to(device)
            
            output = model(data)                # 前向传播
            loss = criterion(output, data)      # 计算损失
            
            optimizer.zero_grad()               # 清梯度
            loss.backward()                     # 反向传播
            optimizer.step()                    # 更新参数
            
            epoch_loss += loss.item()
        
        avg = epoch_loss / len(train_loader)
        losses.append(avg)
        if (epoch + 1) % 5 == 0:
            print(f'  Epoch [{epoch+1}/{num_epochs}], Loss: {avg:.6f}')
    
    return losses

# ======================== 模型参数量统计 ========================
def count_parameters(model):
    """统计模型可训练参数总数"""
    return sum(p.numel() for p in model.parameters() if p.requires_grad)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

# 实例化三个模型
models = {
    'Shallow-Wide (1 hidden)': ShallowWideAutoencoder(input_dim, encoding_dim),
    'Deep-Narrow (4 hidden)': DeepNarrowAutoencoder(input_dim, encoding_dim),
    'Conv-Autoencoder': ConvAutoencoder()
}

# 打印参数量对比
print("=" * 50)
print("模型参数量对比:")
print("=" * 50)
for name, model in models.items():
    params = count_parameters(model)
    print(f'{name}: {params:,} parameters')     # 千位分隔符格式化

# ======================== 训练所有模型 ========================
all_losses = {}
for name, model in models.items():
    print(f'\nTraining {name}...')
    losses = train_model(model, train_loader, num_epochs, learning_rate, device)
    all_losses[name] = losses

# ======================== 损失对比图 ========================
plt.figure(figsize=(10, 6))
for name, losses in all_losses.items():
    plt.plot(losses, label=name, linewidth=2)
plt.xlabel('Epoch', fontsize=12)
plt.ylabel('MSE Loss', fontsize=12)
plt.title('Comparison: Shallow-Wide vs Deep-Narrow vs Conv Autoencoder', fontsize=13)
plt.legend(fontsize=11)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

四、随机编码器和解码器 (Stochastic Encoder and Decoder)

4.1 核心知识点

与确定性映射不同，随机编码器/解码器将编码和解码过程建模为概率分布：

随机编码器： $\mathcal{N}(\mu(x), \sigma^2(x))$ ，输出均值和方差
随机解码器： $\mathcal{N}(\mu(h), \sigma^2)$ 或伯努利分布

这种方法是 变分自编码器（VAE） 的基础。训练目标变为变分下界（ELBO）：

$L=Eq(h∣x)[log⁡p(x∣h)]−DKL(q(h∣x)∥p(h))\mathcal{L} = \mathbb{E}_{q(h|x)}[\log p(x|h)] - D_{KL}(q(h|x) \| p(h))$

其中第一项是重构项，第二项是KL正则项，使后验 $q (h ∣ x)$ 接近先验 $p(h)=N(0,1)p(h)=\mathcal{N}(0,1)$ 。

重参数化技巧（Reparameterization Trick）：
$\mu + \sigma \cdot \epsilon, \quad \epsilon \sim \mathcal{N}(0,1)$

使得采样过程可微分，梯度可以通过 $μ\mu$ 和 $σ\sigma$ 反向传播。

4.2 案例代码：变分自编码器（VAE）

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np

# ======================== 超参数 ========================
batch_size = 128
num_epochs = 30
learning_rate = 1e-3
latent_dim = 20               # 潜在空间维度（z的维度）
input_dim = 784               # 输入维度

# ======================== 数据加载 ========================
transform = transforms.Compose([transforms.ToTensor()])
train_dataset = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform, download=True
)
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset, batch_size=batch_size, shuffle=True, num_workers=2
)

# ======================== VAE定义 ========================
class VAE(nn.Module):
    """
    变分自编码器（Variational Autoencoder）：
    - 编码器输出分布参数（μ, log σ²）而非确定性编码
    - 使用重参数化技巧使采样可微分
    - 损失 = 重构损失 + KL散度
    """
    def __init__(self, input_dim=784, latent_dim=20, hidden_dim=400):
        super(VAE, self).__init__()
        
        # ---- 编码器：输入 → 分布参数(μ, log σ²) ----
        self.encoder_fc = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),    # 784 → 400
            nn.ReLU(inplace=True),               # ReLU激活
            nn.Linear(hidden_dim, hidden_dim),   # 400 → 400
            nn.ReLU(inplace=True)
        )
        # 均值头：输出潜在分布的均值 μ
        self.fc_mu = nn.Linear(hidden_dim, latent_dim)          # 400 → 20
        # 对数方差头：输出 log(σ²)，用log保证数值稳定
        self.fc_logvar = nn.Linear(hidden_dim, latent_dim)      # 400 → 20
        
        # ---- 解码器：潜在变量 → 重构 ----
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, hidden_dim),   # 20 → 400
            nn.ReLU(inplace=True),
            nn.Linear(hidden_dim, hidden_dim),   # 400 → 400
            nn.ReLU(inplace=True),
            nn.Linear(hidden_dim, input_dim),    # 400 → 784
            nn.Sigmoid()                         # 输出[0,1]
        )
    
    def encode(self, x):
        """
        编码过程：输入 → (均值, 对数方差)
        返回分布的参数，而非确定性编码
        """
        h = self.encoder_fc(x)                   # 提取特征
        mu = self.fc_mu(h)                       # 计算均值 μ
        logvar = self.fc_logvar(h)               # 计算 log(σ²)
        return mu, logvar
    
    def reparameterize(self, mu, logvar):
        """
        重参数化技巧：
        z = μ + σ · ε，其中 ε ~ N(0,1)
        
        这样采样过程的随机性来自ε，而μ和σ是确定性的，
        梯度可以通过μ和σ反向传播
        
        参数:
            mu: 均值，形状 [batch_size, latent_dim]
            logvar: 对数方差，形状 [batch_size, latent_dim]
        返回:
            z: 潜在变量，形状 [batch_size, latent_dim]
        """
        std = torch.exp(0.5 * logvar)           # σ = exp(0.5 * log σ²) = √(σ²)
        eps = torch.randn_like(std)              # ε ~ N(0,1)，与std同形状
        z = mu + std * eps                       # 重参数化：z = μ + σ·ε
        return z
    
    def decode(self, z):
        """解码过程：潜在变量 → 重构"""
        return self.decoder(z)
    
    def forward(self, x):
        """
        完整前向传播：
        1. 编码得到分布参数
        2. 重参数化采样
        3. 解码重构
        
        返回: (重构, μ, log σ²)
        """
        mu, logvar = self.encode(x)              # 编码：获取分布参数
        z = self.reparameterize(mu, logvar)      # 采样：重参数化技巧
        reconstructed = self.decode(z)           # 解码：生成重构
        return reconstructed, mu, logvar

# ======================== VAE损失函数 ========================
def vae_loss_function(reconstructed, x, mu, logvar):
    """
    VAE损失函数 = 重构损失 + KL散度
    
    1. 重构损失：二元交叉熵或MSE
       L_recon = Σ_i [x_i log(x̂_i) + (1-x_i)log(1-x̂_i)]
    
    2. KL散度：使后验q(z|x)接近先验p(z)=N(0,I)
       L_KL = -0.5 * Σ_j (1 + log σ²_j - μ²_j - σ²_j)
    
    参数:
        reconstructed: 重构输出
        x: 原始输入
        mu: 编码器输出的均值
        logvar: 编码器输出的对数方差
    返回:
        total_loss, recon_loss, kl_loss
    """
    # 重构损失（BCE）：逐像素比较
    # reduction='sum'：对所有元素求和（而非求均值）
    recon_loss = nn.functional.binary_cross_entropy(
        reconstructed, x, reduction='sum'
    )
    
    # KL散度（解析公式）：
    # KL(N(μ,σ²) || N(0,1)) = -0.5 * Σ(1 + log(σ²) - μ² - σ²)
    # 推导：对高斯分布的KL散度有闭式解
    kl_loss = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
    # logvar.pow(2) 是错误的，应该是 logvar.exp() 即 σ²
    
    total_loss = recon_loss + kl_loss           # ELBO = -总损失
    
    return total_loss, recon_loss, kl_loss

# ======================== 模型初始化与训练 ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = VAE(input_dim, latent_dim).to(device)
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 训练
for epoch in range(num_epochs):
    model.train()                               # 训练模式
    total_loss_epoch = 0
    recon_loss_epoch = 0
    kl_loss_epoch = 0
    
    for batch_idx, (data, _) in enumerate(train_loader):
        data = data.view(data.size(0), -1).to(device)  # 展平
        
        # 前向传播
        reconstructed, mu, logvar = model(data)
        
        # 计算损失
        total_loss, recon_loss, kl_loss = vae_loss_function(
            reconstructed, data, mu, logvar
        )
        
        # 反向传播
        optimizer.zero_grad()
        total_loss.backward()
        optimizer.step()
        
        # 累加损失
        total_loss_epoch += total_loss.item()
        recon_loss_epoch += recon_loss.item()
        kl_loss_epoch += kl_loss.item()
    
    n = len(train_loader.dataset)
    if (epoch + 1) % 5 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}] '
              f'Total: {total_loss_epoch/n:.4f}, '
              f'Recon: {recon_loss_epoch/n:.4f}, '
              f'KL: {kl_loss_epoch/n:.4f}')

# ======================== 从潜在空间采样生成新图像 ========================
model.eval()                                    # 评估模式
with torch.no_grad():
    # 从标准正态分布采样随机潜在向量
    z = torch.randn(16, latent_dim).to(device)  # 16个随机向量，每个20维
    
    # 解码生成图像
    generated = model.decode(z).cpu()            # [16, 784]
    generated = generated.view(16, 1, 28, 28)   # 恢复为图像形状

# 显示生成的图像
fig, axes = plt.subplots(2, 8, figsize=(16, 4))
for i in range(16):
    ax = axes[i // 8, i % 8]
    ax.imshow(generated[i].squeeze(), cmap='gray')
    ax.axis('off')
plt.suptitle('VAE: Generated Samples from Random Latent Vectors', fontsize=14)
plt.tight_layout()
plt.show()

# ======================== 潜在空间插值 ========================
with torch.no_grad():
    # 取两个样本
    sample1, _ = train_dataset[0]               # 第一张图
    sample2, _ = train_dataset[1]               # 第二张图
    
    # 编码得到潜在向量
    s1 = sample1.view(1, -1).to(device)
    s2 = sample2.view(1, -1).to(device)
    mu1, _ = model.encode(s1)                    # 样本1的μ
    mu2, _ = model.encode(s2)                    # 样本2的μ
    
    # 在两个潜在向量之间线性插值
    n_interp = 10                                # 插值步数
    interpolated = []
    for alpha in np.linspace(0, 1, n_interp):
        z = (1 - alpha) * mu1 + alpha * mu2      # 线性插值：z = (1-t)z₁ + tz₂
        img = model.decode(z).cpu().view(28, 28) # 解码
        interpolated.append(img)

# 可视化插值结果
fig, axes = plt.subplots(1, n_interp, figsize=(20, 2.5))
for i, img in enumerate(interpolated):
    axes[i].imshow(img.numpy(), cmap='gray')
    axes[i].axis('off')
    if i == 0:
        axes[i].set_title('Sample 1', fontsize=10)
    elif i == n_interp - 1:
        axes[i].set_title('Sample 2', fontsize=10)
plt.suptitle('Latent Space Interpolation', fontsize=14)
plt.tight_layout()
plt.show()

五、去噪自编码器详解 (Denoising Autoencoder)

5.1 核心知识点

去噪自编码器（Denoising Autoencoder, DAE） 的核心思想：

对输入 $x$ 施加噪声得到 $x~\tilde{x}$
训练目标是：从损坏的输入 $x~\tilde{x}$ 重构出原始干净的输入 $x$

$g(f(\tilde{x})))$

为什么DAE能学到好的表示？

Bengio等人证明：DAE的训练等价于隐式地估计数据分布的得分函数（score function） $∇xlog⁡p(x)\nabla_x \log p(x)$ 。具体来说：

DAE学习的是数据流形的局部方向
靠近流形的点被推回到流形上
这使得DAE具有特征提取和数据生成的双重能力

常用噪声类型：

噪声类型	适用场景	参数
高斯噪声	连续数据	$x~=x+ϵ,ϵ∼N(0,σ2)\tilde{x} = x + \epsilon, \epsilon \sim \mathcal{N}(0, \sigma^2)$
掩码噪声（Dropout）	通用	随机将部分输入置零
椒盐噪声	图像	随机将部分像素设为最大/最小值

5.2 案例代码：多种噪声类型的DAE

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np

# ======================== 超参数 ========================
batch_size = 128
num_epochs = 20
learning_rate = 1e-3
input_dim = 784
encoding_dim = 128            # 编码维度
noise_factor = 0.3            # 高斯噪声强度系数
mask_ratio = 0.3              # 掩码噪声的遮蔽比例
salt_pepper_ratio = 0.1       # 椒盐噪声比例

# ======================== 数据加载 ========================
transform = transforms.Compose([transforms.ToTensor()])
train_dataset = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform, download=True
)
test_dataset = torchvision.datasets.MNIST(
    root='./data', train=False, transform=transform, download=True
)
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset, batch_size=batch_size, shuffle=True, num_workers=2
)
test_loader = torch.utils.data.DataLoader(
    dataset=test_dataset, batch_size=batch_size, shuffle=False, num_workers=2
)

# ======================== 噪声添加函数 ========================
def add_gaussian_noise(x, noise_factor=0.3):
    """
    添加高斯噪声：x̃ = x + σ·ε, ε~N(0,1)
    
    参数:
        x: 干净输入，像素值在[0,1]之间
        noise_factor: 噪声强度σ
    返回:
        加噪后并裁剪到[0,1]的输入
    """
    noise = torch.randn_like(x) * noise_factor   # 生成同形状的标准正态噪声，乘以强度
    noisy_x = x + noise                           # 叠加噪声
    noisy_x = torch.clamp(noisy_x, 0., 1.)       # 裁剪到[0,1]，确保有效像素值
    return noisy_x

def add_mask_noise(x, mask_ratio=0.3):
    """
    添加掩码噪声（类似Dropout）：随机将部分像素值设为0
    
    参数:
        x: 干净输入
        mask_ratio: 被遮蔽的像素比例（0到1之间）
    返回:
        部分像素被置零的输入
    """
    # 生成伯努利掩码：1的概率为(1-mask_ratio)，0的概率为mask_ratio
    mask = torch.bernoulli(torch.full_like(x, 1 - mask_ratio))
    # torch.bernoulli：从伯努利分布中采样，参数为每个元素为1的概率
    return x * mask                               # 逐元素相乘，被遮蔽位置变为0

def add_salt_pepper_noise(x, ratio=0.1):
    """
    添加椒盐噪声：随机将像素设为0（椒）或1（盐）
    
    参数:
        x: 干净输入
        ratio: 被污染的像素总比例
    返回:
        受椒盐噪声污染的输入
    """
    noisy_x = x.clone()                           # 克隆输入，避免修改原始数据
    num_salt = int(ratio / 2 * x.numel())         # 盐噪声数量 = 总像素 × ratio/2
    num_pepper = int(ratio / 2 * x.numel())       # 椒噪声数量 = 总像素 × ratio/2
    
    # 盐噪声：随机位置设为1（最亮）
    # torch.randint生成随机索引
    salt_idx = torch.randint(0, x.numel(), (num_salt,))
    noisy_x.view(-1)[salt_idx] = 1.0
    
    # 椒噪声：随机位置设为0（最暗）
    pepper_idx = torch.randint(0, x.numel(), (num_pepper,))
    noisy_x.view(-1)[pepper_idx] = 0.0
    
    return noisy_x

# ======================== 去噪自编码器模型 ========================
class DenoisingAutoencoder(nn.Module):
    """
    去噪自编码器：
    - 输入：带噪声的数据
    - 目标：重构出干净数据
    - 编码器和解码器使用对称结构
    - 使用Dropout作为正则化手段
    """
    def __init__(self, input_dim=784, encoding_dim=128):
        super(DenoisingAutoencoder, self).__init__()
        
        # 编码器：带Dropout的深层网络
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 512),           # 784 → 512
            nn.BatchNorm1d(512),                 # 批归一化
            nn.ReLU(inplace=True),
            nn.Dropout(0.2),                     # Dropout：训练时随机丢弃20%神经元
            
            nn.Linear(512, 256),                 # 512 → 256
            nn.BatchNorm1d(256),
            nn.ReLU(inplace=True),
            nn.Dropout(0.2),
            
            nn.Linear(256, encoding_dim),        # 256 → 128
            nn.ReLU(inplace=True)
        )
        
        # 解码器：对称结构（解码器通常不用Dropout）
        self.decoder = nn.Sequential(
            nn.Linear(encoding_dim, 256),        # 128 → 256
            nn.BatchNorm1d(256),
            nn.ReLU(inplace=True),
            
            nn.Linear(256, 512),                 # 256 → 512
            nn.BatchNorm1d(512),
            nn.ReLU(inplace=True),
            
            nn.Linear(512, input_dim),           # 512 → 784
            nn.Sigmoid()                         # 输出[0,1]
        )
    
    def forward(self, x):
        """
        前向传播：
        x(带噪) → encoder → 编码(去噪特征) → decoder → 重构(干净)
        """
        encoded = self.encoder(x)                # 编码：从噪声输入中提取有用特征
        decoded = self.decoder(encoded)          # 解码：生成干净的重构
        return decoded

# ======================== 模型初始化 ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = DenoisingAutoencoder(input_dim, encoding_dim).to(device)
criterion = nn.MSELoss()                         # 重构损失
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# ======================== 训练循环 ========================
train_losses = []

for epoch in range(num_epochs):
    model.train()                                # 训练模式（启用Dropout）
    epoch_loss = 0.0
    
    for batch_idx, (data, _) in enumerate(train_loader):
        data = data.view(data.size(0), -1).to(device)  # 展平
        
        # ---- 添加噪声 ----
        # 混合使用多种噪声，增强鲁棒性
        noisy_data = add_gaussian_noise(data, noise_factor)  # 高斯噪声
        
        # ---- 前向传播 ----
        reconstructed = model(noisy_data)        # 从带噪输入重构
        
        # 注意：损失是与干净数据比较，不是与噪声数据比较！
        loss = criterion(reconstructed, data)    # 目标是恢复干净数据
        
        # ---- 反向传播 ----
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        epoch_loss += loss.item()
    
    avg_loss = epoch_loss / len(train_loader)
    train_losses.append(avg_loss)
    
    if (epoch + 1) % 5 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {avg_loss:.6f}')

# ======================== 测试与可视化 ========================
model.eval()                                     # 评估模式（禁用Dropout）
with torch.no_grad():
    # 取一批测试数据
    test_data, _ = next(iter(test_loader))
    test_data = test_data.view(test_data.size(0), -1).to(device)
    
    # 用不同噪声类型测试
    gaussian_noisy = add_gaussian_noise(test_data, noise_factor)
    mask_noisy = add_mask_noise(test_data, mask_ratio)
    sp_noisy = add_salt_pepper_noise(test_data, salt_pepper_ratio)
    
    # 从各噪声版本重构
    recon_gaussian = model(gaussian_noisy)
    recon_mask = model(mask_noisy)
    recon_sp = model(sp_noisy)

# 可视化：原始 → 各噪声版本 → 各重构版本
n = 6  # 展示6个样本
fig, axes = plt.subplots(5, n, figsize=(15, 12))
row_labels = ['Original', 'Gaussian Noise', 'Recon from Gauss',
              'Mask Noise', 'Recon from Mask']

for i in range(n):
    axes[0, i].imshow(test_data[i].cpu().view(28,28), cmap='gray')
    axes[1, i].imshow(gaussian_noisy[i].cpu().view(28,28), cmap='gray')
    axes[2, i].imshow(recon_gaussian[i].cpu().view(28,28), cmap='gray')
    axes[3, i].imshow(mask_noisy[i].cpu().view(28,28), cmap='gray')
    axes[4, i].imshow(recon_mask[i].cpu().view(28,28), cmap='gray')

for i, label in enumerate(row_labels):
    axes[i, 0].set_ylabel(label, fontsize=11, rotation=0, labelpad=100)

for ax_row in axes:
    for ax in ax_row:
        ax.axis('off')

plt.suptitle('Denoising Autoencoder: Multiple Noise Types', fontsize=15)
plt.tight_layout()
plt.show()

六、使用自编码器学习流形 (Learning Manifolds with Autoencoders)

6.1 核心知识点

流形假设（Manifold Hypothesis）：高维数据（如图像）实际上集中分布在低维流形附近。

自编码器学习流形的机制：

编码器：将高维数据投影到低维流形坐标（流形的参数化）
解码器：将低维坐标映射回高维空间（流形的生成）
重构误差最小化迫使解码器的输出空间逼近数据流形

流形学习的关键特性：

流形是局部欧几里得的：流形上相近的点映射到编码空间中也相近
流形有内在维度：低于数据的嵌入维度（如MNIST图像784维，但流形可能只有10-20维）
解码器定义了一个从低维到高维的光滑映射

6.2 案例代码：流形学习可视化

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np
from sklearn.manifold import TSNE              # t-SNE降维可视化

# ======================== 超参数 ========================
batch_size = 128
num_epochs = 30
learning_rate = 1e-3
latent_dim = 2                 # 设为2维，方便直接可视化流形

# ======================== 数据加载 ========================
transform = transforms.Compose([transforms.ToTensor()])
train_dataset = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform, download=True
)
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset, batch_size=batch_size, shuffle=True, num_workers=2
)

# ======================== 2D流形学习自编码器 ========================
class ManifoldAutoencoder(nn.Module):
    """
    将数据压缩到2维潜在空间，直接可视化流形结构
    使用更深的网络来补偿极低的编码维度
    """
    def __init__(self, latent_dim=2):
        super(ManifoldAutoencoder, self).__init__()
        
        # 编码器：784维 → 2维
        self.encoder = nn.Sequential(
            nn.Linear(784, 512),                 # 784 → 512
            nn.LeakyReLU(0.2, inplace=True),    # LeakyReLU：负半轴斜率0.2，避免死神经元
            nn.Linear(512, 256),                 # 512 → 256
            nn.LeakyReLU(0.2, inplace=True),
            nn.Linear(256, 128),                 # 256 → 128
            nn.LeakyReLU(0.2, inplace=True),
            nn.Linear(128, latent_dim)           # 128 → 2（极低维流形坐标）
        )
        
        # 解码器：2维 → 784维
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, 128),          # 2 → 128
            nn.LeakyReLU(0.2, inplace=True),
            nn.Linear(128, 256),                 # 128 → 256
            nn.LeakyReLU(0.2, inplace=True),
            nn.Linear(256, 512),                 # 256 → 512
            nn.LeakyReLU(0.2, inplace=True),
            nn.Linear(512, 784),                 # 512 → 784
            nn.Sigmoid()
        )
    
    def forward(self, x):
        z = self.encoder(x)                      # 编码到流形坐标
        reconstructed = self.decoder(z)          # 从流形坐标重构
        return reconstructed, z                  # 同时返回潜在坐标用于可视化

# ======================== 训练 ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = ManifoldAutoencoder(latent_dim).to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.5)
# 学习率调度器：每10个epoch学习率减半，帮助后期精细调整

for epoch in range(num_epochs):
    model.train()
    epoch_loss = 0.0
    
    for data, _ in train_loader:
        data = data.view(data.size(0), -1).to(device)
        reconstructed, z = model(data)
        loss = criterion(reconstructed, data)
        
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
    
    scheduler.step()                             # 更新学习率
    avg_loss = epoch_loss / len(train_loader)
    
    if (epoch + 1) % 5 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {avg_loss:.6f}, '
              f'LR: {scheduler.get_last_lr()[0]:.6f}')

# ======================== 流形可视化 ========================
model.eval()
all_z = []                                       # 存储所有样本的2D坐标
all_labels = []                                  # 存储对应标签

with torch.no_grad():
    for data, labels in train_loader:
        data = data.view(data.size(0), -1).to(device)
        _, z = model(data)                       # 获取2D潜在坐标
        all_z.append(z.cpu().numpy())
        all_labels.append(labels.numpy())

all_z = np.concatenate(all_z, axis=0)            # 拼接所有batch的坐标
all_labels = np.concatenate(all_labels, axis=0)

# 散点图：每个数字用不同颜色
plt.figure(figsize=(12, 10))
scatter = plt.scatter(
    all_z[:, 0],                                 # x坐标
    all_z[:, 1],                                 # y坐标
    c=all_labels,                                # 颜色按数字标签映射
    cmap='tab10',                                # 10色离散色图
    alpha=0.5,                                   # 透明度
    s=3                                          # 点大小
)
plt.colorbar(scatter, ticks=range(10), label='Digit')
plt.xlabel('Latent Dimension 1', fontsize=12)
plt.ylabel('Latent Dimension 2', fontsize=12)
plt.title('Learned 2D Manifold of MNIST', fontsize=14)
plt.grid(True, alpha=0.3)
plt.show()

# ======================== 流形网格采样：生成数字全景图 ========================
with torch.no_grad():
    # 在2D潜在空间中创建均匀网格
    n_points = 20                                # 每个维度20个点
    # 设定网格范围（根据数据分布调整）
    z1_range = np.linspace(-4, 4, n_points)      # 维度1的范围
    z2_range = np.linspace(-4, 4, n_points)      # 维度2的范围
    
    # 创建网格
    grid_z1, grid_z2 = np.meshgrid(z1_range, z2_range)
    # meshgrid生成2D网格坐标矩阵，每个形状[n_points, n_points]
    
    # 展平为一维并组合
    z_grid = np.column_stack([grid_z1.ravel(), grid_z2.ravel()])
    # ravel()展平为1D，column_stack组合为[N, 2]
    
    # 转为Tensor并解码
    z_tensor = torch.FloatTensor(z_grid).to(device)
    generated = model.decoder(z_tensor).cpu()     # [N, 784]
    generated = generated.view(-1, 1, 28, 28)     # [N, 1, 28, 28]

# 拼接为大图
fig, axes = plt.subplots(n_points, n_points, figsize=(20, 20))
for i in range(n_points):
    for j in range(n_points):
        idx = i * n_points + j                   # 网格索引映射到展平索引
        axes[i, j].imshow(generated[idx].squeeze(), cmap='gray')
        axes[i, j].axis('off')

plt.suptitle('Manifold Grid: Decoded Digits Across 2D Latent Space', fontsize=16, y=0.98)
plt.tight_layout()
plt.show()

七、收缩自编码器 (Contractive Autoencoder, CAE)

7.1 核心知识点

收缩自编码器（CAE） 通过惩罚编码器对输入的Jacobian矩阵来学习鲁棒表示。

核心损失函数：

$LCAE=L(x,g(f(x)))+λ∥∂f(x)∂x∥F2L_{CAE} = L(x, g(f(x))) + \lambda \left\| \frac{\partial f(x)}{\partial x} \right\|_F^2$

其中 $∥Jf(x)∥F2\left\| J_f(x) \right\|_F^2$ 是编码器函数 $f$ 关于输入 $x$ 的 Jacobian 矩阵的 Frobenius 范数。

直观理解：

Jacobian 衡量编码对输入微小变化的敏感度
惩罚 Jacobian 意味着：输入微小变化不应导致编码大幅变化
这迫使编码器学习对噪声鲁棒的特征
与DAE不同，CAE是解析地（而非随机地）实现对输入扰动的不变性

与DAE的联系：

DAE通过随机噪声实现对输入扰动的鲁棒性
CAE通过Jacobian惩罚直接约束编码函数的局部几何
当噪声趋于零时，DAE的去噪误差等价于CAE的收缩惩罚

7.2 案例代码：收缩自编码器

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np

# ======================== 超参数 ========================
batch_size = 128
num_epochs = 20
learning_rate = 1e-3
input_dim = 784
encoding_dim = 64
contractive_weight = 1e-4     # 收缩正则化系数λ

# ======================== 数据加载 ========================
transform = transforms.Compose([transforms.ToTensor()])
train_dataset = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform, download=True
)
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset, batch_size=batch_size, shuffle=True, num_workers=2
)

# ======================== 收缩自编码器定义 ========================
class ContractiveAutoencoder(nn.Module):
    """
    收缩自编码器：
    - 编码器输出h = σ(Wx + b)
    - 收缩惩罚 = ||∂h/∂x||²_F
    - 对于sigmoid激活：||J||²_F = Σ_j h_j²(1-h_j)² Σ_i W_ij²
    """
    def __init__(self, input_dim=784, encoding_dim=64):
        super(ContractiveAutoencoder, self).__init__()
        
        # 编码器：单层线性变换 + Sigmoid激活
        # 使用单层是为了方便解析计算Jacobian
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, encoding_dim),  # 784 → 64
            nn.Sigmoid()                         # Sigmoid激活
        )
        
        # 解码器
        self.decoder = nn.Sequential(
            nn.Linear(encoding_dim, input_dim),  # 64 → 784
            nn.Sigmoid()
        )
        
        # 保存编码器第一个线性层的引用（用于获取权重矩阵W）
        self.encoder_linear = self.encoder[0]    # nn.Linear(784, 64)
    
    def forward(self, x):
        h = self.encoder(x)                      # 编码
        reconstructed = self.decoder(h)          # 解码
        return reconstructed, h
    
    def contractive_loss(self, h):
        """
        计算收缩正则化项：
        
        对于单层Sigmoid编码器 h = σ(Wx + b)：
        Jacobian J = ∂h/∂x = diag(h⊙(1-h)) · W
        
        ||J||²_F = Σ_j (h_j(1-h_j))² · ||W_j||²
        
        其中W_j是权重矩阵的第j行
        """
        # 获取权重矩阵 W，形状 [encoding_dim, input_dim]
        W = self.encoder_linear.weight           # 权重矩阵
        
        # 计算 h_j * (1 - h_j)，形状 [batch_size, encoding_dim]
        # 这是sigmoid的导数：σ'(z) = σ(z)(1 - σ(z))
        h_deri = h * (1 - h)
        
        # Frobenius范数的平方
        # ||J||²_F = Σ_j [h_j(1-h_j)]² · ||W_j||²
        # 对batch取均值
        
        # (h_deri)² 的形状：[batch_size, encoding_dim]
        # W² 对每行求和得到 ||W_j||²，形状：[encoding_dim]
        # 两者逐元素相乘后求和
        
        # W.pow(2).sum(dim=1)：对每个输出神经元，计算其权重向量的L2范数平方
        # 形状：[encoding_dim]
        W_squared_sum = W.pow(2).sum(dim=1)      # ||W_j||² for each j
        
        # 收缩惩罚 = (1/batch) * Σ_sample Σ_j [h_j(1-h_j)]² · ||W_j||²
        contractive = torch.sum(
            h_deri.pow(2) * W_squared_sum,        # [B, D] * [D] 广播
            dim=1                                  # 对编码维度求和
        ).mean()                                   # 对batch取均值
        
        return contractive

# ======================== 模型初始化 ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = ContractiveAutoencoder(input_dim, encoding_dim).to(device)
criterion = nn.MSELoss()                         # 重构损失
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# ======================== 训练循环 ========================
losses_recon = []
losses_contractive = []
losses_total = []

for epoch in range(num_epochs):
    model.train()
    epoch_recon = 0.0
    epoch_contractive = 0.0
    epoch_total = 0.0
    
    for data, _ in train_loader:
        data = data.view(data.size(0), -1).to(device)
        
        # 前向传播
        reconstructed, h = model(data)
        
        # 计算各项损失
        recon_loss = criterion(reconstructed, data)              # 重构损失
        contra_loss = model.contractive_loss(h)                  # 收缩正则项
        total_loss = recon_loss + contractive_weight * contra_loss  # 总损失
        
        # 反向传播
        optimizer.zero_grad()
        total_loss.backward()
        optimizer.step()
        
        epoch_recon += recon_loss.item()
        epoch_contractive += contra_loss.item()
        epoch_total += total_loss.item()
    
    n = len(train_loader)
    losses_recon.append(epoch_recon / n)
    losses_contractive.append(epoch_contractive / n)
    losses_total.append(epoch_total / n)
    
    if (epoch + 1) % 5 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}] '
              f'Recon: {losses_recon[-1]:.6f}, '
              f'Contractive: {losses_contractive[-1]:.6f}, '
              f'Total: {losses_total[-1]:.6f}')

# ======================== 可视化损失分量 ========================
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))

ax1.plot(losses_recon, label='Reconstruction', linewidth=2)
ax1.plot(losses_total, label='Total', linewidth=2)
ax1.set_xlabel('Epoch')
ax1.set_ylabel('Loss')
ax1.set_title('Reconstruction & Total Loss')
ax1.legend()
ax1.grid(True, alpha=0.3)

ax2.plot(losses_contractive, label='Contractive Penalty', linewidth=2, color='red')
ax2.set_xlabel('Epoch')
ax2.set_ylabel('Contractive Loss')
ax2.set_title('Contractive Regularization Term')
ax2.legend()
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

# ======================== 学习到的编码器权重可视化 ========================
model.eval()
# 获取编码器权重矩阵 W，形状 [encoding_dim, input_dim]
W = model.encoder_linear.weight.data.cpu().numpy()

# 可视化前16个编码器神经元学到的权重（可视为特征检测器）
fig, axes = plt.subplots(2, 8, figsize=(16, 4))
for i in range(16):
    ax = axes[i // 8, i % 8]
    weight_img = W[i].reshape(28, 28)           # 将78维权重重塑为28x28图像
    ax.imshow(weight_img, cmap='RdBu_r',        # 红蓝双色图：正红负蓝
              vmin=-weight_img.std()*2,          # 对称色标
              vmax=weight_img.std()*2)
    ax.axis('off')
    ax.set_title(f'Neuron {i}', fontsize=8)

plt.suptitle('Contractive AE: Learned Encoder Weight Patterns', fontsize=14)
plt.tight_layout()
plt.show()

八、预测稀疏分解 (Predictive Sparse Decomposition, PSD)

8.1 核心知识点

预测稀疏分解（PSD） 结合了稀疏编码和自编码器的思想：

稀疏编码：对于输入 $x$ ，寻找稀疏系数 $h$ 使得 $\approx Dh$ ，其中 $D$ 是字典
预测器：训练一个编码器 $f (x)$ 来预测最优稀疏系数，避免在线优化

目标函数：

$min⁡D,θ∑i[∥xi−Dhi∥22+λ∥hi∥1+γ∥hi−fθ(xi)∥22]\min_{D, \theta} \sum_i \left[ \|x_i - D h_i\|_2^2 + \lambda \|h_i\|_1 + \gamma \|h_i - f_\theta(x_i)\|_2^2 \right]$

其中：

$x - Dh\|_2^2$ ：重构误差（字典 $D$ 的约束通常有 $∥D⋅j∥2≤1\|D_{\cdot j}\|_2 \leq 1$ ）
$λ∥h∥1\lambda \|h\|_1$ ：稀疏惩罚（L1范数，鼓励大部分 $h_j=0$ ）
$γ∥h−f(x)∥22\gamma \|h - f(x)\|_2^2$ ：预测一致性惩罚（编码器输出应接近稀疏编码）

训练过程交替进行：

固定 $D$ 和 $θ\theta$ ，对每个样本求解 $h$ （LASSO）
固定 $h$ ，更新 $D$ （字典学习）和 $θ\theta$ （编码器训练）

8.2 案例代码：PSD简化实现

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np
from sklearn.linear_model import Lasso          # LASSO用于稀疏编码求解

# ======================== 超参数 ========================
batch_size = 128
num_epochs = 15
learning_rate = 1e-3
input_dim = 784
n_atoms = 256                # 字典原子数（字典D的列数）
sparse_lambda = 0.1          # 稀疏正则化系数λ
predictor_weight = 0.1       # 预测一致性系数γ
max_iter_lasso = 100         # LASSO最大迭代次数

# ======================== 数据加载 ========================
transform = transforms.Compose([transforms.ToTensor()])
train_dataset = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform, download=True
)
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset, batch_size=batch_size, shuffle=True, num_workers=2
)

# ======================== 稀疏编码预测器（编码器） ========================
class SparsePredictor(nn.Module):
    """
    预测网络：直接从输入预测稀疏编码
    替代在线LASSO求解，实现快速推理
    """
    def __init__(self, input_dim=784, n_atoms=256):
        super(SparsePredictor, self).__init__()
        
        self.network = nn.Sequential(
            nn.Linear(input_dim, 512),           # 784 → 512
            nn.ReLU(inplace=True),
            nn.Linear(512, 256),                 # 512 → 256
            nn.ReLU(inplace=True),
            nn.Linear(256, n_atoms),             # 256 → 256（预测稀疏编码维度）
            nn.Sigmoid()                         # Sigmoid：输出在(0,1)之间
            # 注意：这里用Sigmoid代替ReLU以获得有界的稀疏编码
        )
    
    def forward(self, x):
        """输入→预测的稀疏编码"""
        return self.network(x)

# ======================== PSD模型 ========================
class PSDModel:
    """
    预测稀疏分解模型：
    - 字典D：可学习参数
    - 预测器f_θ：神经网络
    - 交替训练：稀疏编码→字典更新→预测器更新
    """
    def __init__(self, input_dim, n_atoms, device):
        self.device = device
        self.input_dim = input_dim
        self.n_atoms = n_atoms
        
        # 初始化字典D：使用随机正交初始化
        # D的形状：[input_dim, n_atoms]，每列是一个"原子"
        self.dictionary = nn.Parameter(
            torch.randn(input_dim, n_atoms, device=device) * 0.01
        )
        # 约束字典列的范数不超过1
        self._normalize_dictionary()
        
        # 预测器网络
        self.predictor = SparsePredictor(input_dim, n_atoms).to(device)
        
        # 优化器：同时优化字典和预测器参数
        self.optimizer = optim.Adam(
            [
                {'params': [self.dictionary], 'lr': learning_rate},      # 字典参数
                {'params': self.predictor.parameters(), 'lr': learning_rate}  # 预测器参数
            ],
            weight_decay=1e-5
        )
    
    def _normalize_dictionary(self):
        """将字典的每一列归一化到单位范数"""
        with torch.no_grad():
            # dim=0 对行求范数（每列的L2范数）
            norms = torch.norm(self.dictionary, dim=0, keepdim=True)
            norms = torch.clamp(norms, min=1e-8)   # 防止除零
            self.dictionary.data /= norms            # 归一化
    
    def sparse_encode_batch(self, x, n_iters=50, lr=0.01):
        """
        使用ISTA（迭代收缩阈值算法）对batch进行稀疏编码
        
        ISTA更新规则：
        h^(t+1) = S_λ(h^(t) - α·∇L/∇h)
        其中S_λ是软阈值函数：S_λ(x) = sign(x)·max(|x|-λ, 0)
        
        参数:
            x: 输入数据 [batch_size, input_dim]
            n_iters: ISTA迭代次数
            lr: ISTA学习率
        返回:
            h: 稀疏编码 [batch_size, n_atoms]
        """
        B = x.shape[0]
        h = torch.zeros(B, self.n_atoms, device=self.device)  # 初始化稀疏编码为零
        
        D = self.dictionary                                  # [input_dim, n_atoms]
        DtD = D.t() @ D                                      # [n_atoms, n_atoms] 预计算
        Dtx = D.t() @ x.t()                                  # [n_atoms, batch_size]
        Dtx = Dtx.t()                                        # [batch_size, n_atoms]
        
        for _ in range(n_iters):
            # 梯度下降步：∇_h ||x - Dh||² = -2D^T(x - Dh) = 2(DtD·h - Dtx)
            grad = 2.0 * (h @ DtD - Dtx)                    # [B, n_atoms]
            h = h - lr * grad                                # 梯度下降
            
            # 软阈值：S_λ(h) = sign(h) * max(|h| - λ, 0)
            h = torch.sign(h) * torch.clamp(torch.abs(h) - sparse_lambda * lr, min=0)
        
        return h
    
    def train_step(self, x):
        """
        一步训练：
        1. 用当前字典求稀疏编码h（ISTA）
        2. 用预测器预测h_pred
        3. 更新字典和预测器
        """
        # ---- 步骤1：稀疏编码 ----
        with torch.no_grad():
            h_sparse = self.sparse_encode_batch(x, n_iters=50)  # ISTA求解
        
        # ---- 步骤2：预测 ----
        h_pred = self.predictor(x)                   # 预测器输出
        
        # ---- 步骤3：计算损失 ----
        # 重构损失：||x - D·h||²
        x_recon = h_pred @ self.dictionary.t()       # [B, input_dim]
        recon_loss = nn.functional.mse_loss(x_recon, x)
        
        # 稀疏惩罚：||h||₁
        sparsity_loss = torch.mean(torch.abs(h_pred))
        
        # 预测一致性：||h_pred - h_sparse||²
        consistency_loss = nn.functional.mse_loss(h_pred, h_sparse)
        
        # 总损失
        total_loss = recon_loss + sparse_lambda * sparsity_loss + \
                     predictor_weight * consistency_loss
        
        # ---- 步骤4：更新 ----
        self.optimizer.zero_grad()
        total_loss.backward()
        self.optimizer.step()
        self._normalize_dictionary()                  # 归一化字典
        
        return total_loss.item(), recon_loss.item(), sparsity_loss.item()

# ======================== 训练PSD ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
psd = PSDModel(input_dim, n_atoms, device)

for epoch in range(num_epochs):
    epoch_total = 0.0
    epoch_recon = 0.0
    epoch_sparse = 0.0
    
    for data, _ in train_loader:
        data = data.view(data.size(0), -1).to(device)
        t_loss, r_loss, s_loss = psd.train_step(data)
        epoch_total += t_loss
        epoch_recon += r_loss
        epoch_sparse += s_loss
    
    n = len(train_loader)
    if (epoch + 1) % 3 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}] '
              f'Total: {epoch_total/n:.6f}, '
              f'Recon: {epoch_recon/n:.6f}, '
              f'Sparsity: {epoch_sparse/n:.6f}')

# ======================== 可视化学到的字典原子 ========================
# 取字典前64个原子，可视化为28x28图像
D = psd.dictionary.data.cpu().numpy()               # [784, 256]

fig, axes = plt.subplots(4, 16, figsize=(20, 5))
for i in range(64):
    ax = axes[i // 16, i % 16]
    atom = D[:, i].reshape(28, 28)                   # 第i个原子重塑为图像
    # 归一化到[0,1]便于显示
    atom = (atom - atom.min()) / (atom.max() - atom.min() + 1e-8)
    ax.imshow(atom, cmap='gray')
    ax.axis('off')

plt.suptitle('PSD: Learned Dictionary Atoms (First 64)', fontsize=14)
plt.tight_layout()
plt.show()

# ======================== 稀疏编码可视化 ========================
with torch.no_grad():
    sample = next(iter(train_loader))[0][:10].to(device)
    sample_flat = sample.view(10, -1)
    
    # 获取预测器的稀疏编码
    h_pred = psd.predictor(sample_flat).cpu().numpy()  # [10, 256]
    # 获取实际的稀疏编码（ISTA）
    h_sparse = psd.sparse_encode_batch(sample_flat).cpu().numpy()

fig, axes = plt.subplots(2, 10, figsize=(20, 5))
for i in range(10):
    # 上行：预测的稀疏编码
    axes[0, i].bar(range(n_atoms), h_pred[i], width=1.0, color='steelblue')
    axes[0, i].set_xlim(0, n_atoms)
    axes[0, i].set_ylim(0, 1)
    axes[0, i].axis('off')
    if i == 0:
        axes[0, i].set_ylabel('Predicted', fontsize=10)
    
    # 下行：ISTA求解的稀疏编码
    axes[1, i].bar(range(n_atoms), h_sparse[i], width=1.0, color='coral')
    axes[1, i].set_xlim(0, n_atoms)
    axes[1, i].set_ylim(0, h_sparse.max() * 1.1)
    axes[1, i].axis('off')
    if i == 0:
        axes[1, i].set_ylabel('ISTA Sparse', fontsize=10)

plt.suptitle('Top: Predicted Sparse Codes | Bottom: ISTA Sparse Codes', fontsize=14)
plt.tight_layout()
plt.show()

九、自编码器的应用 (Applications of Autoencoders)

9.1 核心知识点

自编码器在实际中有广泛的应用：

应用领域	使用方式	具体场景
数据降维	编码器输出作为低维特征	可视化、聚类、预处理
异常检测	正常数据训练，异常数据重构误差大	工业缺陷检测、网络入侵
图像去噪	DAE从噪声图像恢复干净图像	医学影像、老照片修复
数据压缩	编码器压缩，解码器解压	有损压缩、特征存储
迁移学习	预训练编码器作为特征提取器	跨域学习、少样本学习
数据生成	从潜在空间采样生成新数据	VAE生成图像

9.2 案例代码一：异常检测

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np

# ======================== 超参数 ========================
batch_size = 128
num_epochs = 20
learning_rate = 1e-3
normal_class = 0              # 正常类别：数字0
anomaly_threshold_percentile = 95  # 异常阈值：重构误差的95百分位

# ======================== 数据准备：正常 vs 异常 ========================
transform = transforms.Compose([transforms.ToTensor()])

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

# 只用正常类别（数字0）训练自编码器
# 筛选出标签为normal_class的样本
train_mask = train_dataset.targets == normal_class
train_data_normal = train_dataset.data[train_mask].float() / 255.0  # 归一化到[0,1]

# 创建DataLoader
train_loader = torch.utils.data.DataLoader(
    torch.utils.data.TensorDataset(train_data_normal, train_data_normal),
    batch_size=batch_size, shuffle=True
)

# ======================== 异常检测自编码器 ========================
class AnomalyDetectorAE(nn.Module):
    """
    用于异常检测的自编码器：
    - 仅用正常数据训练
    - 正常数据→低重构误差
    - 异常数据→高重构误差
    """
    def __init__(self):
        super(AnomalyDetectorAE, self).__init__()
        
        self.encoder = nn.Sequential(
            nn.Linear(784, 256),                 # 784 → 256
            nn.ReLU(inplace=True),
            nn.Linear(256, 64),                  # 256 → 64
            nn.ReLU(inplace=True),
            nn.Linear(64, 16),                   # 64 → 16（高度压缩）
            nn.ReLU(inplace=True)
        )
        
        self.decoder = nn.Sequential(
            nn.Linear(16, 64),                   # 16 → 64
            nn.ReLU(inplace=True),
            nn.Linear(64, 256),                  # 64 → 256
            nn.ReLU(inplace=True),
            nn.Linear(256, 784),                 # 256 → 784
            nn.Sigmoid()
        )
    
    def forward(self, x):
        z = self.encoder(x)
        return self.decoder(z), z

# ======================== 训练 ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = AnomalyDetectorAE().to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

for epoch in range(num_epochs):
    model.train()
    epoch_loss = 0.0
    for data, target in train_loader:
        data = data.view(data.size(0), -1).to(device)
        target = target.view(target.size(0), -1).to(device)
        
        reconstructed, _ = model(data)
        loss = criterion(reconstructed, target)
        
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
    
    if (epoch + 1) % 5 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {epoch_loss/len(train_loader):.6f}')

# ======================== 异常检测评估 ========================
model.eval()
test_errors = []                                  # 存储每个样本的重构误差
test_labels = []

with torch.no_grad():
    for i in range(len(test_dataset)):
        data, label = test_dataset[i]
        data_flat = data.view(1, -1).to(device)
        reconstructed, _ = model(data_flat)
        
        # 计算每个样本的MSE
        mse = torch.mean((data_flat - reconstructed) ** 2).item()
        test_errors.append(mse)
        test_labels.append(label)

test_errors = np.array(test_errors)
test_labels = np.array(test_labels)

# 设定阈值：正常类数据重构误差的某个百分位
normal_errors = test_errors[test_labels == normal_class]
threshold = np.percentile(normal_errors, anomaly_threshold_percentile)
print(f'\nAnomaly Threshold (P{anomaly_threshold_percentile}): {threshold:.6f}')

# 计算检测准确率
# 正常类应该低于阈值，异常类应该高于阈值
predictions = test_errors > threshold             # True = 预测为异常
is_anomaly = test_labels != normal_class           # True = 实际为异常

# 精确率和召回率
true_positive = np.sum(predictions & is_anomaly)      # 真正例：异常且被正确检测
false_positive = np.sum(predictions & ~is_anomaly)    # 假正例：正常但被误报
false_negative = np.sum(~predictions & is_anomaly)    # 假负例：异常但未被检测

precision = true_positive / (true_positive + false_positive + 1e-8)
recall = true_positive / (true_positive + false_negative + 1e-8)
f1 = 2 * precision * recall / (precision + recall + 1e-8)

print(f'Precision: {precision:.4f}')
print(f'Recall: {recall:.4f}')
print(f'F1 Score: {f1:.4f}')

# ======================== 可视化：重构误差分布 ========================
fig, axes = plt.subplots(1, 2, figsize=(14, 5))

# 左图：不同数字的重构误差箱线图
errors_by_class = [test_errors[test_labels == d] for d in range(10)]
bp = axes[0].boxplot(errors_by_class, labels=range(10), patch_artist=True)
axes[0].axhline(y=threshold, color='r', linestyle='--', linewidth=2, label=f'Threshold={threshold:.4f}')
axes[0].set_xlabel('Digit', fontsize=12)
axes[0].set_ylabel('Reconstruction Error', fontsize=12)
axes[0].set_title('Reconstruction Error by Class', fontsize=13)
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 右图：正常vs异常的误差直方图
axes[1].hist(normal_errors, bins=50, alpha=0.7, label='Normal (digit 0)', density=True)
axes[1].hist(test_errors[is_anomaly], bins=50, alpha=0.7, label='Anomaly (other digits)', density=True)
axes[1].axvline(x=threshold, color='r', linestyle='--', linewidth=2, label='Threshold')
axes[1].set_xlabel('Reconstruction Error', fontsize=12)
axes[1].set_ylabel('Density', fontsize=12)
axes[1].set_title('Normal vs Anomaly Error Distribution', fontsize=13)
axes[1].legend()
axes[1].grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

9.3 案例代码二：图像超分辨率自编码器

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt

# ======================== 超参数 ========================
batch_size = 64
num_epochs = 20
learning_rate = 1e-3

# ======================== 数据预处理 ========================
# 同时获取高分辨率(28x28)和低分辨率(14x14)版本
transform_hr = transforms.Compose([transforms.ToTensor()])     # 高分辨率目标
transform_lr = transforms.Compose([
    transforms.Resize(14),                  # 缩小到14x14（低分辨率输入）
    transforms.Resize(28),                  # 再放大回28x28（双线性插值，模糊）
    transforms.ToTensor()
])

train_dataset_hr = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform_hr, download=True
)
train_dataset_lr = torchvision.datasets.MNIST(
    root='./data', train=True, transform=transform_lr, download=True
)

train_loader = torch.utils.data.DataLoader(
    list(zip(train_dataset_lr, train_dataset_hr)),  # 将低分辨率和高分辨率配对
    batch_size=batch_size, shuffle=True
)

# ======================== 超分辨率自编码器 ========================
class SuperResolutionAE(nn.Module):
    """
    超分辨率自编码器：
    - 输入：低分辨率图像（模糊的28x28）
    - 输出：高分辨率重构（清晰的28x28）
    - 使用卷积结构保持空间信息
    """
    def __init__(self):
        super(SuperResolutionAE, self).__init__()
        
        # 编码器：从低分辨率提取特征
        self.encoder = nn.Sequential(
            nn.Conv2d(1, 32, kernel_size=3, padding=1),    # [B,1,28,28] → [B,32,28,28]
            nn.LeakyReLU(0.2, inplace=True),
            nn.Conv2d(32, 64, kernel_size=3, stride=2, padding=1),  # → [B,64,14,14]
            nn.LeakyReLU(0.2, inplace=True),
            nn.Conv2d(64, 128, kernel_size=3, stride=2, padding=1), # → [B,128,7,7]
            nn.LeakyReLU(0.2, inplace=True),
        )
        
        # 瓶颈层：特征融合
        self.bottleneck = nn.Sequential(
            nn.Conv2d(128, 128, kernel_size=3, padding=1), # → [B,128,7,7]
            nn.LeakyReLU(0.2, inplace=True),
            nn.Conv2d(128, 128, kernel_size=3, padding=1), # → [B,128,7,7]
            nn.LeakyReLU(0.2, inplace=True),
        )
        
        # 解码器：上采样重建高分辨率
        self.decoder = nn.Sequential(
            nn.ConvTranspose2d(128, 64, kernel_size=3, stride=2, padding=1, output_padding=1),
            # → [B,64,14,14]
            nn.LeakyReLU(0.2, inplace=True),
            nn.ConvTranspose2d(64, 32, kernel_size=3, stride=2, padding=1, output_padding=1),
            # → [B,32,28,28]
            nn.LeakyReLU(0.2, inplace=True),
            nn.Conv2d(32, 1, kernel_size=3, padding=1),    # → [B,1,28,28]
            nn.Sigmoid()
        )
    
    def forward(self, x):
        """低分辨率→编码→解码→高分辨率"""
        encoded = self.encoder(x)                # 下采样编码
        bottleneck = self.bottleneck(encoded)     # 特征融合
        decoded = self.decoder(bottleneck)        # 上采样解码
        return decoded

# ======================== 训练 ========================
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = SuperResolutionAE().to(device)

# 结合MSE和感知损失
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

for epoch in range(num_epochs):
    model.train()
    epoch_loss = 0.0
    
    for (lr_data, _), (hr_data, _) in train_loader:
        lr_data = lr_data.to(device)             # 低分辨率输入
        hr_data = hr_data.to(device)             # 高分辨率目标
        
        # 前向传播
        sr_output = model(lr_data)               # 超分辨率重构
        
        # 损失：与高分辨率目标比较
        loss = criterion(sr_output, hr_data)
        
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
    
    if (epoch + 1) % 5 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {epoch_loss/len(train_loader):.6f}')

# ======================== 可视化超分辨率结果 ========================
model.eval()
with torch.no_grad():
    test_lr, test_hr = next(iter(train_loader))
    test_lr = test_lr[:8].to(device)
    test_hr = test_hr[:8].to(device)
    sr_result = model(test_lr)

fig, axes = plt.subplots(3, 8, figsize=(16, 6))
for i in range(8):
    axes[0, i].imshow(test_lr[i].cpu().squeeze(), cmap='gray')    # 低分辨率输入
    axes[0, i].axis('off')
    axes[1, i].imshow(sr_result[i].cpu().squeeze(), cmap='gray')  # 超分辨率输出
    axes[1, i].axis('off')
    axes[2, i].imshow(test_hr[i].cpu().squeeze(), cmap='gray')    # 高分辨率目标
    axes[2, i].axis('off')

axes[0, 0].set_ylabel('Low-Res\nInput', fontsize=10, rotation=0, labelpad=60)
axes[1, 0].set_ylabel('Super-Res\nOutput', fontsize=10, rotation=0, labelpad=60)
axes[2, 0].set_ylabel('High-Res\nTarget', fontsize=10, rotation=0, labelpad=60)

plt.suptitle('Super-Resolution Autoencoder', fontsize=14)
plt.tight_layout()
plt.show()

9.4 案例代码三：特征提取 + 下游分类

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
from sklearn.svm import SVC                    # 支持向量机分类器
from sklearn.metrics import accuracy_score, classification_report
import numpy as np

# ======================== 数据加载 ========================
transform = transforms.Compose([transforms.ToTensor()])

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

# 为了速度只用部分数据
train_subset = torch.utils.data.Subset(train_dataset, range(10000))
test_subset = torch.utils.data.Subset(test_dataset, range(2000))

train_loader = torch.utils.data.DataLoader(
    train_subset, batch_size=128, shuffle=True
)

# ======================== 预训练自编码器 ========================
class FeatureExtractorAE(nn.Module):
    """作为特征提取器的自编码器"""
    def __init__(self):
        super(FeatureExtractorAE, self).__init__()
        
        self.encoder = nn.Sequential(
            nn.Linear(784, 512),
            nn.ReLU(inplace=True),
            nn.Linear(512, 256),
            nn.ReLU(inplace=True),
            nn.Linear(256, 128),
            nn.ReLU(inplace=True),
            nn.Linear(128, 32),                  # 最终特征维度：32
            nn.ReLU(inplace=True)
        )
        self.decoder = nn.Sequential(
            nn.Linear(32, 128),
            nn.ReLU(inplace=True),
            nn.Linear(128, 256),
            nn.ReLU(inplace=True),
            nn.Linear(256, 512),
            nn.ReLU(inplace=True),
            nn.Linear(512, 784),
            nn.Sigmoid()
        )
    
    def forward(self, x):
        z = self.encoder(x)
        return self.decoder(z)

# ---- 阶段1：无监督预训练 ----
print("Phase 1: Unsupervised Pre-training of Autoencoder...")
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
ae = FeatureExtractorAE().to(device)
optimizer = optim.Adam(ae.parameters(), lr=1e-3)
criterion = nn.MSELoss()

for epoch in range(15):
    ae.train()
    total_loss = 0
    for data, _ in train_loader:
        data = data.view(data.size(0), -1).to(device)
        recon = ae(data)
        loss = criterion(recon, data)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    if (epoch + 1) % 5 == 0:
        print(f'  Epoch [{epoch+1}/15], Loss: {total_loss/len(train_loader):.6f}')

# ---- 阶段2：提取特征 ----
print("\nPhase 2: Extracting Features...")
ae.eval()

def extract_features(dataset, model, device):
    """用预训练编码器提取特征"""
    features = []
    labels = []
    loader = torch.utils.data.DataLoader(dataset, batch_size=256, shuffle=False)
    with torch.no_grad():
        for data, label in loader:
            data = data.view(data.size(0), -1).to(device)
            z = model.encoder(data)              # 编码器输出作为特征
            features.append(z.cpu().numpy())
            labels.append(label.numpy())
    return np.concatenate(features), np.concatenate(labels)

train_features, train_labels = extract_features(train_subset, ae, device)
test_features, test_labels = extract_features(test_subset, ae, device)

print(f'  Train features shape: {train_features.shape}')   # [10000, 32]
print(f'  Test features shape: {test_features.shape}')     # [2000, 32]

# ---- 阶段3：用SVM分类 ----
print("\nPhase 3: SVM Classification on AE Features...")
svm = SVC(kernel='rbf', C=10, gamma='scale')   # RBF核SVM
svm.fit(train_features, train_labels)            # 训练
predictions = svm.predict(test_features)         # 预测

accuracy = accuracy_score(test_labels, predictions)
print(f'\nClassification Accuracy: {accuracy:.4f}')
print(f'\nDetailed Report:')
print(classification_report(test_labels, predictions))

# ---- 对比：用原始像素特征训练SVM ----
print("\nComparison: SVM on Raw Pixel Features...")
train_raw = np.array([train_subset[i][0].numpy().flatten() for i in range(len(train_subset))])
test_raw = np.array([test_subset[i][0].numpy().flatten() for i in range(len(test_subset))])

svm_raw = SVC(kernel='rbf', C=10, gamma='scale')
svm_raw.fit(train_raw, train_labels)
predictions_raw = svm_raw.predict(test_raw)

accuracy_raw = accuracy_score(test_labels, predictions_raw)
print(f'Raw Pixels Accuracy: {accuracy_raw:.4f}')
print(f'AE Features Accuracy: {accuracy:.4f}')
print(f'Improvement: {(accuracy - accuracy_raw)*100:.2f}%')

# ======================== t-SNE可视化特征 ========================
from sklearn.manifold import TSNE

# 用t-SNE将32维特征降到2D
tsne = TSNE(n_components=2, random_state=42, perplexity=30)
features_2d = tsne.fit_transform(test_features[:1000])

plt.figure(figsize=(10, 8))
scatter = plt.scatter(
    features_2d[:, 0], features_2d[:, 1],
    c=test_labels[:1000], cmap='tab10', alpha=0.6, s=10
)
plt.colorbar(scatter, ticks=range(10), label='Digit')
plt.xlabel('t-SNE Dimension 1')
plt.ylabel('t-SNE Dimension 2')
plt.title('t-SNE Visualization of AE Learned Features')
plt.grid(True, alpha=0.3)
plt.show()

十、总结对照表

┌─────────────────────┬────────────────────────┬──────────────────────────┐
│       类型          │       核心约束         │        损失函数          │
├─────────────────────┼────────────────────────┼──────────────────────────┤
│ 欠完备自编码器       │ dim(h) < dim(x)       │ L(x, g(f(x)))           │
│ 稀疏自编码器         │ h 稀疏                │ L + λ·KL(ρ ‖ ρ̂)        │
│ 深度自编码器         │ 多层非线性            │ L(x, g(f(x)))           │
│ VAE                 │ h ~ 分布              │ L_recon + KL(q‖p)       │
│ 去噪自编码器         │ 从噪声中恢复          │ L(x, g(f(x̃)))          │
│ 收缩自编码器         │ ‖∂f/∂x‖ 小          │ L + λ·‖J_f‖²_F          │
│ PSD                 │ h 稀疏且可预测        │ ‖x-Dh‖²+λ‖h‖₁+γ‖h-f(x)‖│
└─────────────────────┴────────────────────────┴──────────────────────────┘

亚马逊云科技技术品牌专区

更多推荐

基于STM32的全屋环境智能监测与预警终端设计

亚马逊云科技技术品牌专区

Keras：面向人类的深度学习库

Keras3支持多后端（TensorFlow、PyTorch、JAX），通过设置环境变量切换，兼容指定版本。pip安装后可用Sequential API快速构建CNN模型，如MNIST分类示例，实

亚马逊云科技技术品牌专区

AI行业日报｜2026-09-25：5个热点事件

本期行业动态聚焦于端侧算力重构、自主智能体工作流、开发工具经济模型、网络隔离架构设计与公共部门合规审查。以下结合一线技术场景进行结构化梳理，供 AI 研究员与软件工程团队参考。 PrismML 正将其轻量级语言模型适配至搭载高通处理器的智能眼镜终端 1 。该平台的核心战略指向开源权重模型的本地化运行，旨在最大化挖掘穿戴设备现有算力的利用效率 1 。对于嵌入式与边缘计算开发者而言，此类微型模型向低功