在深度学习项目实践中,很多开发者会遇到模型选择困难、原理理解不透彻、代码实现不完整等问题。本文系统梳理CNN、RNN、Transformer、GAN、扩散模型等十大核心算法的原理与实战,包含完整可运行的代码示例,适合从入门到项目落地的全阶段学习。

1. 深度学习核心算法概述

深度学习作为机器学习的重要分支,通过多层神经网络模拟人脑处理信息的机制。近年来,各种深度学习算法在计算机视觉、自然语言处理、语音识别等领域取得了突破性进展。掌握这些核心算法不仅有助于理解现代人工智能技术的底层原理,更能为实际项目开发提供坚实的技术基础。

1.1 算法发展历程与关系

深度学习算法的发展呈现出明显的演进路径。从早期的卷积神经网络(CNN)解决图像识别问题,到循环神经网络(RNN)处理序列数据,再到Transformer模型在自然语言处理领域的革命性突破,每种算法都在特定领域发挥着重要作用。生成对抗网络(GAN)和扩散模型则在生成式AI领域展现出强大能力。

这些算法并非相互替代,而是各有侧重。CNN擅长处理网格状数据(如图像),RNN适合时序数据,Transformer在长序列建模上表现优异,GAN和扩散模型在数据生成方面各具特色。理解它们之间的关系有助于在实际项目中做出正确的技术选型。

1.2 核心算法应用场景

在实际项目中,不同算法对应不同的应用场景。CNN广泛应用于图像分类、目标检测、人脸识别等计算机视觉任务;RNN及其变体LSTM、GRU常用于语音识别、时间序列预测、文本生成;Transformer不仅是BERT、GPT等大语言模型的基础,也在视觉任务中表现出色;GAN可用于图像生成、风格迁移、数据增强;扩散模型在图像生成、音频合成等领域效果显著。

2. 环境准备与工具配置

2.1 基础环境要求

深度学习项目开发需要准备合适的软硬件环境。推荐使用Python 3.8+作为编程语言,PyTorch 2.0+或TensorFlow 2.12+作为深度学习框架。硬件方面,建议配备NVIDIA GPU(至少8GB显存)以加速模型训练,虽然CPU也能运行但训练速度较慢。

操作系统可以选择Windows、Linux或macOS,但Linux在深度学习社区支持度最好。必备的Python库包括NumPy用于数值计算,Matplotlib用于可视化,Pandas用于数据处理,以及框架特定的扩展库。

2.2 开发环境配置

以下是最小化的环境配置示例,使用conda创建虚拟环境:

# 创建并激活虚拟环境
conda create -n dl-tutorial python=3.9
conda activate dl-tutorial

# 安装核心依赖
pip install torch torchvision torchaudio
pip install tensorflow
pip install numpy pandas matplotlib jupyter

对于IDE选择,Jupyter Notebook适合实验和调试,PyCharm或VS Code适合大型项目开发。建议配置GPU加速,安装对应版本的CUDA和cuDNN:

# 检查CUDA是否可用
python -c "import torch; print(torch.cuda.is_available())"

3. 卷积神经网络(CNN)原理与实战

3.1 CNN核心原理详解

卷积神经网络通过卷积核在输入数据上的滑动窗口操作提取特征。其核心组件包括卷积层、池化层和全连接层。卷积层负责特征提取,通过局部连接和权重共享大幅减少参数数量;池化层(最大池化、平均池化)实现特征降维和平移不变性;全连接层完成最终分类任务。

CNN的关键优势在于其层次化特征学习能力。浅层卷积核学习边缘、颜色等低级特征,中层组合成纹理、部件等中级特征,深层形成物体、场景等高级特征。这种层次结构使其特别适合处理图像数据。

3.2 CNN图像分类实战

下面实现一个完整的CNN图像分类模型,使用CIFAR-10数据集:

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

