企业官网建设流程全解析

从零拆解 Diffusers：用模型与调度器手写你自己的扩散系统

【免费下载链接】diffusers🤗 Diffusers: State-of-the-art diffusion models for image, video, and audio generation in PyTorch.项目地址: https://gitcode.com/GitHub_Trending/di/diffusers

本文以 🤗 Diffusers 官方教程《Understanding pipelines, models and schedulers》为主线，讲解如何拆解DiffusionPipeline，将模型（Model）与调度器（Scheduler）解耦，从零手写一套可运行的扩散采样系统。内容覆盖基础 DDPM 管线的完整去噪循环，以及 Stable Diffusion 这种多组件条件扩散模型的逐段实现，读完你可以不依赖DiffusionPipeline，仅用UNet2DModel、UNet2DConditionModel、AutoencoderKL、CLIP 文本编码器与各类调度器，自主组装出任意扩散系统。

核心思路：管线只是模型 + 调度器的"打包"

🧨 Diffusers 被设计成一个面向扩散系统的灵活工具箱，其核心只有两类构件：模型（models）与调度器（schedulers）。

• 模型负责学习去噪能力，例如 UNet2DModel、UNet2DConditionModel 与 AutoencoderKL；
• 调度器负责定义扩散过程的数学规则，例如 DDPMScheduler、PNDMScheduler、UniPCMultistepScheduler 等，它们位于 src/diffusers/schedulers 目录下。

DiffusionPipeline只是把二者打包以提供开箱即用的便利，但它完全可以被"解绑"——这也正是官方教程倡导的：把管线拆开，用模型和调度器分别组装，从而创建属于你自己的扩散系统。整个教程的核心结论只有一句话：无论多复杂的扩散系统，本质都是一个"去噪循环（denoising loop）"。

拆解基础管线：DDPMPipeline 内部发生了什么

用DDPMPipeline生成一张图片只需四行代码：

>>> from diffusers import DDPMPipeline >>> ddpm = DDPMPipeline.from_pretrained("google/ddpm-cat-256").to("cuda") >>> image = ddpm(num_inference_steps=25).images[0] >>> image

从源码看，DDPMPipeline 的构造函数只注册了两个模块：

class DDPMPipeline(DiffusionPipeline): model_cpu_offload_seq = "unet" def __init__(self, unet: UNet2DModel, scheduler: DDPMScheduler): super().__init__() self.register_modules(unet=unet, scheduler=scheduler)

也就是说这个管线内部只有UNet2DModel模型和DDPMScheduler调度器。它的工作方式是：生成一张与期望输出尺寸相同的随机噪声图，让模型反复"过"这张图多次；在每一个 timestep，模型预测出noise residual（噪声残差），调度器利用该残差预测出"噪声更少"的上一张图；如此循环，直到跑完设定的推理步数。

对照 pipeline_ddpm.py 的__call__实现，可以看到这 4 行代码背后其实就是一个标准去噪循环：

# 1. 采样高斯噪声作为起点 image = randn_tensor(image_shape, generator=generator, device=device, dtype=self.unet.dtype) # 2. 设置推理步数 self.scheduler.set_timesteps(num_inference_steps) for t in self.progress_bar(self.scheduler.timesteps): # 3. 模型预测噪声 model_output = self.unet(image, t).sample # 4. 调度器计算上一时刻图像 x_t -> x_t-1 image = self.scheduler.step(model_output, t, image, generator=generator).prev_sample # 5. 将张量归一化到 [0,1] 并转为 PIL 图像 image = (image / 2 + 0.5).clamp(0, 1)

这个模式就是所有扩散系统的通用骨架。下面我们把它手动拆开写一遍。

手写去噪循环：DDPM 六步走

第 1 步：加载模型与调度器

与直接加载管线不同，这里把两个组件分开加载：

>>> from diffusers import DDPMScheduler, UNet2DModel >>> scheduler = DDPMScheduler.from_pretrained("google/ddpm-cat-256") >>> model = UNet2DModel.from_pretrained("google/ddpm-cat-256").to("cuda")

