扩散模型与多模态：AI 如何生成图像、看懂图文音

# 简化版：Stable Diffusion 式文生图推理流程

def text_to_image(prompt, num_steps=30):
    # 1. 把文字提示编码成向量，作为去噪过程的"导航仪"
    text_embedding = text_encoder.encode(prompt)

    # 2. 在潜在空间里生成一个纯随机噪声，作为起点
    latent = random_noise(shape=(4, 64, 64))

    # 3. 反复调用去噪网络，一步步减少噪声
    for step in reversed(range(num_steps)):
        noise_level = schedule[step]          # 当前这一步理论上的噪声强度
        predicted_noise = denoise_unet(
            latent,
            timestep=step,
            condition=text_embedding           # 交叉注意力：让文字指挥去噪方向
        )
        # 用预测出的噪声，把 latent 往"更干净"的方向推进一点
        latent = remove_noise(latent, predicted_noise, noise_level)

    # 4. 把去噪完成的潜在表示，用解码器还原成像素图片
    image = vae_decoder.decode(latent)
    return image

最终预测噪声 = 无条件预测噪声 + CFG强度 × (有条件预测噪声 - 无条件预测噪声)

# 简化版：多模态大模型"看图说话"推理流程

def multimodal_chat(image, question):
    # 1. 图像分块并编码：图片 -> 一串视觉特征向量
    patches = split_into_patches(image, patch_size=16)
    visual_features = vision_encoder(patches)      # 类似 CLIP 的图像塔

    # 2. 投影层：把视觉向量翻译成语言模型的"词向量空间"
    visual_tokens = projection_layer(visual_features)

    # 3. 文本 token 化
    text_tokens = tokenizer.encode(question)

    # 4. 拼接视觉 token 与文本 token，一起喂给语言模型
    input_tokens = concat(visual_tokens, text_tokens)
    answer = llm.generate(input_tokens)

    return answer

q(x_t | x_{t-1}) = N(x_t; sqrt(1 - beta_t) * x_{t-1}, beta_t * I)

q(x_t | x_0) = N(x_t; sqrt(alpha_bar_t) * x_0, (1 - alpha_bar_t) * I)

等价的重参数化采样写法（epsilon 是标准正态噪声）：
x_t = sqrt(alpha_bar_t) * x_0 + sqrt(1 - alpha_bar_t) * epsilon

import torch
import math

def linear_beta_schedule(timesteps, beta_start=1e-4, beta_end=2e-2):
    return torch.linspace(beta_start, beta_end, timesteps)

def cosine_beta_schedule(timesteps, s=0.008):
    # improved DDPM 的 cosine 调度
    steps = timesteps + 1
    x = torch.linspace(0, timesteps, steps)
    alphas_bar = torch.cos(((x / timesteps) + s) / (1 + s) * math.pi * 0.5) ** 2
    alphas_bar = alphas_bar / alphas_bar[0]
    betas = 1 - (alphas_bar[1:] / alphas_bar[:-1])
    return torch.clip(betas, 0.0001, 0.9999)

T = 1000
betas = cosine_beta_schedule(T)
alphas = 1.0 - betas
alphas_bar = torch.cumprod(alphas, dim=0)   # alpha_bar_t 累乘
sqrt_alphas_bar = torch.sqrt(alphas_bar)
sqrt_one_minus_alphas_bar = torch.sqrt(1.0 - alphas_bar)

L_simple = E_{x_0, t, epsilon} [ || epsilon - epsilon_theta(x_t, t) ||^2 ]

其中 x_t = sqrt(alpha_bar_t) * x_0 + sqrt(1 - alpha_bar_t) * epsilon

import torch
import torch.nn.functional as F

def q_sample(x_0, t, noise, sqrt_ab, sqrt_1m_ab):
    """前向加噪：一步得到 x_t（利用闭式解）"""
    # t 是一个 batch 的时间步索引，用 gather 取出对应系数
    a = sqrt_ab[t].view(-1, 1, 1, 1)
    b = sqrt_1m_ab[t].view(-1, 1, 1, 1)
    return a * x_0 + b * noise