# 定义CNN模型
class SimpleCNN(nn.Module):
    def __init__(self, num_classes=10):
        super(SimpleCNN, self).__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 32, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=2, stride=2),
            
            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=2, stride=2),
            
            nn.Conv2d(64, 128, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=2, stride=2),
        )
        self.classifier = nn.Sequential(
            nn.Dropout(0.5),
            nn.Linear(128 * 4 * 4, 512),
            nn.ReLU(inplace=True),
            nn.Dropout(0.5),
            nn.Linear(512, num_classes)
        )
    
    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), -1)
        x = self.classifier(x)
        return x

# 数据预处理
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
])

# 加载数据集
trainset = torchvision.datasets.CIFAR10(root='./data', train=True,
                                        download=True, transform=transform)
trainloader = DataLoader(trainset, batch_size=128, shuffle=True)

testset = torchvision.datasets.CIFAR10(root='./data', train=False,
                                       download=True, transform=transform)
testloader = DataLoader(testset, batch_size=128, shuffle=False)

# 训练配置
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = SimpleCNN().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 训练循环
for epoch in range(10):
    running_loss = 0.0
    for i, data in enumerate(trainloader, 0):
        inputs, labels = data
        inputs, labels = inputs.to(device), labels.to(device)
        
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        
        running_loss += loss.item()
        if i % 100 == 99:
            print(f'Epoch {epoch+1}, Batch {i+1}: Loss {running_loss/100:.3f}')
            running_loss = 0.0

print('训练完成')

这个示例展示了完整的CNN训练流程,包括模型定义、数据加载、训练循环等关键环节。在实际项目中,可以根据具体任务调整网络结构和超参数。

4. 循环神经网络(RNN)与序列建模

4.1 RNN基本原理与变体

循环神经网络专为处理序列数据设计,通过循环连接保持历史信息。传统RNN存在梯度消失/爆炸问题,因此发展出LSTM(长短期记忆网络)和GRU(门控循环单元)等变体。

LSTM通过输入门、遗忘门、输出门三个门控机制控制信息流动,能够学习长期依赖关系。GRU是LSTM的简化版本,将遗忘门和输入门合并为更新门,参数更少但效果相当。选择哪种变体取决于具体任务和计算资源。

4.2 RNN文本分类实战

下面实现基于LSTM的文本情感分类模型:

import torch
import torch.nn as nn
import torch.optim as optim
from torchtext.legacy import data, datasets
import spacy

# 定义LSTM模型
class TextLSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, n_layers, dropout):
        super(TextLSTM, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, n_layers, 
                           dropout=dropout, batch_first=True)
        self.fc = nn.Linear(hidden_dim, output_dim)
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, text):
        embedded = self.dropout(self.embedding(text))
        output, (hidden, cell) = self.lstm(embedded)
        hidden = self.dropout(hidden[-1,:,:])
        return self.fc(hidden)

# 数据预处理
TEXT = data.Field(tokenize='spacy', include_lengths=True)
LABEL = data.LabelField(dtype=torch.float)

# 加载IMDB电影评论数据集
train_data, test_data = datasets.IMDB.splits(TEXT, LABEL)

# 构建词汇表
MAX_VOCAB_SIZE = 25000
TEXT.build_vocab(train_data, max_size=MAX_VOCAB_SIZE)
LABEL.build_vocab(train_data)

# 创建数据迭代器
BATCH_SIZE = 64
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
train_iterator, test_iterator = data.BucketIterator.splits(
    (train_data, test_data), 
    batch_size=BATCH_SIZE,
    device=device,
    sort_within_batch=True,
    sort_key=lambda x: len(x.text)
)

# 模型初始化
INPUT_DIM = len(TEXT.vocab)
EMBEDDING_DIM = 100
HIDDEN_DIM = 256
OUTPUT_DIM = 1
N_LAYERS = 2
DROPOUT = 0.5

model = TextLSTM(INPUT_DIM, EMBEDDING_DIM, HIDDEN_DIM, OUTPUT_DIM, N_LAYERS, DROPOUT)
model = model.to(device)

# 训练配置
optimizer = optim.Adam(model.parameters())
criterion = nn.BCEWithLogitsLoss()

def binary_accuracy(preds, y):
    rounded_preds = torch.round(torch.sigmoid(preds))
    correct = (rounded_preds == y).float()
    acc = correct.sum() / len(correct)
    return acc

