【Hugging Face 開源庫】Diffusers 庫 —

Diffusers 的三個主要組件
- 1. DiffusionPipeline：端到端推理工具
- - `__call__` 函數
  - `callback_on_step_end` 管道回調函數
- 2. 預訓練模型架構和模塊
- - UNet
  - VAE（Variational AutoEncoder）
  - 圖像尺寸與 UNet 和 VAE 的關系
  - EMA（Exponential Moving Average）
- 3. 調度器（Schedulers）

Diffusers 是 Hugging Face 開源的 Python 庫，專門用于加載、訓練和推理擴散模型（Diffusion Models）。

擴散模型是一類生成式模型，它們通過添加和去除噪聲來生成高質量圖像、音頻和視頻。

《從零開始學擴散模型》

在這里插入圖片描述

Diffusers 的三個主要組件

1. DiffusionPipeline：端到端推理工具

DiffusionPipeline 是 Diffusers 庫的核心組件之一，它提供了一個高層 API，幫助用戶快速從預訓練的擴散模型中生成樣本，而無需深入了解底層實現。

示例：使用 Stable Diffusion 生成圖像

from diffusers import StableDiffusionPipeline
import torch# 加載預訓練的 Stable Diffusion 模型
pipeline = StableDiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5")
pipeline.to("cuda")  # 使用 GPU 加速# 生成圖像
prompt = "a futuristic city at sunset, high detail, digital painting"
image = pipeline(prompt).images[0]# 顯示圖像
image.show()

通過 from_pretrained() 加載 Hugging Face Hub 上的 Stable Diffusion 預訓練模型。
unwayml/stable-diffusion-v1-5 是 Stable Diffusion v1.5 預訓練模型的 權重（weights），它被托管在 Hugging Face Hub 上，供用戶下載并進行推理或微調。
在 Diffusers 庫中，from_pretrained("runwayml/stable-diffusion-v1-5") 其實是加載該模型的預訓練參數，包括：
- UNet（去噪網絡）
- VAE（變分自編碼器，用于圖像編碼和解碼）
- Text Encoder（如 CLIP，用于處理文本輸入）
- 調度器（Scheduler，用于指導去噪過程）
這些組件的權重都是從 runwayml/stable-diffusion-v1-5 倉庫中下載的。
只需輸入 prompt（文本描述），就能生成相應的圖像。

`call` 函數

在 Python 中，__call__ 是一個特殊的方法，它 允許一個對象像函數一樣被調用。當你調用一個對象時，Python 實際上是調用了這個對象的 __call__ 方法。

在這里插入圖片描述
在 diffusers 庫中，所有的管道對象（如 StableDiffusionPipeline）都實現了一個 __call__ 方法，用于處理圖像生成任務，所以說 管道（pipeline）對象可以像函數一樣被調用。

讓我們實現一個 簡單的管道對象（Pipeline），用來模擬 Diffusers 的 __call__ 方法是如何工作的。這個管道將接受一個文本 prompt，然后通過一個簡單的 UNet 模型 生成一個偽圖像（這里只是模擬，不是實際的圖像生成）。

示例：實現一個簡單的 DiffusionPipeline

import torch
import torch.nn as nnclass SimpleUNet(nn.Module):""" 一個簡單的 UNet 模型模擬去噪過程 """def __init__(self):super().__init__()self.fc = nn.Linear(100, 100)  # 簡化的全連接層def forward(self, x):return self.fc(x)  # 這里只是簡單的線性變換class SimplePipeline:""" 一個簡單的管道對象，模擬 DiffusionPipeline 的 __call__ 方法 """def __init__(self):self.unet = SimpleUNet()  # 預訓練的去噪模型self.device = "cuda" if torch.cuda.is_available() else "cpu"self.unet.to(self.device)def __call__(self, prompt: str):""" 模擬調用管道進行圖像生成 """print(f"Processing prompt: {prompt}")# 1. 生成隨機噪聲作為輸入noise = torch.randn(1, 100).to(self.device)# 2. 通過 UNet 進行處理output = self.unet(noise)# 3. 模擬圖像輸出return output.detach().cpu().numpy()# 使用管道
pipeline = SimplePipeline()
generated_image = pipeline("A beautiful sunset over the ocean")  # 通過 __call__ 觸發
print("Generated image shape:", generated_image.shape)