def train_step(model, x_0, optimizer, T, sqrt_ab, sqrt_1m_ab, device):
    model.train()
    batch = x_0.shape[0]
    # 1. 每张图随机采一个 timestep
    t = torch.randint(0, T, (batch,), device=device).long()
    # 2. 采一团标准正态噪声
    noise = torch.randn_like(x_0)
    # 3. 前向：一步造出加噪图 x_t
    x_t = q_sample(x_0, t, noise, sqrt_ab, sqrt_1m_ab)
    # 4. 让 U-Net / DiT 预测这团噪声
    predicted_noise = model(x_t, t)
    # 5. 均方误差损失（predict-epsilon 目标）
    loss = F.mse_loss(noise, predicted_noise)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
    return loss.item()

# 训练主循环
for epoch in range(num_epochs):
    for x_0 in dataloader:
        x_0 = x_0.to(device)          # 干净图 batch，范围归一化到 [-1, 1]
        loss = train_step(model, x_0, optimizer, T,
                          sqrt_alphas_bar.to(device),
                          sqrt_one_minus_alphas_bar.to(device), device)

x_{t-1} = sqrt(alpha_bar_{t-1}) * x0_pred
          + sqrt(1 - alpha_bar_{t-1}) * epsilon_theta(x_t, t)

其中 x0_pred = (x_t - sqrt(1 - alpha_bar_t) * epsilon_theta) / sqrt(alpha_bar_t)

# 潜在空间与像素空间的往返（diffusers 的 AutoencoderKL）
from diffusers import AutoencoderKL
import torch

vae = AutoencoderKL.from_pretrained(
    "stabilityai/sd-vae-ft-mse", torch_dtype=torch.float16
).to("cuda")

# 编码：像素图 -> 潜在表示（注意 SD 的缩放因子 0.18215）
image = torch.randn(1, 3, 512, 512, device="cuda", dtype=torch.float16)
with torch.no_grad():
    latent = vae.encode(image).latent_dist.sample() * 0.18215
print(latent.shape)   # torch.Size([1, 4, 64, 64])

# 解码：潜在表示 -> 像素图
with torch.no_grad():
    recon = vae.decode(latent / 0.18215).sample
print(recon.shape)    # torch.Size([1, 3, 512, 512])

from diffusers import StableDiffusionControlNetPipeline, ControlNetModel
import torch, cv2, numpy as np
from PIL import Image

# 加载 Canny 边缘的 ControlNet
controlnet = ControlNetModel.from_pretrained(
    "lllyasviel/sd-controlnet-canny", torch_dtype=torch.float16
)
pipe = StableDiffusionControlNetPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    controlnet=controlnet, torch_dtype=torch.float16
).to("cuda")

# 从原图提取 Canny 边缘作为结构条件
src = np.array(Image.open("pose_ref.png").convert("RGB"))
edges = cv2.Canny(src, 100, 200)
edges = np.stack([edges] * 3, axis=-1)
control_image = Image.fromarray(edges)

image = pipe(
    prompt="a cyberpunk knight, neon light, ultra detailed",
    image=control_image,
    controlnet_conditioning_scale=0.8,   # 结构约束强度
    num_inference_steps=25,
).images[0]
image.save("controlnet_out.png")

# diffusers 里加载并叠加多个 LoRA
pipe.load_lora_weights("path/to/style_lora.safetensors", adapter_name="style")
pipe.load_lora_weights("path/to/character_lora.safetensors", adapter_name="char")
pipe.set_adapters(["style", "char"], adapter_weights=[0.8, 0.6])  # 各自权重
image = pipe("a portrait in the trained style", num_inference_steps=25).images[0]

from diffusers import StableDiffusionInpaintPipeline
import torch
from PIL import Image

pipe = StableDiffusionInpaintPipeline.from_pretrained(
    "runwayml/stable-diffusion-inpainting", torch_dtype=torch.float16
).to("cuda")

init_image = Image.open("room.png").convert("RGB").resize((512, 512))
mask = Image.open("mask.png").convert("RGB").resize((512, 512))  # 白色=重绘区

result = pipe(
    prompt="a vase of fresh flowers on the table",
    image=init_image,
    mask_image=mask,
    strength=0.9,               # 重绘强度
    num_inference_steps=30,
).images[0]
result.save("inpaint_out.png")

# 用 LCM-LoRA 让任意 SDXL 模型秒级出图
from diffusers import StableDiffusionXLPipeline, LCMScheduler
import torch