# 训练函数
def train(model, iterator, optimizer, criterion):
    epoch_loss = 0
    epoch_acc = 0
    model.train()
    
    for batch in iterator:
        text, text_lengths = batch.text
        predictions = model(text).squeeze(1)
        loss = criterion(predictions, batch.label)
        acc = binary_accuracy(predictions, batch.label)
        
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        epoch_loss += loss.item()
        epoch_acc += acc.item()
    
    return epoch_loss / len(iterator), epoch_acc / len(iterator)

# 开始训练
N_EPOCHS = 5
for epoch in range(N_EPOCHS):
    train_loss, train_acc = train(model, train_iterator, optimizer, criterion)
    print(f'Epoch: {epoch+1:02}')
    print(f'\tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc*100:.2f}%')

这个示例展示了RNN在文本分类任务中的应用,包括数据预处理、模型定义、训练评估等完整流程。

5. Transformer模型架构详解

5.1 自注意力机制原理

Transformer的核心创新是自注意力机制,它允许模型在处理每个位置时关注输入序列的所有位置。自注意力通过查询(Query)、键(Key)、值(Value)三个矩阵计算注意力权重,公式为:Attention(Q,K,V)=softmax(QK^T/√d_k)V。

多头注意力将自注意力机制并行执行多次,使模型能够同时关注不同表示子空间的信息。位置编码则弥补了Transformer缺乏位置信息的缺陷,通常使用正弦余弦函数生成。

5.2 Transformer文本翻译实战

下面实现一个简化的Transformer机器翻译模型:

import torch
import torch.nn as nn
import torch.optim as optim
import math
import time

# 位置编码
class PositionalEncoding(nn.Module):
    def __init__(self, d_model, dropout=0.1, max_len=5000):
        super(PositionalEncoding, self).__init__()
        self.dropout = nn.Dropout(p=dropout)
        
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * 
                           (-math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0).transpose(0, 1)
        self.register_buffer('pe', pe)
    
    def forward(self, x):
        x = x + self.pe[:x.size(0), :]
        return self.dropout(x)

# Transformer模型
class TransformerModel(nn.Module):
    def __init__(self, src_vocab, tgt_vocab, d_model=512, nhead=8, 
                 num_encoder_layers=6, num_decoder_layers=6, 
                 dim_feedforward=2048, dropout=0.1):
        super(TransformerModel, self).__init__()
        self.d_model = d_model
        self.embed_src = nn.Embedding(src_vocab, d_model)
        self.embed_tgt = nn.Embedding(tgt_vocab, d_model)
        self.pos_encoder = PositionalEncoding(d_model, dropout)
        self.pos_decoder = PositionalEncoding(d_model, dropout)
        
        self.transformer = nn.Transformer(d_model=d_model, nhead=nhead,
                                         num_encoder_layers=num_encoder_layers,
                                         num_decoder_layers=num_decoder_layers,
                                         dim_feedforward=dim_feedforward,
                                         dropout=dropout)
        self.fc_out = nn.Linear(d_model, tgt_vocab)
        
    def forward(self, src, tgt, src_mask=None, tgt_mask=None, 
                memory_mask=None, src_key_padding_mask=None,
                tgt_key_padding_mask=None, memory_key_padding_mask=None):
        src = self.embed_src(src) * math.sqrt(self.d_model)
        tgt = self.embed_tgt(tgt) * math.sqrt(self.d_model)
        
        src = self.pos_encoder(src)
        tgt = self.pos_decoder(tgt)
        
        output = self.transformer(src, tgt, src_mask, tgt_mask, memory_mask,
                                 src_key_padding_mask, tgt_key_padding_mask,
                                 memory_key_padding_mask)
        return self.fc_out(output)

# 训练配置
def create_mask(src, tgt, pad_idx):
    src_seq_len = src.shape[0]
    tgt_seq_len = tgt.shape[0]
    
    tgt_mask = nn.Transformer.generate_square_subsequent_mask(tgt_seq_len)
    src_mask = torch.zeros((src_seq_len, src_seq_len)).type(torch.bool)
    
    src_padding_mask = (src == pad_idx).transpose(0, 1)
    tgt_padding_mask = (tgt == pad_idx).transpose(0, 1)
    
    return src_mask, tgt_mask, src_padding_mask, tgt_padding_mask