UNet2DModel是无条件图像去噪模型，输出与输入同尺寸；DDPMScheduler负责噪声调度。

第 2 步：设置去噪的 timestep 数量

>>> scheduler.set_timesteps(50)

set_timesteps会在调度器上生成一条等间距的 timestep 序列，其长度等于推理步数。可以在 scheduling_ddpm.py 中看到其实现逻辑：它会根据timestep_spacing配置（linspace/leading/trailing）从训练时的总步数（默认 1000）中等间隔抽取推理步，并反转成严格递减的序列；同时校验num_inference_steps不得超过训练步数num_train_timesteps。

第 3 步：查看生成的 timesteps 张量

>>> scheduler.timesteps tensor([980, 960, 940, 920, 900, 880, 860, 840, 820, 800, 780, 760, 740, 720, 700, 680, 660, 640, 620, 600, 580, 560, 540, 520, 500, 480, 460, 440, 420, 400, 380, 360, 340, 320, 300, 280, 260, 240, 220, 200, 180, 160, 140, 120, 100, 80, 60, 40, 20, 0])

每个元素代表模型对图像去噪的一个时间刻度，从接近完全噪声的 980 一直递减到 0。后续去噪循环就是迭代这个张量。

第 4 步：生成与目标输出同形状的随机噪声

>>> import torch >>> sample_size = model.config.sample_size >>> noise = torch.randn((1, 3, sample_size, sample_size), device="cuda")

model.config.sample_size来自 UNet 的配置文件，定义了模型期望的输入分辨率（本案例为 256）；通道数 3 对应 RGB。

第 5 步：编写去噪循环

>>> input = noise >>> for t in scheduler.timesteps: ... with torch.no_grad(): ... noisy_residual = model(input, t).sample ... previous_noisy_sample = scheduler.step(noisy_residual, t, input).prev_sample ... input = previous_noisy_sample

循环每一步做两件事：

1. 模型前向：model(input, t)调用UNet2DModel.forward，预测当前噪声图像中的噪声残差；
2. 调度器反向一步：scheduler.step(noisy_residual, t, input)返回prev_sample，即上一时刻（噪声更少）的图像，作为下一轮模型输入。

DDPMScheduler.step的内部实现（见 scheduling_ddpm.py）揭示了其数学本质：先由预测的噪声反推"预测原始样本"（pred_original_sample，对应 DDPM 论文公式 15，依据prediction_type为epsilon/sample/v_prediction采用不同反推公式），再按公式 (7) 组合出上一时刻样本的均值，最后按variance_type决定是否叠加随机方差项。整个过程即是对扩散 SDE 的反向求解。

第 6 步：把去噪结果转成图片

>>> from PIL import Image >>> import numpy as np >>> image = (input / 2 + 0.5).clamp(0, 1) >>> image = image.cpu().permute(0, 2, 3, 1).numpy()[0] >>> image = Image.fromarray((image * 255).round().astype("uint8")) >>> image

先将张量从[-1, 1]归一化到[0, 1]，再转换通道顺序（NCHW → NHWC）并映射到uint8的 0~255 范围，最后包装成 PIL 图像。

至此，整个去噪循环完成。这个模式可以复用到任意扩散系统中——差异只在于组件的种类和数量。

拆解 Stable Diffusion 管线：多组件条件扩散模型

Stable Diffusion 是 text-to-image 的latent diffusion（潜在扩散）模型。它不是在真实像素空间，而是在图像的低维潜在表示上工作，因而内存效率更高。其组件结构为：

• 编码器（VAE Encoder）：把图像压缩成更小的潜在表示；
• 解码器（VAE Decoder）：把压缩表示还原成图像；
• Tokenizer + 文本编码器：为 text-to-image 生成文本嵌入（text embeddings）；
• UNet 模型与调度器：在潜在空间中执行去噪。

相比只有 UNet 的 DDPM 管线，Stable Diffusion 包含三个独立的预训练模型：VAE、文本编码器与 UNet。加载方法相同——用from_pretrained从预训练检查点的各子文件夹（subfolder）中分别加载：

