一、核心思想
扩散模型(Diffusion Model)是一种生成模型,受热力学中扩散过程的启发,通过模拟数据从噪声中逐步去噪的过程来生成样本。其核心思想是渐进式地添加噪声(正向过程)和逐步去噪(反向过程)。在正向过程中,逐步向数据中添加高斯噪声,最终将数据转化为纯噪声;在反向过程中,学习如何从噪声中逐步去噪,恢复出原始数据分布。
二、前向扩散过程(Forward Diffusion)
目标:将真实数据逐步“破坏”为随机噪声。
过程:对原始数据(如图像)进行 T 步微小的高斯噪声添加,每一步都让数据更接近纯噪声。
数学上,第t 步的状态由第 t-1 步的状态和噪声(服从标准正态分布)生成:
其中,是控制噪声强度的参数(逐渐接近随机噪声。
结果:经过 T 步后,原始数据完全转化为与训练数据无关的高斯噪声。
三、逆向扩散过程(Reverse Diffusion)
目标:从纯噪声中逐步“恢复”出有意义的数据(即生成新样本)。
过程:训练一个神经网络(通常是 U-Net 结构)学习“去噪”能力 —— 给定第 t 步的带噪声数据,预测它在第 t-1 步的状态(或直接预测添加的噪声)。
实际生成时,从随机噪声出发,利用训练好的网络反向迭代 T 步,每一步都去除部分噪声,最终得到接近真实数据分布的生成结果。
核心:神经网络通过学习噪声的分布规律,实现从噪声到数据的“逆推”。
四、Python示例
构建一个基础的扩散模型,用于生成一维数据。
import matplotlibmatplotlib.use('TkAgg')import numpy as npimport matplotlib.pyplot as pltimport torchimport torch.nn as nnimport torch.optim as optimfrom torch.utils.data import DataLoader, TensorDatasetplt.rcParams['font.sans-serif']=['SimHei'] # 中文支持plt.rcParams['axes.unicode_minus']=False # 负号显示# 设置随机种子,确保结果可复现np.random.seed(42)torch.manual_seed(42)# 生成一维数据(示例数据:混合高斯分布)def generate_data(n_samples=1000): # 生成两个高斯分布的数据 cluster1 = np.random.normal(loc=-2.0, scale=0.5, size=(n_samples // 2, 1)) cluster2 = np.random.normal(loc=2.0, scale=0.5, size=(n_samples // 2, 1)) data = np.vstack([cluster1, cluster2]) np.random.shuffle(data) return data# 前向过程:逐步添加噪声def forward_process(x_0, timesteps, betas): """ 执行扩散过程的前向步骤,逐步向数据添加噪声 """ # 计算alpha和alpha_bar alphas = 1. - betas alphas_cumprod = torch.cumprod(alphas, dim=0) # 随机选择一个时间步 t = torch.randint(0, timesteps, (x_0.shape[0],), device=x_0.device) # 从标准正态分布采样噪声 noise = torch.randn_like(x_0) # 计算x_t sqrt_alphas_cumprod_t = torch.sqrt(alphas_cumprod[t]).reshape(-1, 1) sqrt_one_minus_alphas_cumprod_t = torch.sqrt(1 - alphas_cumprod[t]).reshape(-1, 1) x_t = sqrt_alphas_cumprod_t * x_0 + sqrt_one_minus_alphas_cumprod_t * noise return x_t, t, noise# 简单的神经网络模型,用于预测噪声class SimpleDenoiser(nn.Module): def __init__(self, input_dim=1, hidden_dim=128): super(SimpleDenoiser, self).__init__() self.model = nn.Sequential( nn.Linear(input_dim + 1, hidden_dim), # +1 for time embedding nn.SiLU(), nn.Linear(hidden_dim, hidden_dim), nn.SiLU(), nn.Linear(hidden_dim, input_dim) ) def forward(self, x, t): # 将时间步t嵌入为模型输入的一部分 t_emb = t.unsqueeze(-1).float() x_with_t = torch.cat([x, t_emb], dim=1) return self.model(x_with_t)# 训练函数def train_diffusion_model(model, dataloader, num_epochs=1000, lr=1e-3): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model = model.to(device) optimizer = optim.Adam(model.parameters(), lr=lr) criterion = nn.MSELoss() # 定义扩散过程的参数 timesteps = 100 betas = torch.linspace(0.0001, 0.02, timesteps, device=device) for epoch in range(num_epochs): epoch_loss = 0.0 for batch in dataloader: x_0 = batch[0].to(device) # 前向过程:添加噪声 x_t, t, noise = forward_process(x_0, timesteps, betas) # 模型预测噪声 noise_pred = model(x_t, t) # 计算损失 loss = criterion(noise_pred, noise) # 反向传播和优化 optimizer.zero_grad() loss.backward() optimizer.step() epoch_loss += loss.item() if (epoch + 1) % 100 == 0: print(f"Epoch {epoch + 1}/{num_epochs}, Loss: {epoch_loss / len(dataloader):.6f}") return model# 采样函数:从噪声中生成数据def sample(model, sample_size=1000, timesteps=100): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model = model.to(device) model.eval() # 定义扩散过程的参数 betas = torch.linspace(0.0001, 0.02, timesteps, device=device) alphas = 1. - betas alphas_cumprod = torch.cumprod(alphas, dim=0) alphas_cumprod_prev = torch.cat([torch.tensor([1.], device=device), alphas_cumprod[:-1]]) sqrt_recip_alphas = torch.sqrt(1.0 / alphas) posterior_variance = betas * (1. - alphas_cumprod_prev) / (1. - alphas_cumprod) # 从标准正态分布开始采样 x = torch.randn(sample_size, 1, device=device) with torch.no_grad(): for i in reversed(range(timesteps)): t = torch.full((sample_size,), i, device=device, dtype=torch.long) noise_pred = model(x, t) # 计算均值 sqrt_recip_alphas_t = sqrt_recip_alphas[i] x = sqrt_recip_alphas_t * (x - betas[i] / torch.sqrt(1 - alphas_cumprod[i]) * noise_pred) # 添加方差(最后一步不添加) if i > 0: noise = torch.randn_like(x) posterior_variance_t = posterior_variance[i] x = x + torch.sqrt(posterior_variance_t) * noise return x.cpu().numpy()# 主函数def main(): # 生成数据 data = generate_data(n_samples=1000) data_tensor = torch.tensor(data, dtype=torch.float32) # 创建数据加载器 dataset = TensorDataset(data_tensor) dataloader = DataLoader(dataset, batch_size=32, shuffle=True) # 初始化模型 model = SimpleDenoiser(input_dim=1) # 训练模型 trained_model = train_diffusion_model(model, dataloader, num_epochs=1000) # 生成样本 samples = sample(trained_model, sample_size=1000) # 可视化结果 plt.figure(figsize=(12, 5)) plt.subplot(1, 2, 1) plt.hist(data, bins=50, density=True, alpha=0.7, label='真实数据') plt.title('真实数据分布') plt.xlabel('值') plt.ylabel('密度') plt.legend() plt.subplot(1, 2, 2) plt.hist(samples, bins=50, density=True, alpha=0.7, label='生成数据', color='orange') plt.title('扩散模型生成的数据分布') plt.xlabel('值') plt.ylabel('密度') plt.legend() plt.tight_layout() plt.show()if __name__ == "__main__": main()
示例展示了扩散模型的主要过程:
数据生成:使用两个高斯分布的混合作为示例数据
前向过程:逐步向数据添加噪声,最终将数据转换为噪声
模型架构:使用一个简单的神经网络来学习预测噪声
训练过程:通过最小化预测噪声与实际噪声之间的差异来训练模型
采样过程:从噪声开始,逐步恢复数据
五、小结
扩散模型通过“加噪-去噪”的框架,将生成问题转化为对噪声分布的逐步修正,其核心在于反向过程的参数化学习和噪声调度的设计。这一方法在生成任务中展现了强大的潜力,成为当前生成式AI的重要技术之一。