# 示例训练循环
def train_epoch(model, optimizer, criterion, train_iter, pad_idx):
    model.train()
    losses = 0
    
    for src, tgt in train_iter:
        src = src.transpose(0, 1)  # (batch, seq) -> (seq, batch)
        tgt = tgt.transpose(0, 1)
        
        tgt_input = tgt[:-1, :]  # 解码器输入
        tgt_output = tgt[1:, :]   # 解码器目标输出
        
        src_mask, tgt_mask, src_padding_mask, tgt_padding_mask = create_mask(
            src, tgt_input, pad_idx)
        
        optimizer.zero_grad()
        output = model(src, tgt_input, src_mask, tgt_mask, None,
                      src_padding_mask, tgt_padding_mask, src_padding_mask)
        
        loss = criterion(output.reshape(-1, output.shape[-1]), 
                        tgt_output.reshape(-1))
        loss.backward()
        optimizer.step()
        
        losses += loss.item()
    
    return losses / len(train_iter)

# 模型初始化
SRC_VOCAB_SIZE = 10000
TGT_VOCAB_SIZE = 10000
model = TransformerModel(SRC_VOCAB_SIZE, TGT_VOCAB_SIZE)
optimizer = optim.Adam(model.parameters(), lr=0.0001, betas=(0.9, 0.98), eps=1e-9)
criterion = nn.CrossEntropyLoss(ignore_index=0)  # 忽略padding

print("Transformer模型定义完成")

这个Transformer实现包含了编码器、解码器、位置编码等核心组件,展示了机器翻译任务的基本框架。

6. 生成对抗网络(GAN)原理与应用

6.1 GAN基本架构与训练过程

生成对抗网络包含生成器(Generator)和判别器(Discriminator)两个神经网络。生成器负责从随机噪声生成假数据,判别器负责区分真实数据和生成数据。两者通过对抗训练共同进步,最终生成器能够产生以假乱真的数据。

GAN的训练过程是一个极小极大博弈:min_G max_D V(D,G) = E_{x~p_data}[log D(x)] + E_{z~p_z}[log(1-D(G(z)))]。在实际训练中,通常交替训练判别器和生成器,使用梯度下降方法优化目标函数。

6.2 GAN图像生成实战

下面实现一个DCGAN(深度卷积生成对抗网络)用于手写数字生成:

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
from torch.utils.data import DataLoader
import matplotlib.pyplot as plt
import numpy as np

# 生成器网络
class Generator(nn.Module):
    def __init__(self, latent_dim=100, img_channels=1, feature_size=64):
        super(Generator, self).__init__()
        self.main = nn.Sequential(
            # 输入: latent_dim x 1 x 1
            nn.ConvTranspose2d(latent_dim, feature_size * 8, 4, 1, 0, bias=False),
            nn.BatchNorm2d(feature_size * 8),
            nn.ReLU(True),
            # 输出: (feature_size*8) x 4 x 4
            
            nn.ConvTranspose2d(feature_size * 8, feature_size * 4, 4, 2, 1, bias=False),
            nn.BatchNorm2d(feature_size * 4),
            nn.ReLU(True),
            # 输出: (feature_size*4) x 8 x 8
            
            nn.ConvTranspose2d(feature_size * 4, feature_size * 2, 4, 2, 1, bias=False),
            nn.BatchNorm2d(feature_size * 2),
            nn.ReLU(True),
            # 输出: (feature_size*2) x 16 x 16
            
            nn.ConvTranspose2d(feature_size * 2, feature_size, 4, 2, 1, bias=False),
            nn.BatchNorm2d(feature_size),
            nn.ReLU(True),
            # 输出: (feature_size) x 32 x 32
            
            nn.ConvTranspose2d(feature_size, img_channels, 4, 2, 1, bias=False),
            nn.Tanh()
            # 输出: img_channels x 64 x 64
        )
    
    def forward(self, input):
        return self.main(input)

