扩散模型统一理论:测度论视角

概述

扩散模型、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 MatchingSDE框架自然
密度估计Score Matching分数易计算

训练技巧

  1. 噪声调度:DDPM推荐 调度,Flow Matching推荐线性调度
  2. 时间嵌入:使用Transformer式的时间编码
  3. 网络架构:U-Net + Self-Attention是图像生成的标准架构
  4. 正则化:避免数值不稳定,特别是大时间步

统一理论的意义

理论价值

  1. 统一视角:将三类方法纳入同一数学框架
  2. 理论保证:提供收敛性和最优性保证
  3. 新方法发现:为设计新方法提供指导

实践价值

  1. 灵活切换:不同场景可选择最优方法
  2. 性能预测:理论指导实践参数选择
  3. 组合优化:混合不同方法的优点

总结

从测度论视角,扩散模型、Score Matching和Flow Matching都统一于Wasserstein梯度流框架:

  1. 数学基础:SDE、Fokker-Planck方程、Wasserstein距离
  2. 统一目标:学习目标向量场,使分布沿梯度流演化
  3. 方法差异:体现在目标向量场 的定义方式
  4. 实践选择:根据任务特性选择合适的方法和调度策略

这一统一理论不仅深化了我们对生成模型的理解,也为未来更强大的生成方法奠定了理论基础。

参考文献

Footnotes

  1. Ho et al. (2020). “Denoising Diffusion Probabilistic Models.” NeurIPS 2020

  2. Song et al. (2021). “Score-Based Generative Modeling through Stochastic Differential Equations.” ICLR 2021

  3. [arXiv 2605.06829] “A Unified Measure-Theoretic View of Diffusion, Score-Based, and Flow Matching Models” 2

  4. Hyvarinen (2005). “Estimation of Non-Normalized Statistical Models by Score Matching.” JMLR 2005

  5. Lipman et al. (2022). “Flow Matching for Generative Modeling.” ICLR 2022