>>> from PIL import Image >>> import torch >>> from transformers import CLIPTextModel, CLIPTokenizer >>> from diffusers import AutoencoderKL, UNet2DConditionModel, PNDMScheduler >>> vae = AutoencoderKL.from_pretrained("CompVis/stable-diffusion-v1-4", subfolder="vae") >>> tokenizer = CLIPTokenizer.from_pretrained("CompVis/stable-diffusion-v1-4", subfolder="tokenizer") >>> text_encoder = CLIPTextModel.from_pretrained("CompVis/stable-diffusion-v1-4", subfolder="text_encoder") >>> unet = UNet2DConditionModel.from_pretrained("CompVis/stable-diffusion-v1-4", subfolder="unet")

调度器可以随意替换：把默认的PNDMScheduler换成UniPCMultistepScheduler，只需一行代码，体现了调度器"即插即用"的设计：

>>> from diffusers import UniPCMultistepScheduler >>> scheduler = UniPCMultistepScheduler.from_pretrained("CompVis/stable-diffusion-v1-4", subfolder="scheduler")

💡 调度器没有可训练权重，可以留在 CPU 上；而模型有训练权重，为了加快推理应显式移到 GPU：

>>> torch_device = "cuda" >>> vae.to(torch_device) >>> text_encoder.to(torch_device) >>> unet.to(torch_device)

生成文本嵌入（Text Embeddings）

文本嵌入用于条件化（condition）UNet，把扩散过程"牵引"向与输入提示词相似的方向。guidance_scale（无分类器引导尺度）决定生成时提示词对结果的影响力大小。

先定义推理参数：

>>> prompt = ["a photograph of an astronaut riding a horse"] >>> height = 512 # Stable Diffusion 默认高度 >>> width = 512 # Stable Diffusion 默认宽度 >>> num_inference_steps = 25 # 去噪步数 >>> guidance_scale = 7.5 # classifier-free guidance 的尺度 >>> generator = torch.manual_seed(0) # 用于生成初始潜在噪声的随机种子 >>> batch_size = len(prompt)

对文本做 tokenize 并编码为嵌入：

>>> text_input = tokenizer( ... prompt, padding="max_length", max_length=tokenizer.model_max_length, truncation=True, return_tensors="pt" ... ) >>> with torch.no_grad(): ... text_embeddings = text_encoder(text_input.input_ids.to(torch_device))[0]

还需要生成无条件文本嵌入（unconditional text embeddings），即空提示（padding token）的嵌入。它的形状（batch_size与seq_length）必须与条件嵌入一致：

>>> max_length = text_input.input_ids.shape[-1] >>> uncond_input = tokenizer([""] * batch_size, padding="max_length", max_length=max_length, return_tensors="pt") >>> uncond_embeddings = text_encoder(uncond_input.input_ids.to(torch_device))[0]

将条件与无条件嵌入拼接成一个 batch，从而在一次前向中同时计算两者，避免两次前向：

>>> text_embeddings = torch.cat([uncond_embeddings, text_embeddings])

生成随机噪声（初始潜在变量）

扩散过程的起点是随机噪声，也就是图像"潜在表示"的初始形态，它将逐步被去噪。此时潜在图尺寸小于最终图片——这是有意为之，因为 VAE 有 3 个下采样层，高度和宽度被除以 8：

💡 可以运行下面的表达式验证下采样倍数：2 ** (len(vae.config.block_out_channels) - 1) == 8

