自编码器 Autoencoder：压缩与重建

import torch
import torch.nn as nn

class Autoencoder(nn.Module):
    """基础自编码器：Encoder + Decoder"""
    def __init__(self, input_dim, hidden_dim, latent_dim):
        super().__init__()
        # 编码器：逐步降维到潜空间
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, latent_dim),
            nn.ReLU()
        )
        # 解码器：从潜空间重建
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, input_dim),
            nn.Sigmoid()  # 假设输入归一化到 [0,1]
        )
    
    def forward(self, x):
        z = self.encoder(x)
        x_recon = self.decoder(z)
        return x_recon, z

# 训练循环
def train_ae(model, dataloader, epochs=50, lr=1e-3):
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    criterion = nn.MSELoss()
    
    for epoch in range(epochs):
        total_loss = 0.0
        for batch in dataloader:
            x = batch.view(batch.size(0), -1)
            x_recon, _ = model(x)
            
            loss = criterion(x_recon, x)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        
        avg_loss = total_loss / len(dataloader)
        if (epoch + 1) % 10 == 0:
            print(f"Epoch {epoch+1}/{epochs}, Loss: {avg_loss:.4f}")

# 对比 PCA 与欠完备自编码器
from sklearn.decomposition import PCA
import numpy as np

# PCA 线性降维
pca = PCA(n_components=32)
z_pca = pca.fit_transform(X)  # X: (N, 784)
X_pca_recon = pca.inverse_transform(z_pca)

# 欠完备自编码器等价于非线性 PCA
# 当 encoder/decoder 均为 Linear 且无激活函数时
# AE 的潜空间 = PCA 的主成分子空间
# 差异：AE 可以加入 ReLU/Sigmoid 等非线性

class UndercompleteAE(nn.Module):
    """严格欠完备的自编码器"""
    def __init__(self, input_dim, latent_dim):
        super().__init__()
        assert latent_dim < input_dim, "欠完备要求潜维度 < 输入维度"
        
        layers = []
        dims = [input_dim, 512, 256, 128, latent_dim]
        for i in range(len(dims) - 1):
            layers.extend([
                nn.Linear(dims[i], dims[i+1]),
                nn.BatchNorm1d(dims[i+1]),
                nn.ReLU()
            ])
        layers.pop()  # 最后一层不加激活
        
        self.encoder = nn.Sequential(*layers)
        # 解码器对称结构
        dec_dims = [latent_dim] + dims[:-1][::-1]
        dec_layers = []
        for i in range(len(dec_dims) - 1):
            dec_layers.extend([
                nn.Linear(dec_dims[i], dec_dims[i+1]),
                nn.BatchNorm1d(dec_dims[i+1]),
                nn.ReLU()
            ])
        dec_layers[-1] = nn.Linear(dec_dims[-2], input_dim)
        self.decoder = nn.Sequential(*dec_layers)
    
    def forward(self, x):
        return self.decoder(self.encoder(x)), self.encoder(x)

class SparseAutoencoder(nn.Module):
    """稀疏自编码器：加入 L1 稀疏性约束"""
    def __init__(self, input_dim, hidden_dim, latent_dim, sparsity=0.05, beta=3.0):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, latent_dim), nn.Sigmoid()
        )
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, input_dim), nn.Sigmoid()
        )
        self.sparsity = sparsity  # 目标平均激活率
        self.beta = beta  # 稀疏惩罚权重
    
    def sparse_penalty(self, z):
        """KL 散度稀疏惩罚"""
        p = self.sparsity
        p_hat = z.mean(dim=0)  # 每个神经元的平均激活
        p_hat = torch.clamp(p_hat, 1e-8, 1 - 1e-8)
        kl = p * torch.log(p / p_hat) + (1 - p) * torch.log((1 - p) / (1 - p_hat))
        return kl.sum()
    
    def forward(self, x):
        z = self.encoder(x)
        x_recon = self.decoder(z)
        recon_loss = nn.MSELoss()(x_recon, x)
        loss = recon_loss + self.beta * self.sparse_penalty(z)
        return x_recon, z, loss