pipe = StableDiffusionXLPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0", torch_dtype=torch.float16
).to("cuda")
pipe.load_lora_weights("latent-consistency/lcm-lora-sdxl")
pipe.scheduler = LCMScheduler.from_config(pipe.scheduler.config)

image = pipe(
    prompt="astronaut riding a horse on mars, cinematic",
    num_inference_steps=4,     # 仅 4 步
    guidance_scale=1.5,        # LCM 需要很低的 CFG
).images[0]

# 伪代码：时空 patch 化 + 时序注意力的思想
def spacetime_patchify(video_latent, ps=2, pt=1):
    # video_latent: [B, C, T, H, W]  (时间 T，空间 H×W)
    # 在 T、H、W 三个维度上切 patch，展平成 token 序列
    B, C, T, H, W = video_latent.shape
    tokens = rearrange(
        video_latent,
        "b c (t pt) (h ps1) (w ps2) -> b (t h w) (c pt ps1 ps2)",
        pt=pt, ps1=ps, ps2=ps,
    )
    return tokens   # [B, num_tokens, dim]

class SpatioTemporalBlock(nn.Module):
    def forward(self, x, t_emb, text_emb):
        # 1. 空间注意力：同一帧内 patch 互相看
        x = x + self.spatial_attn(self.norm1(x))
        # 2. 时序注意力：跨帧同位置 patch 互相看，保证连贯不闪烁
        x = x + self.temporal_attn(self.norm2(x))
        # 3. 交叉注意力：注入文本条件
        x = x + self.cross_attn(self.norm3(x), text_emb)
        x = x + self.mlp(self.norm4(x))
        return x

from diffusers import CogVideoXPipeline
from diffusers.utils import export_to_video
import torch

pipe = CogVideoXPipeline.from_pretrained(
    "THUDM/CogVideoX-5b", torch_dtype=torch.bfloat16
).to("cuda")
pipe.enable_model_cpu_offload()   # 显存不够时分层卸载到内存

video = pipe(
    prompt="A panda playing guitar by a lake at sunset, cinematic lighting",
    num_frames=49,
    num_inference_steps=50,
    guidance_scale=6.0,
).frames[0]
export_to_video(video, "panda.mp4", fps=8)

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

def clip_infonce_loss(image_features, text_features, logit_scale):
    # image_features / text_features: [N, D]
    # 1. L2 归一化到单位球面
    image_features = F.normalize(image_features, dim=-1)
    text_features = F.normalize(text_features, dim=-1)

    # 2. 相似度矩阵，乘上可学习温度 logit_scale = exp(t)
    logits_per_image = logit_scale * image_features @ text_features.t()  # [N, N]
    logits_per_text = logits_per_image.t()

    # 3. 对角线是正确配对，标签就是 0..N-1
    labels = torch.arange(image_features.shape[0], device=image_features.device)

    # 4. 图->文 与 文->图 两个方向的交叉熵取平均
    loss_i = F.cross_entropy(logits_per_image, labels)
    loss_t = F.cross_entropy(logits_per_text, labels)
    return (loss_i + loss_t) / 2

class PatchEmbed(nn.Module):
    def __init__(self, img_size=224, patch_size=16, in_ch=3, embed_dim=768):
        super().__init__()
        self.num_patches = (img_size // patch_size) ** 2   # 14*14 = 196
        # 用一个 stride=patch 的卷积一步完成"切块+线性投影"
        self.proj = nn.Conv2d(in_ch, embed_dim,
                              kernel_size=patch_size, stride=patch_size)
        self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim))
        self.pos_embed = nn.Parameter(
            torch.zeros(1, self.num_patches + 1, embed_dim))

    def forward(self, x):
        B = x.shape[0]
        x = self.proj(x)                    # [B, 768, 14, 14]
        x = x.flatten(2).transpose(1, 2)    # [B, 196, 768]  每块变一个 token
        cls = self.cls_token.expand(B, -1, -1)
        x = torch.cat([cls, x], dim=1)      # 拼上 [CLS] token -> [B, 197, 768]
        x = x + self.pos_embed              # 加位置编码
        return x

import torch.nn as nn