SimpleUNet:
- 這里用一個簡單的 全連接層 代替真正的 UNet（通常是 CNN）。
- 這個網絡用于處理隨機噪聲，模擬去噪過程。
SimplePipeline:
- __init__ 方法：創建一個 UNet 模型并加載到 GPU（如果可用）。
- __call__ 方法：
  - ① 接收文本提示 prompt（但這里的代碼沒有真正解析文本，僅模擬處理）。
  - ② 生成隨機噪聲，作為輸入。
  - ③ 通過 UNet 處理，得到輸出。
  - ④ 返回最終“生成的圖像”（其實只是一個數值數組）。
如何使用 __call__ 方法:
- pipeline("A beautiful sunset over the ocean") 直接調用實例，會自動觸發 __call__ 方法。
- 這樣 對象本身就像一個函數一樣可以調用，符合 Diffusers 設計風格。

可以在 __call__ 方法中 添加真正的 VAE、文本編碼器、調度器 來讓它更接近 Diffusers 的 DiffusionPipeline。

這樣，pipeline("prompt") 的行為就類似于 StableDiffusionPipeline(prompt) 了！ 🚀

在實際的 diffusers 庫中，管道對象的 __call__ 方法會處理各種輸入嵌入、噪聲調度器、生成模型等，最終生成高質量的圖像。例如，在 StableDiffusionPipeline 中，__call__ 方法會接受提示、圖像嵌入等，并通過擴散模型逐步生成圖像。

`callback_on_step_end` 管道回調函數

callback_on_step_end 允許我們在 擴散管道的每一步去噪迭代結束時 執行 自定義回調函數。

這樣，可以 動態修改管道的屬性或調整張量，而 無需修改 Diffusers 庫的底層代碼。

舉個栗子，使用回調函數在去噪的不同階段動態調整 guidance_scale（引導比例），讓模型在去噪的前幾步加強條件引導（更遵循 prompt），后幾步減少 guidance_scale 以生成更自然的圖像。

import torch
from diffusers import StableDiffusionPipeline, DDIMScheduler# 加載 Stable Diffusion 管道
pipeline = StableDiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5")
pipeline.scheduler = DDIMScheduler.from_config(pipeline.scheduler.config) # 切換 DDIMScheduler 作為調度器
pipeline.to("cuda")# 定義回調函數
def dynamic_guidance_callback(pipe, i, latents):"""在去噪過程的每一步，動態修改 guidance_scale:param pipe: 當前管道對象:param i: 當前去噪步數:param latents: 當前的潛變量"""total_steps = pipe.scheduler.config.num_train_timestepsif i < total_steps * 0.3:  # 在前 30% 的步數里，增加 guidance_scalepipe.guidance_scale = 10.0  elif i < total_steps * 0.6:  # 在 30% - 60% 的步數里，降低 guidance_scalepipe.guidance_scale = 7.5  else:  # 在最后 40% 的步數里，進一步減少pipe.guidance_scale = 5.0  print(f"Step {i}: guidance_scale set to {pipe.guidance_scale}")# 生成圖像
prompt = "A futuristic city with neon lights at night"# 在 pipeline() 調用時傳遞 callback_on_step_end
image = pipeline(prompt, callback_on_step_end=dynamic_guidance_callback).images[0]# 顯示圖像
image.show()

這個回調函數在 每次去噪步驟結束后執行，并動態調整 guidance_scale：