# 判别器网络
class Discriminator(nn.Module):
    def __init__(self, img_channels=1, feature_size=64):
        super(Discriminator, self).__init__()
        self.main = nn.Sequential(
            # 输入: img_channels x 64 x 64
            nn.Conv2d(img_channels, feature_size, 4, 2, 1, bias=False),
            nn.LeakyReLU(0.2, inplace=True),
            # 输出: feature_size x 32 x 32
            
            nn.Conv2d(feature_size, feature_size * 2, 4, 2, 1, bias=False),
            nn.BatchNorm2d(feature_size * 2),
            nn.LeakyReLU(0.2, inplace=True),
            # 输出: (feature_size*2) x 16 x 16
            
            nn.Conv2d(feature_size * 2, feature_size * 4, 4, 2, 1, bias=False),
            nn.BatchNorm2d(feature_size * 4),
            nn.LeakyReLU(0.2, inplace=True),
            # 输出: (feature_size*4) x 8 x 8
            
            nn.Conv2d(feature_size * 4, feature_size * 8, 4, 2, 1, bias=False),
            nn.BatchNorm2d(feature_size * 8),
            nn.LeakyReLU(0.2, inplace=True),
            # 输出: (feature_size*8) x 4 x 4
            
            nn.Conv2d(feature_size * 8, 1, 4, 1, 0, bias=False),
            nn.Sigmoid()
            # 输出: 1 x 1 x 1
        )
    
    def forward(self, input):
        return self.main(input).view(-1, 1).squeeze(1)

# 训练配置
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
latent_dim = 100
lr = 0.0002
beta1 = 0.5

# 初始化网络
generator = Generator(latent_dim).to(device)
discriminator = Discriminator().to(device)

# 损失函数和优化器
criterion = nn.BCELoss()
optimizerG = optim.Adam(generator.parameters(), lr=lr, betas=(beta1, 0.999))
optimizerD = optim.Adam(discriminator.parameters(), lr=lr, betas=(beta1, 0.999))

# 数据加载
transform = transforms.Compose([
    transforms.Resize(64),
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,))
])

dataset = torchvision.datasets.MNIST(root='./data', train=True,
                                    download=True, transform=transform)
dataloader = DataLoader(dataset, batch_size=128, shuffle=True)

# 训练循环
num_epochs = 50
for epoch in range(num_epochs):
    for i, (real_imgs, _) in enumerate(dataloader):
        batch_size = real_imgs.size(0)
        real_imgs = real_imgs.to(device)
        
        # 训练判别器
        optimizerD.zero_grad()
        
        # 真实图像损失
        real_labels = torch.ones(batch_size).to(device)
        output = discriminator(real_imgs)
        loss_real = criterion(output, real_labels)
        
        # 生成图像损失
        z = torch.randn(batch_size, latent_dim, 1, 1).to(device)
        fake_imgs = generator(z)
        fake_labels = torch.zeros(batch_size).to(device)
        output = discriminator(fake_imgs.detach())
        loss_fake = criterion(output, fake_labels)
        
        # 总损失
        loss_d = loss_real + loss_fake
        loss_d.backward()
        optimizerD.step()
        
        # 训练生成器
        optimizerG.zero_grad()
        
        output = discriminator(fake_imgs)
        loss_g = criterion(output, real_labels)  # 骗过判别器
        loss_g.backward()
        optimizerG.step()
        
        if i % 100 == 0:
            print(f'Epoch [{epoch}/{num_epochs}], Batch [{i}/{len(dataloader)}], '
                  f'Loss D: {loss_d.item():.4f}, Loss G: {loss_g.item():.4f}')

print('GAN训练完成')

# 生成示例图像
with torch.no_grad():
    z = torch.randn(16, latent_dim, 1, 1).to(device)
    generated = generator(z)
    generated = generated.cpu()
    
    fig, axes = plt.subplots(4, 4, figsize=(8, 8))
    for i, ax in enumerate(axes.flat):
        ax.imshow(generated[i].squeeze(), cmap='gray')
        ax.axis('off')
    plt.show()

这个DCGAN实现展示了生成对抗网络的完整训练流程,包括生成器、判别器的定义,对抗训练过程,以及结果可视化。