class VisionProjector(nn.Module):
    """把 CLIP 视觉特征投影到 LLM 词向量空间（LLaVA-1.5 式两层 MLP）"""
    def __init__(self, vision_dim=1024, llm_dim=4096):
        super().__init__()
        self.proj = nn.Sequential(
            nn.Linear(vision_dim, llm_dim),
            nn.GELU(),
            nn.Linear(llm_dim, llm_dim),
        )

    def forward(self, image_features):
        # image_features: [B, num_patches, vision_dim]
        return self.proj(image_features)   # [B, num_patches, llm_dim]

# 前向拼接：视觉 token 直接当成一段"外语词"插进文本序列
# input_embeds = [text_emb_前, projected_visual_tokens, text_emb_后]

{
  "image": "coco/000000123.jpg",
  "conversations": [
    {"from": "human", "value": "<image>\n这张图里有几个人，他们在做什么？"},
    {"from": "gpt", "value": "图中有三个人，他们围坐在野餐垫上分享食物，背景是公园。"}
  ]
}

# Whisper 做语音识别（transformers）
from transformers import pipeline
import torch

asr = pipeline(
    "automatic-speech-recognition",
    model="openai/whisper-large-v3",
    torch_dtype=torch.float16,
    device="cuda",
)
result = asr("meeting.wav", generate_kwargs={"language": "chinese"})
print(result["text"])

# EnCodec 把波形压成离散 code，再还原（音频"tokenize"）
from transformers import EncodecModel, AutoProcessor
import torch

model = EncodecModel.from_pretrained("facebook/encodec_24khz")
processor = AutoProcessor.from_pretrained("facebook/encodec_24khz")

inputs = processor(raw_audio=wav_array, sampling_rate=24000, return_tensors="pt")
encoded = model.encode(inputs["input_values"], inputs["padding_mask"])
codes = encoded.audio_codes         # 离散音频 token，形如 [B, n_codebooks, T]
# 语音到语音：LLM 在这些 code 上做自回归，再用 model.decode 还原成波形
audio_out = model.decode(encoded.audio_codes, encoded.audio_scales,
                         inputs["padding_mask"])[0]

from diffusers import StableDiffusionXLPipeline
import torch

pipe = StableDiffusionXLPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    torch_dtype=torch.float16, variant="fp16", use_safetensors=True,
).to("cuda")

image = pipe(
    prompt="a majestic snow leopard on a cliff, golden hour, 85mm photo",
    negative_prompt="blurry, lowres, deformed, extra limbs, watermark",
    num_inference_steps=30,
    guidance_scale=7.0,
    generator=torch.Generator("cuda").manual_seed(42),   # 固定种子可复现
).images[0]
image.save("leopard.png")

from diffusers import StableDiffusionXLImg2ImgPipeline
from diffusers.utils import load_image
import torch

pipe = StableDiffusionXLImg2ImgPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0", torch_dtype=torch.float16
).to("cuda")

init = load_image("sketch.png").resize((1024, 1024))
image = pipe(
    prompt="oil painting, vivid colors, impressionist style",
    image=init,
    strength=0.6,          # 0 完全保留原图, 1 几乎重画; 越大越自由
    num_inference_steps=30,
    guidance_scale=7.5,
).images[0]
image.save("img2img_out.png")

from transformers import AutoProcessor, LlavaForConditionalGeneration
from PIL import Image
import torch

model = LlavaForConditionalGeneration.from_pretrained(
    "llava-hf/llava-1.5-7b-hf", torch_dtype=torch.float16, device_map="cuda"
)
processor = AutoProcessor.from_pretrained("llava-hf/llava-1.5-7b-hf")

image = Image.open("chart.png")
prompt = "USER: <image>\n这张财报截图里，营收同比增长了多少？ ASSISTANT:"
inputs = processor(text=prompt, images=image, return_tensors="pt").to("cuda", torch.float16)

out = model.generate(**inputs, max_new_tokens=200)
print(processor.decode(out[0], skip_special_tokens=True))

from transformers import CLIPProcessor, CLIPModel
from PIL import Image
import torch

model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

images = [Image.open(p) for p in ["cat.jpg", "dog.jpg", "car.jpg"]]
texts = ["a photo of a cat", "a photo of a dog", "a photo of a car"]

inputs = processor(text=texts, images=images, return_tensors="pt", padding=True)
with torch.no_grad():
    outputs = model(**inputs)

# logits_per_image[i][j] = 第 i 张图与第 j 段文字的相似度
probs = outputs.logits_per_image.softmax(dim=1)
print(probs)   # 对角线应最高，说明图文正确匹配