扩散模型与生成式建模：DDPM、U-Net、潜在扩散与 LoRA 微调实战

前向过程（固定，无需学习）：x0 → x1 → ... → xT   （逐步加高斯噪声）
反向过程（需要学习）      ：xT → ... → x1 → x0   （逐步去噪）

x_t = √(ᾱ_t) · x_0 + √(1 - ᾱ_t) · ε     其中 ε ~ N(0, I)，ᾱ_t = ∏(1-β_i)

import torch

def q_sample(x0, t, alphas_cumprod):
    """前向扩散：从 x0 直接采样 x_t"""
    sqrt_alpha_bar = alphas_cumprod[t].sqrt().view(-1, 1, 1, 1)
    sqrt_one_minus = (1 - alphas_cumprod[t]).sqrt().view(-1, 1, 1, 1)
    noise = torch.randn_like(x0)
    return sqrt_alpha_bar * x0 + sqrt_one_minus * noise, noise

import torch.nn.functional as F

def train_step(model, x0, alphas_cumprod, optimizer):
    batch = x0.size(0)
    t = torch.randint(0, 1000, (batch,), device=x0.device)   # 随机时间步
    x_t, noise = q_sample(x0, t, alphas_cumprod)             # 加噪
    noise_pred = model(x_t, t)                               # 预测噪声
    loss = F.mse_loss(noise_pred, noise)                     # 简单 MSE
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
    return loss.item()

import numpy as np

def cosine_schedule(timesteps=1000, s=0.008):
    steps = timesteps + 1
    x = np.linspace(0, timesteps, steps)
    alphas_cumprod = np.cos(((x / timesteps) + s) / (1 + s) * np.pi * 0.5) ** 2
    alphas_cumprod = alphas_cumprod / alphas_cumprod[0]
    betas = 1 - (alphas_cumprod[1:] / alphas_cumprod[:-1])
    return np.clip(betas, 0.0001, 0.9999)

输入 x_t (H×W×C)
  ↓ 编码器（下采样，通道翻倍）
  ↓ 中间块（自注意力 + 残差）
  ↑ 解码器（上采样 + 跳跃连接）
输出 ε_pred (H×W×C)
时间嵌入 t → 注入每个残差块

import torch
import torch.nn as nn

class DiTBlock(nn.Module):
    def __init__(self, dim, heads, mlp_ratio=4.0):
        super().__init__()
        self.norm1 = nn.LayerNorm(dim)
        self.attn = nn.MultiheadAttention(dim, heads, batch_first=True)
        self.norm2 = nn.LayerNorm(dim)
        self.mlp = nn.Sequential(
            nn.Linear(dim, int(dim * mlp_ratio)), nn.GELU(),
            nn.Linear(int(dim * mlp_ratio), dim),
        )
        # adaLN：时间/条件通过调制 scale/shift 注入
        self.ada = nn.Linear(dim, dim * 6)

    def forward(self, x, cond):
        shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = \
            self.ada(cond).chunk(6, dim=-1)
        h = self.norm1(x) * (1 + scale_msa.unsqueeze(1)) + shift_msa.unsqueeze(1)
        x = x + gate_msa.unsqueeze(1) * self.attn(h, h, h)[0]
        h = self.norm2(x) * (1 + scale_mlp.unsqueeze(1)) + shift_mlp.unsqueeze(1)
        x = x + gate_mlp.unsqueeze(1) * self.mlp(h)
        return x

图像 512×512×3 → [VAE 编码器] → 隐表示 64×64×4
                                    ↓ 在隐空间做扩散（快 48 倍）
隐表示 64×64×4 → [VAE 解码器] → 图像 512×512×3

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

class DiagonalGaussian(nn.Module):
    """把编码器输出拆成 mean 和 logvar，重参数化采样"""
    def forward(self, z):
        mean, logvar = torch.chunk(z, 2, dim=1)
        logvar = torch.clamp(logvar, -30.0, 20.0)
        std = torch.exp(0.5 * logvar)
        if self.training:
            return mean + std * torch.randn_like(mean)
        return mean

from diffusers import AutoencoderKL

vae = AutoencoderKL.from_pretrained("stabilityai/sd-vae-ft-mse")
scale_factor = vae.config.scaling_factor   # 0.18215

文本 "a red cup" → [Text Encoder] → 文本嵌入
                                      ↓ 交叉注意力（Q 来自图像，K/V 来自文本）
图像特征 ─────────────────────────────┘

ε_guided = ε_uncond + w · (ε_cond − ε_uncond)

# 训练时随机把条件置空（10% 概率），让模型同时学会有条件/无条件
import random
if random.random() < 0.1:
    text_emb = null_text_emb   # 无条件

# 推理时 CFG
noise_uncond = model(x_t, t, null_text_emb)
noise_cond = model(x_t, t, text_emb)
noise = noise_uncond + guidance_scale * (noise_cond - noise_uncond)

from diffusers import StableDiffusionPipeline, DPMSolverMultistepScheduler

pipe = StableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5", torch_dtype=torch.float16
).to("cuda")

# 换 DPM-Solver，20 步出图
pipe.scheduler = DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)

image = pipe("a red cup on a wooden table", num_inference_steps=20,
             guidance_scale=7.5).images[0]

import torch
from diffusers import StableDiffusionPipeline, DPMSolverMultistepScheduler

pipe = StableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    torch_dtype=torch.float16,
    safety_checker=None,
).to("cuda")
pipe.scheduler = DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
pipe.enable_attention_slicing()   # 省显存

prompt = "a red ceramic cup on a wooden table, soft window light, 50mm photo"
negative = "blurry, low quality, watermark"

image = pipe(
    prompt=prompt,
    negative_prompt=negative,
    num_inference_steps=25,
    guidance_scale=7.5,
    generator=torch.Generator("cuda").manual_seed(42),
).images[0]
image.save("cup.png")

accelerate launch train_dreambooth_lora.py \
  --pretrained_model_name_or_path="runwayml/stable-diffusion-v1-5" \
  --instance_data_dir="./my_concept" \
  --instance_prompt="a photo of sks cup" \
  --resolution=512 --train_batch_size=1 \
  --gradient_accumulation_steps=4 \
  --learning_rate=1e-4 --max_train_steps=800 \
  --mixed_precision="fp16" --output_dir="./lora_cup"

pipe.load_lora_weights("./lora_cup")
pipe.fuse_lora(lora_scale=0.8)   # 0.8 控制 LoRA 影响强度
image = pipe("a photo of sks cup on a beach", num_inference_steps=25).images[0]

数学层：前向加噪 + 反向去噪（DDPM）
调度层：余弦 β + v-prediction
架构层：U-Net（中小规模）/ DiT（大规模）
效率层：VAE 潜在空间 + DPM-Solver 采样加速
控制层：CFG + ControlNet + LoRA