扩散模型统一理论:测度论视角
概述
扩散模型、Score Matching和Flow Matching是当前生成式AI的核心技术12。最近的理论工作提出了一个统一的测度论框架,将这三类方法纳入同一理论体系3。本章从测度论角度深入分析这些方法的数学基础和内在联系。
测度论基础
概率测度与随机过程
设 为概率空间, 为随机变量。
定义(概率分布):随机变量 的概率分布是测度空间 上的概率测度 :
Kullback-Leibler散度
两个概率分布 和 之间的KL散度:
前提是 ( 关于 绝对连续)。
Wasserstein距离
表示具有 阶矩的概率分布空间:
其中 是边际为 的联合分布集合。
随机微分方程视角
前向扩散过程
设 为 上的随机过程:
其中:
- 是漂移函数(drift)
- 是扩散系数(diffusion coefficient)
- 是标准布朗运动
特殊情况:DDPM使用常系数:
SDE的解与转移概率
SDE的解给出转移概率核 :
Fokker-Planck方程
SDE的前向过程对应Fokker-Planck方程:
扩散模型的理论框架
证据下界(ELBO)
设 为数据生成的前向过程, 为模型。ELBO:
展开后:
Score Matching目标
Stein分数:
score matching损失4:
去噪扩散模型
DDPM的训练目标简化形式:
Flow Matching
条件概率路径
Flow Matching通过定义确定性的概率路径 从 (噪声)到 (数据):
这不是合理的概率密度混合,而是插值分布。
最优传输Flow Matching
最优传输条件Flow Matching (OT-CFM)5:
这定义了一个向量场 ,其积分曲线从噪声流向数据。
速度预测模型
Flow Matching训练速度预测模型 :
其中 (线性插值)。
统一测度论框架
核心定理
定理(统一框架3):扩散模型、Score Matching和Flow Matching都可以统一为**Wasserstein梯度流(Wasserstein Gradient Flow)**的学习问题。
Wasserstein梯度流
概率分布空间上的梯度流定义为:
其中 是自由能泛函:
三种方法的统一视角
| 方法 | 梯度流形式 | 目标泛函 |
|---|---|---|
| Score Matching | 匹配真实分布 | |
| Flow Matching | 沿向量场流动 | |
| DDPM | 最小化KL散度 |
统一损失函数
在统一框架下,所有方法都可以表示为:
其中:
- 是模型预测的向量场/分数
- 是目标向量场(具体形式取决于方法)
训练动态分析
分数空间的收敛性
设 为真实分数, 为预测分数。
收敛定理:在适当条件下,当 优化至最优时:
分布收敛
定理:如果训练收敛,则生成分布 满足:
其中 与训练误差相关。
实现细节
统一训练框架
import torch
import torch.nn as nn
from typing import Callable, Tuple
class UnifiedDiffusionFramework:
"""
统一的扩散/Flow Matching训练框架
支持:DDPM, Score Matching, Flow Matching
"""
def __init__(
self,
model: nn.Module,
method: str = 'ddpm', # 'ddpm', 'score_matching', 'flow_matching'
sigma_min: float = 0.01,
sigma_max: float = 50.0,
):
self.model = model
self.method = method
self.sigma_min = sigma_min
self.sigma_max = sigma_max
def get_noise_schedule(self, t: torch.Tensor) -> torch.Tensor:
"""噪声调度"""
if self.method == 'ddpm':
# DDPM调度
return self.sigma_min * (self.sigma_max / self.sigma_min) ** t
elif self.method == 'flow_matching':
# Flow Matching:线性调度
return self.sigma_min + (self.sigma_max - self.sigma_min) * t
else:
# Score Matching
return torch.exp(t * torch.log(self.sigma_max) +
(1-t) * torch.log(self.sigma_min))
def compute_target(
self,
x0: torch.Tensor,
xt: torch.Tensor,
t: torch.Tensor,
noise: torch.Tensor,
) -> torch.Tensor:
"""
计算目标向量(根据方法不同)
Returns:
目标向量:分数、速度或噪声
"""
sigma = self.get_noise_schedule(t)
if self.method == 'ddpm':
# DDPM:预测噪声
return noise
elif self.method == 'score_matching':
# Score Matching:预测分数
return -noise / sigma.unsqueeze(-1)
elif self.method == 'flow_matching':
# Flow Matching:预测速度
return x0 - xt # 从噪声到数据的向量
def training_loss(
self,
x0: torch.Tensor,
) -> Tuple[torch.Tensor, dict]:
"""
计算训练损失
Args:
x0: 真实数据 [batch, dim]
Returns:
loss: 标量损失
info: 诊断信息
"""
batch_size = x0.shape[0]
device = x0.device
# 采样时间步
t = torch.rand(batch_size, device=device)
# 采样噪声
noise = torch.randn_like(x0)
# 获取噪声调度
sigma = self.get_noise_schedule(t)
# 加噪
xt = x0 + sigma.unsqueeze(-1) * noise
# 模型预测
pred = self.model(xt, t)
# 计算目标
target = self.compute_target(x0, xt, t, noise)
# 计算损失
loss = 0.5 * ((pred - target) ** 2).sum(dim=-1).mean()
info = {
't_mean': t.mean().item(),
'sigma_mean': sigma.mean().item(),
'pred_norm': pred.norm().item() / batch_size,
}
return loss, info
@torch.no_grad()
def sampling(
self,
shape: Tuple[int, ...],
num_steps: int = 100,
ode_solver: str = 'euler',
) -> torch.Tensor:
"""
采样生成
Args:
shape: 输出形状
num_steps: 采样步数
ode_solver: ODE求解器 ('euler', 'heun')
Returns:
x0: 生成样本
"""
x = torch.randn(shape, device=next(self.model.parameters()).device)
dt = 1.0 / num_steps
for i in range(num_steps):
t = torch.full((shape[0],), i / num_steps, device=x.device)
# 预测向量场
v = self.model(x, t)
if self.method == 'ddpm':
# DDPM采样(离散)
alpha_bar = 1 - (self.sigma_min ** 2) * \
(self.sigma_max ** 2 / self.sigma_min ** 2) ** t
pred_x0 = (x - (1 - alpha_bar).sqrt() * v) / alpha_bar.sqrt()
x = x - (1 - alpha_bar) * pred_x0 + v * dt
else:
# Flow Matching / Score Matching
x = x + v * dt
return x概率路径的可视化
import matplotlib.pyplot as plt
import numpy as np
def visualize_probability_paths(
framework: UnifiedDiffusionFramework,
data_samples: np.ndarray,
num_times: int = 5,
):
"""
可视化不同方法的概率路径
Args:
data_samples: 真实数据样本 [n_samples, dim]
num_times: 要可视化的时间步数
"""
fig, axes = plt.subplots(1, 3, figsize=(15, 4))
methods = ['ddpm', 'score_matching', 'flow_matching']
titles = ['DDPM', 'Score Matching', 'Flow Matching']
for ax, method, title in zip(axes, methods, titles):
framework.method = method
for i, t_val in enumerate(np.linspace(0, 1, num_times)):
t = torch.full((len(data_samples),), t_val)
sigma = framework.get_noise_schedule(t).numpy()
# 加噪
noise = np.random.randn(*data_samples.shape)
xt = data_samples + sigma[:, None] * noise
ax.scatter(xt[:, 0], xt[:, 1],
alpha=0.5, s=10,
label=f't={t_val:.1f}' if i == 0 else None)
ax.set_title(title)
ax.legend()
ax.set_xlabel('x₁')
ax.set_ylabel('x₂')
plt.tight_layout()
plt.savefig('probability_paths.png')
plt.show()实践注意事项
方法选择指南
| 场景 | 推荐方法 | 理由 |
|---|---|---|
| 高维图像生成 | DDPM / Flow Matching | 理论基础扎实 |
| 快速采样 | Flow Matching / Consistency | 收敛快 |
| 连续时间建模 | Score Matching | SDE框架自然 |
| 密度估计 | Score Matching | 分数易计算 |
训练技巧
- 噪声调度:DDPM推荐 调度,Flow Matching推荐线性调度
- 时间嵌入:使用Transformer式的时间编码
- 网络架构:U-Net + Self-Attention是图像生成的标准架构
- 正则化:避免数值不稳定,特别是大时间步
统一理论的意义
理论价值
- 统一视角:将三类方法纳入同一数学框架
- 理论保证:提供收敛性和最优性保证
- 新方法发现:为设计新方法提供指导
实践价值
- 灵活切换:不同场景可选择最优方法
- 性能预测:理论指导实践参数选择
- 组合优化:混合不同方法的优点
总结
从测度论视角,扩散模型、Score Matching和Flow Matching都统一于Wasserstein梯度流框架:
- 数学基础:SDE、Fokker-Planck方程、Wasserstein距离
- 统一目标:学习目标向量场,使分布沿梯度流演化
- 方法差异:体现在目标向量场 的定义方式
- 实践选择:根据任务特性选择合适的方法和调度策略
这一统一理论不仅深化了我们对生成模型的理解,也为未来更强大的生成方法奠定了理论基础。
参考文献
Footnotes
-
Ho et al. (2020). “Denoising Diffusion Probabilistic Models.” NeurIPS 2020 ↩
-
Song et al. (2021). “Score-Based Generative Modeling through Stochastic Differential Equations.” ICLR 2021 ↩
-
[arXiv 2605.06829] “A Unified Measure-Theoretic View of Diffusion, Score-Based, and Flow Matching Models” ↩ ↩2
-
Hyvarinen (2005). “Estimation of Non-Normalized Statistical Models by Score Matching.” JMLR 2005 ↩
-
Lipman et al. (2022). “Flow Matching for Generative Modeling.” ICLR 2022 ↩