implement-diffusion-network
关于
This skill implements a complete diffusion model from scratch, including noise scheduling, a U-Net architecture, and training/sampling loops with DDIM acceleration. It's designed for prototyping custom diffusion models for image, audio, or molecular synthesis before scaling to production frameworks. Use it when you need to implement a paper, add custom conditioning, or replace a GAN with a diffusion-based generator.
快速安装
Claude Code
推荐npx skills add pjt222/agent-almanac -a claude-code/plugin add https://github.com/pjt222/agent-almanacgit clone https://github.com/pjt222/agent-almanac.git ~/.claude/skills/implement-diffusion-network在 Claude Code 中复制并粘贴此命令以安装该技能
技能文档
實擴散網
從頭建去噪擴散概率模(DDPM)或分數生成模,含正向加噪、U-Net 去噪器、訓目標、反向採樣、經 DDIM 或 DPM-Solver 加速推理。
用
- 建像、音、分子之生成模
- 由論文實 DDPM 或分數擴散
- 加自定噪表或條件機於擴散管線
- 以擴散替 GAN 生成器
- 擴散模原型,後以 diffusers 等生產框擴
入
- 必:訓數據集(像、譜圖、點雲或他續數據)
- 必:目標分辨率與通道數
- 必:算預算(GPU 型數、訓時限)
- 可:噪表型(默 cosine)
- 可:擴散時步 T(默 1000)
- 可:條件信號(類標、文嵌或他導)
- 可:採樣加速法(默 DDIM 50 步)
行
一:定正向程(噪表)
配控數據漸噪之變量表。
- 定 beta 表(線、cosine 或學):
import torch
import numpy as np
def cosine_beta_schedule(timesteps, s=0.008):
"""Cosine schedule from Nichol & Dhariwal (2021)."""
steps = timesteps + 1
t = torch.linspace(0, timesteps, steps) / timesteps
alphas_cumprod = torch.cos((t + s) / (1 + s) * np.pi / 2) ** 2
alphas_cumprod = alphas_cumprod / alphas_cumprod[0]
betas = 1 - (alphas_cumprod[1:] / alphas_cumprod[:-1])
return torch.clip(betas, 0.0001, 0.9999)
def linear_beta_schedule(timesteps, beta_start=1e-4, beta_end=0.02):
"""Original DDPM linear schedule."""
return torch.linspace(beta_start, beta_end, timesteps)
- 預算訓與採用之衍量:
class DiffusionSchedule:
def __init__(self, betas):
self.betas = betas
self.alphas = 1.0 - betas
self.alphas_cumprod = torch.cumprod(self.alphas, dim=0)
self.alphas_cumprod_prev = torch.cat([torch.tensor([1.0]), self.alphas_cumprod[:-1]])
self.sqrt_alphas_cumprod = torch.sqrt(self.alphas_cumprod)
self.sqrt_one_minus_alphas_cumprod = torch.sqrt(1.0 - self.alphas_cumprod)
self.posterior_variance = (
betas * (1.0 - self.alphas_cumprod_prev) / (1.0 - self.alphas_cumprod)
)
- 實正向加噪函(q-sample):
def q_sample(self, x_0, t, noise=None):
"""Add noise to x_0 at timestep t: q(x_t | x_0)."""
if noise is None:
noise = torch.randn_like(x_0)
sqrt_alpha = self.sqrt_alphas_cumprod[t].reshape(-1, 1, 1, 1)
sqrt_one_minus_alpha = self.sqrt_one_minus_alphas_cumprod[t].reshape(-1, 1, 1, 1)
return sqrt_alpha * x_0 + sqrt_one_minus_alpha * noise
- 視察表:
schedule = DiffusionSchedule(cosine_beta_schedule(1000))
print(f"alpha_cumprod at t=0: {schedule.alphas_cumprod[0]:.4f}") # ~1.0 (clean)
print(f"alpha_cumprod at t=500: {schedule.alphas_cumprod[500]:.4f}") # ~0.5 (half noise)
print(f"alpha_cumprod at t=999: {schedule.alphas_cumprod[999]:.4f}") # ~0.0 (pure noise)
得:alphas_cumprod 由近 1.0 單降至近 0.0。cosine 表於中時步較線性更緩降。
敗:alphas_cumprod 於 t=T 未至近零→模不能由純噪生。增 T 或調表。若為負→察 betas 之剪邊。
二:設去噪網架構
建具時條件之 U-Net,於噪入予下預噪。
- 定時嵌模:
import torch.nn as nn
import math
class SinusoidalTimeEmbedding(nn.Module):
def __init__(self, dim):
super().__init__()
self.dim = dim
def forward(self, t):
half_dim = self.dim // 2
emb = math.log(10000) / (half_dim - 1)
emb = torch.exp(torch.arange(half_dim, device=t.device) * -emb)
emb = t[:, None].float() * emb[None, :]
return torch.cat([emb.sin(), emb.cos()], dim=-1)
- 定具時條件之殘差塊:
class ResBlock(nn.Module):
def __init__(self, in_ch, out_ch, time_dim):
super().__init__()
self.conv1 = nn.Conv2d(in_ch, out_ch, 3, padding=1)
self.conv2 = nn.Conv2d(out_ch, out_ch, 3, padding=1)
self.time_mlp = nn.Linear(time_dim, out_ch)
self.norm1 = nn.GroupNorm(8, out_ch)
self.norm2 = nn.GroupNorm(8, out_ch)
self.skip = nn.Conv2d(in_ch, out_ch, 1) if in_ch != out_ch else nn.Identity()
def forward(self, x, t_emb):
h = self.norm1(torch.nn.functional.silu(self.conv1(x)))
h = h + self.time_mlp(torch.nn.functional.silu(t_emb))[:, :, None, None]
h = self.norm2(torch.nn.functional.silu(self.conv2(h)))
return h + self.skip(x)
- 組 U-Net 含編、瓶、解三部:
class UNet(nn.Module):
def __init__(self, in_channels=3, base_channels=64, channel_mults=(1, 2, 4, 8)):
super().__init__()
time_dim = base_channels * 4
self.time_embed = nn.Sequential(
SinusoidalTimeEmbedding(base_channels),
nn.Linear(base_channels, time_dim),
nn.SiLU(),
nn.Linear(time_dim, time_dim)
)
# Encoder, bottleneck, and decoder built from ResBlocks
# with skip connections between encoder and decoder stages
# (full implementation depends on resolution and channel config)
- 驗架構接目標分辨率之入:
model = UNet(in_channels=3, base_channels=64)
x_test = torch.randn(2, 3, 64, 64)
t_test = torch.randint(0, 1000, (2,))
out = model(x_test, t_test)
assert out.shape == x_test.shape, f"Output shape {out.shape} != input shape {x_test.shape}"
print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")
得:模出與入同形(預相符維之噪)。參數數應與分辨率成比:64x64 約 30-60M,256x256 約 100-300M。
敗:形失配常示下/上採樣比誤。驗各編階半空維,各解階雙。GroupNorm 需通道為群數之倍。
三:實訓迴
訓去噪器於各時步預所加噪。
- 立訓目標(簡 DDPM 損):
def training_loss(model, schedule, x_0):
batch_size = x_0.shape[0]
t = torch.randint(0, len(schedule.betas), (batch_size,), device=x_0.device)
noise = torch.randn_like(x_0)
x_t = schedule.q_sample(x_0, t, noise)
predicted_noise = model(x_t, t)
loss = torch.nn.functional.mse_loss(predicted_noise, noise)
return loss
- 配優化器與學率表:
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=0.01)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100000)
- 行訓迴並記:
from torch.utils.data import DataLoader
dataloader = DataLoader(dataset, batch_size=64, shuffle=True, num_workers=4, pin_memory=True)
for epoch in range(num_epochs):
model.train()
epoch_loss = 0.0
for batch_idx, x_0 in enumerate(dataloader):
x_0 = x_0.to(device)
loss = training_loss(model, schedule, x_0)
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
scheduler.step()
epoch_loss += loss.item()
avg_loss = epoch_loss / len(dataloader)
print(f"Epoch {epoch}: loss={avg_loss:.4f}, lr={scheduler.get_last_lr()[0]:.6f}")
- 定期存檢點:
if (epoch + 1) % 10 == 0:
torch.save({
"epoch": epoch,
"model_state": model.state_dict(),
"optimizer_state": optimizer.state_dict(),
"loss": avg_loss
}, f"checkpoint_epoch_{epoch+1}.pt")
得:損隨訓穩降。像數據規一化至 [-1, 1] 時,初損應近 1.0(預隨噪)。收斂後損依數據複雜性於 0.01-0.10 範。
敗:損早平(>0.5)→察:(a) 數據規一化(必 [-1, 1] 或 [0, 1] 配合末激活);(b) 學率(試 3e-4 或 5e-5);(c) 梯剪(1.0 標準)。損為 NaN→減學率並察表中除以零。
四:實採樣(反向程)
由純高斯噪迭代去噪以生新樣。
- 實標 DDPM 採樣迴:
@torch.no_grad()
def ddpm_sample(model, schedule, shape, device):
"""Sample via the full DDPM reverse process (T steps)."""
x = torch.randn(shape, device=device)
T = len(schedule.betas)
for t in reversed(range(T)):
t_batch = torch.full((shape[0],), t, device=device, dtype=torch.long)
predicted_noise = model(x, t_batch)
alpha = schedule.alphas[t]
alpha_cumprod = schedule.alphas_cumprod[t]
beta = schedule.betas[t]
mean = (1 / torch.sqrt(alpha)) * (
x - (beta / torch.sqrt(1 - alpha_cumprod)) * predicted_noise
)
if t > 0:
noise = torch.randn_like(x)
sigma = torch.sqrt(schedule.posterior_variance[t])
x = mean + sigma * noise
else:
x = mean
return x
- 生並視樣:
samples = ddpm_sample(model, schedule, shape=(16, 3, 64, 64), device=device)
samples = (samples.clamp(-1, 1) + 1) / 2 # rescale to [0, 1]
得:生樣示可識構(非純噪或均色)。64x64 分辨率 100K+ 訓步後,出應視似訓分佈。
敗:樣模糊→訓久或增容量。樣噪→反向程或有蟲;驗表索引合訓。諸樣皆同→察模崩(試異隨種)。
五:加採樣加速
用 DDIM 或 DPM-Solver 減採樣步數。
- 實 DDIM 採樣(確定、少步):
@torch.no_grad()
def ddim_sample(model, schedule, shape, device, num_steps=50, eta=0.0):
"""DDIM sampling with configurable step count and stochasticity."""
T = len(schedule.betas)
step_indices = torch.linspace(0, T - 1, num_steps, dtype=torch.long)
x = torch.randn(shape, device=device)
for i in reversed(range(len(step_indices))):
t = step_indices[i]
t_batch = torch.full((shape[0],), t, device=device, dtype=torch.long)
predicted_noise = model(x, t_batch)
alpha_t = schedule.alphas_cumprod[t]
alpha_prev = schedule.alphas_cumprod[step_indices[i - 1]] if i > 0 else torch.tensor(1.0)
predicted_x0 = (x - torch.sqrt(1 - alpha_t) * predicted_noise) / torch.sqrt(alpha_t)
predicted_x0 = predicted_x0.clamp(-1, 1)
sigma = eta * torch.sqrt((1 - alpha_prev) / (1 - alpha_t) * (1 - alpha_t / alpha_prev))
direction = torch.sqrt(1 - alpha_prev - sigma**2) * predicted_noise
x = torch.sqrt(alpha_prev) * predicted_x0 + direction
if i > 0 and eta > 0:
x = x + sigma * torch.randn_like(x)
return x
- 比諸步數之樣品質:
for n_steps in [10, 25, 50, 100, 250]:
samples = ddim_sample(model, schedule, shape=(16, 3, 64, 64), device=device, num_steps=n_steps)
print(f"DDIM {n_steps} steps: generated {samples.shape[0]} samples")
# Save grid for visual comparison
- 測採樣速:
import time
for method, n_steps in [("DDPM", 1000), ("DDIM-50", 50), ("DDIM-25", 25)]:
start = time.time()
_ = ddim_sample(model, schedule, (1, 3, 64, 64), device, num_steps=n_steps if "DDIM" in method else 1000)
elapsed = time.time() - start
print(f"{method}: {elapsed:.2f}s per sample")
得:DDIM 50 步生視似 DDPM 1000 步之樣,速 20 倍。質降至約 20-25 步仍緩。
敗:同步數下 DDIM 劣於 DDPM→驗 alpha 索引。DDIM 直用 alphas_cumprod,非 alphas。低步數樣極噪→先試 eta=0.0(全確定)。
六:評樣品質
用標準指量生成質。
- 算 FID(Frechet Inception Distance):
from torchmetrics.image.fid import FrechetInceptionDistance
fid_metric = FrechetInceptionDistance(feature=2048, normalize=True)
# Add real images
for batch in real_dataloader:
fid_metric.update(batch.to(device), real=True)
# Add generated images
n_generated = 0
while n_generated < 10000:
samples = ddim_sample(model, schedule, (64, 3, 64, 64), device, num_steps=50)
samples = ((samples.clamp(-1, 1) + 1) / 2 * 255).byte()
fid_metric.update(samples, real=False)
n_generated += samples.shape[0]
fid_score = fid_metric.compute()
print(f"FID: {fid_score:.2f}")
- 評樣多樣(察模崩):
# Compute pairwise LPIPS distances among generated samples
from torchmetrics.image.lpip import LearnedPerceptualImagePatchSimilarity
lpips = LearnedPerceptualImagePatchSimilarity(net_type="alex")
n_pairs = 50
diversity_scores = []
for i in range(n_pairs):
s1 = ddim_sample(model, schedule, (1, 3, 64, 64), device, num_steps=50)
s2 = ddim_sample(model, schedule, (1, 3, 64, 64), device, num_steps=50)
score = lpips(s1.clamp(-1, 1), s2.clamp(-1, 1))
diversity_scores.append(score.item())
print(f"Mean pairwise LPIPS: {np.mean(diversity_scores):.4f} (higher = more diverse)")
- 記果:
results = {
"fid": fid_score.item(),
"mean_lpips_diversity": float(np.mean(diversity_scores)),
"sampling_method": "DDIM-50",
"training_epochs": num_epochs,
"model_params": sum(p.numel() for p in model.parameters())
}
print("Evaluation results:", results)
得:標基(CIFAR-10、CelebA)訓佳模 FID 於 50 下。LPIPS 多樣 >0.4 示無模崩。前沿模 CIFAR-10 得 FID 2-10。
敗:FID 高(>100)→訓誤或紀元不足。多樣低(LPIPS <0.2)→模崩;增容量、察數據增強或訓久。FID 至少 10K 樣方穩估。
驗
- 正向程於 t=T 生純噪(視察加數:均近 0、標差近 1)
- U-Net 出形諸目標分辨率皆符入形
- 訓損首 1000 步單降
- DDPM 採樣於充訓後生可識出
- DDIM 50 步生質比 DDPM 1000 步
- 目標數據集 FID 於 50 下(依域調閾)
- 樣多樣(LPIPS)確無模崩
- 檢點已存且可載無誤
忌
- 數據規一化誤:DDPM 假數據於 [-1, 1]。若像於 [0, 255]→損大訓散。訓前規一化,採後反規
- 表索引差一:正向程 t 步噪樣用
alphas_cumprod[t]。採樣中差一誤(用 t+1 或 t-1)→樣顯劣 - 忘梯剪:無
clip_grad_norm_(1.0)大模訓不穩。首紀元尤重 - DDIM 過少步:20 步下 DDIM 質速降。至少 25 步方可;50 步近 DDPM 質
- FID 樣少:FID 於小樣有偏。用至少 10K 生像與 10K 真像方穩
- 略 EMA:模權之指數移均顯提升樣質。用衰 0.9999 並自 EMA 模採,非自訓模
參
analyze-diffusion-dynamicsfit-drift-diffusion-modelsetup-gpu-trainingcontainerize-application
GitHub 仓库
相关推荐技能
content-collections
元Content Collections 是一个 TypeScript 优先的构建工具,可将本地 Markdown/MDX 文件转换为类型安全的数据集合。它专为构建博客、文档站和内容密集型 Vite+React 应用而设计,提供基于 Zod 的自动模式验证。该工具涵盖从 Vite 插件配置、MDX 编译到生产环境部署的完整工作流。
polymarket
元这个Claude Skill为开发者提供完整的Polymarket预测市场开发支持,涵盖API调用、交易执行和市场数据分析。关键特性包括实时WebSocket数据流,可监控实时交易、订单和市场动态。开发者可用它构建预测市场应用、实施交易策略并集成实时市场预测功能。
creating-opencode-plugins
元该Skill帮助开发者创建OpenCode插件,用于接入命令、文件、LSP等25+种事件。它提供了插件结构、事件API规范和JavaScript/TypeScript实现模式,适合需要拦截操作、扩展功能或自定义事件处理的场景。开发者可通过它快速构建响应式模块来增强OpenCode AI助手的能力。
sglang
元SGLang是一个专为LLM设计的高性能推理框架,特别适用于需要结构化输出的场景。它通过RadixAttention前缀缓存技术,在处理JSON、正则表达式、工具调用等具有重复前缀的复杂工作流时,能实现极速生成。如果你正在构建智能体或多轮对话系统,并追求远超vLLM的推理性能,SGLang是理想选择。