7. 扩散模型原理与图像生成

7.1 扩散过程与去噪原理

扩散模型包含前向扩散和反向去噪两个过程。前向过程逐步向数据添加高斯噪声,最终将数据转化为纯噪声;反向过程学习从噪声中重建原始数据。DDPM(去噪扩散概率模型)通过U-Net架构预测添加的噪声,实现高质量图像生成。

扩散模型的优势在于训练稳定性优于GAN,生成质量高且多样性好。其数学基础是马尔可夫链和变分推断,通过优化变分下界训练模型。近年来出现的改进版本如DDIM(去噪扩散隐式模型)加速了采样过程,Stable Diffusion等模型在文本到图像生成领域取得突破。

7.2 扩散模型实战实现

下面实现一个简化的扩散模型用于MNIST数字生成:

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision import datasets, transforms
from torch.utils.data import DataLoader
import matplotlib.pyplot as plt
import numpy as np

# 扩散模型调度器
class Diffusion:
    def __init__(self, T=1000, beta_start=1e-4, beta_end=0.02):
        self.T = T
        self.beta = torch.linspace(beta_start, beta_end, T)
        self.alpha = 1 - self.beta
        self.alpha_bar = torch.cumprod(self.alpha, dim=0)
    
    def forward_process(self, x0, t):
        """前向扩散过程"""
        sqrt_alpha_bar = torch.sqrt(self.alpha_bar[t])
        sqrt_one_minus_alpha_bar = torch.sqrt(1 - self.alpha_bar[t])
        
        noise = torch.randn_like(x0)
        xt = sqrt_alpha_bar[:, None, None, None] * x0 + \
             sqrt_one_minus_alpha_bar[:, None, None, None] * noise
        return xt, noise
    
    def reverse_process(self, model, x, t):
        """反向去噪过程"""
        pred_noise = model(x, t)
        alpha_t = self.alpha[t][:, None, None, None]
        alpha_bar_t = self.alpha_bar[t][:, None, None, None]
        beta_t = self.beta[t][:, None, None, None]
        
        if t[0] > 0:
            z = torch.randn_like(x)
        else:
            z = 0
        
        x_prev = 1 / torch.sqrt(alpha_t) * (
            x - (1 - alpha_t) / torch.sqrt(1 - alpha_bar_t) * pred_noise
        ) + torch.sqrt(beta_t) * z
        return x_prev

# U-Net模型
class UNet(nn.Module):
    def __init__(self, in_channels=1, out_channels=1, T=1000):
        super(UNet, self).__init__()
        self.T = T
        
        # 时间步嵌入
        self.time_embed = nn.Sequential(
            nn.Linear(1, 128),
            nn.SiLU(),
            nn.Linear(128, 256)
        )
        
        # 编码器
        self.enc1 = self._block(in_channels, 64)
        self.enc2 = self._block(64, 128)
        self.enc3 = self._block(128, 256)
        self.enc4 = self._block(256, 512)
        
        # 解码器
        self.dec1 = self._block(512 + 256, 256)
        self.dec2 = self._block(256 + 128, 128)
        self.dec3 = self._block(128 + 64, 64)
        self.dec4 = nn.Conv2d(64, out_channels, 3, padding=1)
        
        self.pool = nn.MaxPool2d(2)
        self.upsample = nn.Upsample(scale_factor=2, mode='bilinear', align_corners=True)
    
    def _block(self, in_channels, out_channels):
        return nn.Sequential(
            nn.Conv2d(in_channels, out_channels, 3, padding=1),
            nn.BatchNorm2d(out_channels),
            nn.SiLU(),
            nn.Conv2d(out_channels, out_channels, 3, padding=1),
            nn.BatchNorm2d(out_channels),
            nn.SiLU()
        )
    
    def forward(self, x, t):
        # 时间嵌入
        t = t.float().view(-1, 1) / self.T
        t_embed = self.time_embed(t).unsqueeze(-1).unsqueeze(-1)
        
        # 编码器路径
        s1 = self.enc1(x)
        x = self.pool(s1)
        
        s2 = self.enc2(x)
        x = self.pool(s2)
        
        s3 = self.enc3(x)
        x = self.pool(s3)
        
        x = self.enc4(x)
        
        # 解码器路径
        x = self.upsample(x)
        x = torch.cat([x, s3], dim=1)
        x = self.dec1(x)
        
        x = self.upsample(x)
        x = torch.cat([x, s2], dim=1)
        x = self.dec2(x)
        
        x = self.upsample(x)
        x = torch.cat([x, s1], dim=1)
        x = self.dec3(x)
        
        x = self.dec4(x)
        return x