前 30% 的步數：使用更高的 guidance_scale = 10.0，讓生成的圖像更符合 prompt 描述。
30% - 60% 步數：降低 guidance_scale 到 7.5，讓圖像稍微放松對 prompt 的嚴格約束。
最后 40% 步數：進一步降低到 5.0，讓圖像更自然，減少過度引導導致的“過擬合”問題。

Pipeline callbacks
除了動態調整 guidance_scale，還可以用 callback_on_step_end 進行：

添加自定義去噪步驟（比如在中間步驟插入額外的圖像操作）
修改 latents 變量（例如，在某些步數中加入額外的噪聲或調整顏色分布）
記錄或可視化去噪過程（比如，每隔 10 步保存當前的潛變量圖像，觀察去噪演化）

2. 預訓練模型架構和模塊

Diffusers 提供了許多 預訓練的模型組件，可以用來構建新的擴散系統，例如：

UNet（去噪神經網絡）
VAE（Variational Autoencoder）（用于圖像編碼和解碼）
Text Encoder（例如 CLIP，用于理解文本提示）

示例：使用 UNet 作為去噪模型

from diffusers import UNet2DModel# 定義一個 UNet 模型
unet = UNet2DModel(sample_size=64,  # 圖像大小in_channels=3,    # RGB 顏色通道out_channels=3,layers_per_block=2,block_out_channels=(64, 128, 256),
)# 查看模型參數
print(unet)

UNet2DModel 是擴散模型的核心組件之一，負責在訓練和推理過程中去噪。
這里的 UNet 結構可以自定義，如通道數、塊的層數等。

UNet

U-Net: Convolutional Networks for Biomedical Image Segmentation

Unet 最初設計用于生物醫學圖像分割。
在這里插入圖片描述
UNet 是一種 卷積神經網絡 架構，結構類似于一個對稱的 U 字形，由 編碼器（下采樣）和解碼器（上采樣） 組成。

編碼器逐步提取圖像特征并縮小空間維度，
解碼器則將這些特征還原到原始的空間維度，同時逐步增加分辨率。

UNet 的關鍵特性：

對稱結構：編碼器和解碼器對稱分布。
跳躍連接：直接將編碼器的中間層輸出傳遞到解碼器的對應層，保留了高分辨率特征。
多尺度特征提取：在不同尺度上提取特征，提升了網絡對細節的捕捉能力。

在這里插入圖片描述

VAE（Variational AutoEncoder）

VAE（Variational AutoEncoder）變分自編碼器是一種生成模型，通過學習輸入數據的潛在表示來生成新數據。

VAE 由編碼器和解碼器組成：

編碼器：將 輸入圖像 轉換為 潛在空間的分布（均值和方差）。
解碼器：從潛在空間的采樣生成 新圖像。

VAE 的關鍵特性：

概率模型：VAE 學習輸入數據的概率分布，從而生成多樣化的樣本。
連續潛在空間：潛在空間中的小變化會導致生成圖像的小變化，具有很好的連續性。

圖像尺寸與 UNet 和 VAE 的關系

在圖像生成任務中，輸入圖像的尺寸需要匹配 UNet 和 VAE 的預期輸入輸出尺寸。

在 diffusers 庫的 MimicBrushPipeline（或類似的圖像生成管道）中，默認的輸入圖像尺寸是通過以下代碼計算的：

height = height or self.unet.config.sample_size * self.vae_scale_factor
width = width or self.unet.config.sample_size * self.vae_scale_factor

Stable Diffusion 生成圖像時，涉及 VAE（變分自編碼器）和 UNet（去噪網絡）：

VAE 作用：將高清圖像壓縮成一個 低維潛空間（latent space），然后再解碼回原始尺寸。
UNet 作用：在潛空間中去噪，逐步優化潛變量，使其接近真實圖像的潛變量。

關鍵點：VAE 會對圖像進行 vae_scale_factor 倍縮放。舉個栗子吧，