>>> latents = torch.randn( ... (batch_size, unet.config.in_channels, height // 8, width // 8), ... generator=generator, ... device=torch_device, ... )

图像去噪（含无分类器引导）

先用初始噪声分布的标准差sigma（即init_noise_sigma）缩放输入。这是UniPCMultistepScheduler这类改进型调度器所必需的（见 scheduling_unipc_multistep.py 中init_noise_sigma的定义；对 DDPM 其值恒为 1.0，见 scheduling_ddpm.py）：

>>> latents = latents * scheduler.init_noise_sigma

去噪循环需要完成三件事：① 设置调度器的 timesteps；② 迭代 timesteps；③ 在每个 timestep 调用 UNet 预测噪声残差，交给调度器计算上一时刻的噪声样本。加上无分类器引导（CFG）后完整代码如下：

>>> from tqdm.auto import tqdm >>> scheduler.set_timesteps(num_inference_steps) >>> for t in tqdm(scheduler.timesteps): ... # 做 classifier-free guidance 时把 latents 复制一份，避免两次前向 ... latent_model_input = torch.cat([latents] * 2) ... latent_model_input = scheduler.scale_model_input(latent_model_input, timestep=t) ... # 预测噪声残差 ... with torch.no_grad(): ... noise_pred = unet(latent_model_input, t, encoder_hidden_states=text_embeddings).sample ... # 执行 guidance ... noise_pred_uncond, noise_pred_text = noise_pred.chunk(2) ... noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond) ... # 计算上一时刻噪声样本 x_t -> x_t-1 ... latents = scheduler.step(noise_pred, t, latents).prev_sample

这段循环中有三个值得留意的点：

• torch.cat([latents] * 2)把潜在变量复制成两份，分别对应无条件分支与条件分支，与前面拼接后的text_embeddings（同样两份）在 batch 维度对齐；
• scheduler.scale_model_input用于兼容那些需要按当前 timestep 缩放模型输入的调度器；对DDPMScheduler（scheduling_ddpm.py）与UniPCMultistepScheduler（scheduling_unipc_multistep.py）而言，缩放因子为 1，即原样返回，接口的存在是为了保证调度器之间的互换性；
• CFG 公式noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)是经典的无分类器引导形式，guidance_scale越大，图像与提示词的贴合度越高。

用 VAE 解码图像

最后一步：把潜在表示从 VAE 的缩放系数还原，再交给vae.decode解码成图像：

# 缩放 latents 并用 vae 解码 latents = 1 / 0.18215 * latents with torch.no_grad(): image = vae.decode(latents).sample

💡 这里的0.18215是 Stable Diffusion VAE 的默认scaling_factor（缩放因子），可以在 pipeline_stable_diffusion.py 中看到官方实现同样使用1 / self.vae.config.scaling_factor * latents的写法。该默认值定义在 convert_from_ckpt.py 中。

最后把张量转成PIL.Image查看生成结果：

>>> image = (image / 2 + 0.5).clamp(0, 1) >>> image = image.detach().cpu().permute(0, 2, 3, 1).numpy() >>> images = (image * 255).round().astype("uint8") >>> pil_images = [Image.fromarray(image) for image in images] >>> pil_images[0]

总结：你只需要一个去噪循环

从最简单的 DDPM 到复杂的 Stable Diffusion，可以发现：编写自己的扩散系统，全部核心就是一个去噪循环。这个循环需要：

1. 设置调度器的 timesteps；
2. 迭代 timesteps；
3. 交替执行——调用 UNet 模型预测噪声残差，把残差交给调度器计算上一时刻的噪声样本。

这正是 🧨 Diffusers 的设计初衷：让开发者能够基于模型与调度器，直观、轻松地搭建属于自己的扩散系统。

如果想要继续深入，可以尝试：

• 阅读 src/diffusers/pipelines/ddpm/pipeline_ddpm.py 与 src/diffusers/pipelines/stable_diffusion/pipeline_stable_diffusion.py，对照官方实现检查自己手写的循环是否正确；
• 浏览 src/diffusers/schedulers 目录下的全部调度器（DDPM、PNDM、UniPC、DDIM、DPM-Solver 等），尝试把同一个去噪循环中的调度器逐个替换，体会"换调度器如换插件"的设计哲学；
• 在 src/diffusers/models/unets 中研究不同 UNet 变体（UNet2DModel与UNet2DConditionModel的输入差异），理解条件扩散模型如何通过encoder_hidden_states接收文本条件；
• 参考官方文档 docs/source/ko/using-diffusers 目录下的其他教程，进一步了解管线构建与社区贡献的完整流程。

【免费下载链接】diffusers🤗 Diffusers: State-of-the-art diffusion models for image, video, and audio generation in PyTorch.项目地址: https://gitcode.com/GitHub_Trending/di/diffusers