# 训练配置
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
T = 1000
batch_size = 64
lr = 1e-4
epochs = 20

# 初始化模型和扩散过程
model = UNet(T=T).to(device)
diffusion = Diffusion(T=T)
optimizer = optim.Adam(model.parameters(), lr=lr)

# 数据加载
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,))
])
dataset = datasets.MNIST('./data', train=True, download=True, transform=transform)
dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

# 训练循环
for epoch in range(epochs):
    total_loss = 0
    for batch_idx, (data, _) in enumerate(dataloader):
        data = data.to(device)
        batch_size = data.size(0)
        
        # 随机选择时间步
        t = torch.randint(0, T, (batch_size,)).to(device)
        
        # 前向扩散过程
        noisy_data, noise = diffusion.forward_process(data, t)
        
        # 预测噪声
        pred_noise = model(noisy_data, t)
        
        # 计算损失
        loss = F.mse_loss(pred_noise, noise)
        
        # 反向传播
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
        
        if batch_idx % 100 == 0:
            print(f'Epoch: {epoch} [{batch_idx * len(data)}/{len(dataloader.dataset)}] '
                  f'Loss: {loss.item():.6f}')
    
    avg_loss = total_loss / len(dataloader)
    print(f'Epoch {epoch} Average Loss: {avg_loss:.6f}')

# 生成图像
def generate_images(model, diffusion, num_images=16):
    model.eval()
    with torch.no_grad():
        # 从纯噪声开始
        x = torch.randn(num_images, 1, 28, 28).to(device)
        
        for t in reversed(range(diffusion.T)):
            t_batch = torch.full((num_images,), t, device=device)
            x = diffusion.reverse_process(model, x, t_batch)
        
        # 反标准化
        x = (x.clamp(-1, 1) + 1) / 2
        return x.cpu()

# 生成并显示图像
generated = generate_images(model, diffusion)
fig, axes = plt.subplots(4, 4, figsize=(8, 8))
for i, ax in enumerate(axes.flat):
    ax.imshow(generated[i].squeeze(), cmap='gray')
    ax.axis('off')
plt.show()

这个扩散模型实现展示了完整的前向扩散和反向去噪过程,包括时间步嵌入、U-Net架构设计等关键技术点。

8. 注意力机制深度解析

8.1 注意力机制数学原理

注意力机制的核心思想是根据输入的不同部分对输出的贡献程度分配不同的权重。其数学表达式为:Attention(Q,K,V)=softmax(QK^T/√d_k)V,其中Q是查询矩阵,K是键矩阵,V是值矩阵,d_k是键向量的维度。

缩放点积注意力通过除以√d_k防止点积过大导致softmax梯度消失。多头注意力将输入投影到不同的子空间,分别计算注意力后拼接结果,使模型能够关注不同方面的信息。

8.2 自注意力机制实现

下面实现一个完整的自注意力机制模块:

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

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads, dropout=0.1):
        super(MultiHeadAttention, self).__init__()
        assert d_model % num_heads == 0
        
        self.d_model = d_model
        self.num_heads = num_heads
        self.d_k = d_model // num_heads
        
        self.w_q = nn.Linear(d_model, d_model)
        self.w_k = nn.Linear(d_model, d_model)
        self.w_v = nn.Linear(d_model, d_model)
        self.w_o = nn.Linear(d_model, d_model)
        
        self.dropout = nn.Dropout(dropout)
        self.scale = math.sqrt(self.d_k)
    
    def forward(self, query, key, value, mask=None):
        batch_size = query.size(0)
        
        # 线性变换并分头

更多推荐