輸入 VAE 的圖像： 512×512
經過 VAE 編碼后： 512/8 = 64×64（縮小 8 倍）
UNet 處理的就是 64 × 64 的潛變量。

所以：

height=64×8=512
width=64×8=512

這確保了：

UNet 處理 64 × 64 潛變量時尺寸正確。
VAE 進行解碼時，最終輸出的是 512 × 512 的圖像。

EMA（Exponential Moving Average）

EMA（指數移動平均）是一種 平滑技術，在深度學習中，常用于 存儲模型可學習參數的局部平均值。

可以把它看作一個“影子模型”，它的參數不是簡單地復制原模型，而是隨著訓練 以指數衰減的方式 逐步向原模型靠攏。

為什么要使用 EMA？

提高模型穩定性：在訓練過程中，模型參數可能會劇烈波動，EMA 平均化了參數，使其更穩定。
提升泛化能力：直接使用 EMA 計算的參數進行推理，通常比原始參數表現更好，尤其是在 少量訓練步數 下。
適用于生成模型（如 Diffusion Models）：Diffusers 庫中的 Stable Diffusion 訓練時使用 EMA 來平滑 UNet 權重，使生成的圖像更加穩定。
在半監督學習中常用：如 Mean Teacher 方法，使用 EMA 計算的模型作為“教師”模型指導學生模型學習。

EMA 在累積歷史信息的同時，更關注最近的更新，從而對新數據變化更敏感，而不會受太早的參數擾動。

假設：

$\theta_t$ 是第 $t$ 輪訓練的模型參數
$\theta_{\text{EMA},t}$ 是第 $t$ 輪的 EMA 計算的影子參數
$\alpha$ 是 EMA 衰減系數（通常取 0.99 ~ 0.999）

EMA 參數的更新方式：
$\theta_{\text{EMA},t} = \alpha \cdot \theta_{\text{EMA},t-1} + (1 - \alpha) \cdot \theta_t$
這意味著：

較早的參數影響力逐漸減弱（因為乘以了 $\alpha$ ）。
最近的參數更新權重更大（乘以 $\alpha$ ）。
選擇 較大的 $\alpha$ （如 0.999），EMA 更新較慢，適用于平滑長時間的變化。

為什么較早的參數影響力逐漸減弱？

我們可以將 EMA 當前參數展開，看看它是如何由歷史所有參數的加權平均組成的：

$\theta_{\text{EMA},t} = (1 - \alpha) \cdot \theta_t + \alpha (1 - \alpha) \cdot \theta_{t-1} + \alpha^2 (1 - \alpha) \cdot \theta_{t-2} + \alpha^3 (1 - \alpha) \cdot \theta_{t-3} + \dots$

這說明：

最近的參數 $\theta_t$ 乘以 $\alpha$ （即 0.01），雖然數值小，但它是最新的更新，影響直接而強烈。
較早的參數 $\theta_{t-1}, \theta_{t-2}$ 乘以 $\alpha, \alpha^2$ 等次冪，影響力隨著時間推移呈指數級衰減。
老的參數貢獻依然存在，但比重越來越小，這使得 EMA 更關注近期變化，而不會被早期的不穩定訓練步驟影響太多。

💡直覺理解 EMA 的本質是一種帶有“記憶衰減”的平滑機制：

老的參數不會立刻丟失，但它的影響會隨著時間逐步減弱，讓新數據有更大的話語權。
雖然最近參數的權重（1 - α = 0.01）看似小，但它不會被 EMA 繼續削弱，因此它的相對影響力更大。
較早的參數影響力會隨著 $\alpha^t$ 指數級減少，長期來看其貢獻會趨近于 0。

如果 $\alpha = 0.99$ ，那么過去 5 個時間步的參數貢獻依次為：
$\begin{aligned} \text{Step } t: & \quad (1 - \alpha) = 0.01 \\ \text{Step } t-1: & \quad 0.99 \times 0.01 = 0.0099 \\ \text{Step } t-2: & \quad 0.99^2 \times 0.01 = 0.009801 \\ \text{Step } t-3: & \quad 0.99^3 \times 0.01 = 0.00970299 \\ \text{Step } t-4: & \quad 0.99^4 \times 0.01 = 0.0096059601 \\ \end{aligned}$