class DenoisingAutoencoder(nn.Module):
    """去噪自编码器：输入加噪声后重建"""
    def __init__(self, input_dim, hidden_dim, latent_dim, noise_level=0.3):
        super().__init__()
        self.noise_level = noise_level
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, latent_dim), nn.ReLU()
        )
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, input_dim), nn.Sigmoid()
        )
    
    def add_noise(self, x):
        """添加高斯噪声或随机掩码"""
        noise = torch.randn_like(x) * self.noise_level
        return torch.clamp(x + noise, 0, 1)
    
    def forward(self, x):
        x_noisy = self.add_noise(x)
        z = self.encoder(x_noisy)
        x_recon = self.decoder(z)
        loss = nn.MSELoss()(x_recon, x)  # 对比干净输入
        return x_recon, z, loss

class VAE(nn.Module):
    """变分自编码器：编码为分布，支持生成"""
    def __init__(self, input_dim, hidden_dim, latent_dim):
        super().__init__()
        # 编码器输出 mu 和 log_var
        self.fc_enc = nn.Sequential(
            nn.Linear(input_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim), nn.ReLU()
        )
        self.fc_mu = nn.Linear(hidden_dim, latent_dim)
        self.fc_logvar = nn.Linear(hidden_dim, latent_dim)
        # 解码器
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, input_dim), nn.Sigmoid()
        )
    
    def reparameterize(self, mu, logvar):
        """重参数化技巧：使采样可导"""
        std = torch.exp(0.5 * logvar)
        eps = torch.randn_like(std)
        return mu + eps * std
    
    def forward(self, x):
        h = self.fc_enc(x)
        mu = self.fc_mu(h)
        logvar = self.fc_logvar(h)
        z = self.reparameterize(mu, logvar)
        return self.decoder(z), mu, logvar

def vae_loss(x_recon, x, mu, logvar, beta=1.0):
    """VAE 损失 = 重建损失 + KL 散度"""
    recon_loss = nn.MSELoss()(x_recon, x)
    # KL(q(z|x) || p(z))，闭式解
    kl_loss = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
    kl_loss /= x.size(0) * x.size(1)  # 归一化
    return recon_loss + beta * kl_loss, recon_loss, kl_loss

# 采样新数据
def sample_vae(model, n_samples, latent_dim, device):
    model.eval()
    with torch.no_grad():
        z = torch.randn(n_samples, latent_dim, device=device)
        samples = model.decoder(z)
        return samples  # 生成的新样本

class Discriminator(nn.Module):
    """判别潜向量分布的判别器"""
    def __init__(self, latent_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(latent_dim, 128), nn.LeakyReLU(0.2),
            nn.Linear(128, 64), nn.LeakyReLU(0.2),
            nn.Linear(64, 1), nn.Sigmoid()
        )
    
    def forward(self, z):
        return self.net(z)


class AAE(nn.Module):
    def __init__(self, input_dim, hidden_dim, latent_dim):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, latent_dim)
        )
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, input_dim), nn.Sigmoid()
        )
        self.discriminator = Discriminator(latent_dim)

# AAE 训练：交替优化
def train_aae_step(aae, x, optimizer_ae, optimizer_d, device):
    batch_size = x.size(0)
    
    # --- 阶段 1: 重建阶段 ---
    optimizer_ae.zero_grad()
    z = aae.encoder(x)
    x_recon = aae.decoder(z)
    recon_loss = nn.MSELoss()(x_recon, x)
    recon_loss.backward(retain_graph=True)
    
    # --- 阶段 2: 判别器阶段 ---
    optimizer_d.zero_grad()
    z_fake = aae.encoder(x).detach()
    z_real = torch.randn_like(z_fake)
    d_fake = aae.discriminator(z_fake)
    d_real = aae.discriminator(z_real)
    d_loss = -(torch.log(d_real + 1e-8).mean() + torch.log(1 - d_fake + 1e-8).mean())
    d_loss.backward()
    optimizer_d.step()
    
    # --- 阶段 3: 对抗阶段（更新编码器） ---
    optimizer_ae.zero_grad()
    z_fake = aae.encoder(x)
    d_fake = aae.discriminator(z_fake)
    g_loss = -torch.log(d_fake + 1e-8).mean()
    g_loss.backward()
    optimizer_ae.step()
    
    return recon_loss.item(), d_loss.item(), g_loss.item()