下面是一個簡單的 PyTorch EMA 代碼示例，展示如何在訓練過程中維護一個 EMA 版本的模型參數。

import torch
import torch.nn as nnclass EMA:"""指數移動平均（EMA），用于平滑模型參數"""def __init__(self, model, decay=0.999):self.model = modelself.decay = decay  # EMA 影子參數衰減系數self.shadow = {name: param.clone().detach() for name, param in model.named_parameters()}def update(self):"""更新 EMA 影子模型參數"""for name, param in self.model.named_parameters():if param.requires_grad:self.shadow[name] = self.decay * self.shadow[name] + (1 - self.decay) * param.detach()def apply_shadow(self):"""使用 EMA 參數更新原模型（推理時調用）"""for name, param in self.model.named_parameters():if param.requires_grad:param.data.copy_(self.shadow[name])# 創建簡單的神經網絡
class SimpleModel(nn.Module):def __init__(self):super().__init__()self.fc = nn.Linear(10, 1)def forward(self, x):return self.fc(x)# 初始化模型和 EMA 影子模型
model = SimpleModel()
ema = EMA(model, decay=0.99)# 模擬訓練過程
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
for step in range(100):# 訓練步驟（假設 x 是輸入數據）x = torch.randn(16, 10)loss = model(x).mean()optimizer.zero_grad()loss.backward()optimizer.step()# 更新 EMA 影子模型ema.update()if step % 10 == 0:print(f"Step {step}: loss={loss.item():.4f}")# 在推理時應用 EMA 參數
ema.apply_shadow()

EMA 類
- 維護了 shadow（影子模型參數）。
- 通過 update() 逐步更新 EMA 版本的參數。
- apply_shadow() 用于推理時將 EMA 參數應用到原模型上。
訓練過程中
- 每次模型參數更新后，調用 ema.update()，讓影子模型參數緩慢跟隨原模型更新。
推理時
- ema.apply_shadow() 把 EMA 版本的參數復制到模型，通常能獲得 更好的性能。

在 diffusers 庫中，EMA 主要用于 訓練 UNet（去噪網絡）：

訓練過程中，EMA 版本的 UNet 逐步更新。
在推理時，使用 EMA 版本的 UNet 進行采樣，以提高圖像質量。

Diffusers 使用 EMAModel 進行 EMA 計算：
from diffusers.models import EMAModel
# 初始化 EMA 模型
ema_unet = EMAModel(pipeline.unet.parameters(), decay=0.999)
# 在訓練后更新 EMA 影子模型
ema_unet.step(pipeline.unet.parameters())
# 復制 EMA 參數到 UNet（推理時）
ema_unet.copy_to(pipeline.unet.parameters())

3. 調度器（Schedulers）

Scheduler，中文譯為“調度器”，在擴散模型中負責控制噪聲的添加和去除過程。

它定義了 在每個擴散步驟中，向數據添加多少噪聲，以及在去噪過程中如何逐步恢復原始數據。

在這里插入圖片描述

Diffusers 庫提供了多種調度器，例如：

DDIMScheduler（去噪擴散隱變量模型）
PNDMScheduler（更快的推理）
DPMSolverMultistepScheduler（更穩定的采樣）

示例：使用不同調度器進行推理

from diffusers import StableDiffusionPipeline, DPMSolverMultistepScheduler# 加載 Stable Diffusion 并更換調度器
pipeline = StableDiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5")
pipeline.scheduler = DPMSolverMultistepScheduler.from_config(pipeline.scheduler.config)# 生成圖像
prompt = "a magical forest with glowing trees"
image = pipeline(prompt).images[0]
image.show()