class AnomalyDetector:
    """基于自编码器的异常检测器"""
    def __init__(self, autoencoder, threshold_percentile=95):
        self.ae = autoencoder
        self.threshold_percentile = threshold_percentile
        self.threshold = None
    
    def fit_threshold(self, dataloader):
        """在正常数据上确定异常阈值"""
        self.ae.eval()
        errors = []
        with torch.no_grad():
            for batch in dataloader:
                x = batch.view(batch.size(0), -1)
                x_recon, _ = self.ae(x)
                error = torch.mean((x - x_recon) ** 2, dim=1)
                errors.extend(error.cpu().numpy())
        # 使用百分位数作为阈值
        self.threshold = np.percentile(errors, self.threshold_percentile)
    
    def predict(self, x):
        """判断是否为异常"""
        self.ae.eval()
        with torch.no_grad():
            x_recon, _ = self.ae(x)
            error = torch.mean((x - x_recon) ** 2, dim=1)
            return (error > self.threshold).int()

# 潜空间属性编辑：图像生成示例
def latent_interpolation(model, z1, z2, n_steps=10):
    """在两个潜向量之间线性插值"""
    alphas = torch.linspace(0, 1, n_steps).view(-1, 1)
    z_interp = (1 - alphas) * z1 + alphas * z2
    model.eval()
    with torch.no_grad():
        images = model.decoder(z_interp)
        return images  # 平滑过渡的图像序列


def latent_arithmetic(model, z_a, z_b, z_c, alpha=1.0):
    """潜空间向量运算：z = z_a - z_b + z_c"""
    z_result = z_a - z_b + z_c
    model.eval()
    with torch.no_grad():
        return model.decoder(z_result)

class ConvVAE(nn.Module):
    """卷积变分自编码器 for MNIST"""
    def __init__(self, latent_dim=20):
        super().__init__()
        # 编码器：28x28 -> 4x4
        self.enc = nn.Sequential(
            nn.Conv2d(1, 32, 4, stride=2, padding=1),   # 28->14
            nn.BatchNorm2d(32), nn.ReLU(),
            nn.Conv2d(32, 64, 4, stride=2, padding=1),   # 14->7
            nn.BatchNorm2d(64), nn.ReLU(),
            nn.Conv2d(64, 128, 4, stride=2, padding=1),  # 7->4
            nn.BatchNorm2d(128), nn.ReLU(),
        )
        # 全连接层
        self.fc_mu = nn.Linear(128 * 4 * 4, latent_dim)
        self.fc_logvar = nn.Linear(128 * 4 * 4, latent_dim)
        self.fc_decode = nn.Linear(latent_dim, 128 * 4 * 4)
        # 解码器：4x4 -> 28x28
        self.dec = nn.Sequential(
            nn.ConvTranspose2d(128, 64, 4, stride=2, padding=1),
            nn.BatchNorm2d(64), nn.ReLU(),
            nn.ConvTranspose2d(64, 32, 4, stride=2, padding=1),
            nn.BatchNorm2d(32), nn.ReLU(),
            nn.ConvTranspose2d(32, 1, 4, stride=2, padding=1),
            nn.Sigmoid()
        )
    
    def reparameterize(self, mu, logvar):
        return mu + torch.exp(0.5 * logvar) * torch.randn_like(mu)
    
    def forward(self, x):
        h = self.enc(x)
        h = h.view(h.size(0), -1)
        mu, logvar = self.fc_mu(h), self.fc_logvar(h)
        z = self.reparameterize(mu, logvar)
        h_dec = self.fc_decode(z).view(-1, 128, 4, 4)
        return self.dec(h_dec), mu, logvar

# 完整训练与可视化
import torchvision.transforms as transforms
from torchvision.datasets import MNIST
from torch.utils.data import DataLoader

transform = transforms.Compose([transforms.ToTensor()])
train_ds = MNIST(root='./data', train=True, download=True, transform=transform)
train_loader = DataLoader(train_ds, batch_size=128, shuffle=True)

model = ConvVAE(latent_dim=20)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

for epoch in range(30):
    model.train()
    total_loss = 0
    for x, _ in train_loader:
        x_recon, mu, logvar = model(x)
        recon = nn.BCELoss()(x_recon, x)
        kl = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp()) / x.size(0)
        loss = recon + kl
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"Epoch {epoch+1}: Loss={total_loss/len(train_loader):.4f}")

# 潜空间插值可视化
model.eval()
with torch.no_grad():
    z1 = torch.randn(1, 20)  # 数字 A 的潜向量
    z2 = torch.randn(1, 20)  # 数字 B 的潜向量
    alphas = torch.linspace(0, 1, 8).view(-1, 1)
    z_path = (1 - alphas) * z1 + alphas * z2
    interp_images = model.decoder(model.fc_decode(z_path).view(-1, 128, 4, 4))
    # interp_images: (8, 1, 28, 28) 从 A 平滑过渡到 B