先从加噪公式解出对原图的估计：

$$\hat{\mathbf{x}}_0 =\frac{\mathbf{x}_t- \sqrt{1-\bar{\alpha}_t}\hat{\boldsymbol{\epsilon}}} {\sqrt{\bar{\alpha}_t}}$$

本例把这个估计截到 $[-1,1]$，再计算：

$$\begin{aligned} A_t&=\frac{\sqrt{\bar{\alpha}_{t-1}}\beta_t} {1-\bar{\alpha}_t}\\ B_t&=\frac{\sqrt{\alpha_t}(1-\bar{\alpha}_{t-1})} {1-\bar{\alpha}_t}\\ V_t&=\frac{\beta_t(1-\bar{\alpha}_{t-1})} {1-\bar{\alpha}_t}\\ \mathbf{x}_{t-1} &=A_t\hat{\mathbf{x}}_0+B_t\mathbf{x}_t +\sqrt{V_t}\mathbf{z} \end{aligned}$$

$\mathbf{z}$ 是新抽取的标准正态噪声，$\bar{\alpha}_0=1$。因此 $t=1$ 时 $V_t=0$，最后一步不再添加随机扰动。

这些系数来自已知原图时高斯加噪过程的后验分布；生成时把未知原图替换为网络的估计。这里只使用固定后验方差，没有训练网络去预测方差。对 $\hat{\mathbf{x}}_0$ 的截断也是具体实现选择，不应悄悄从复现条件里省掉。

"""第 42 章：8×8 数字上的自编码、条件 DDPM 与条件流匹配；CPU 从零训练。"""

from argparse import ArgumentParser
from copy import deepcopy
from hashlib import sha256
import json
from math import pi
from pathlib import Path
import platform
from time import perf_counter

import numpy as np
from sklearn.linear_model import LogisticRegression
import torch
from torch import nn

from digits_project import make_data


STEPS = 64
EPOCHS = {"autoencoder": 80, "ddpm": 120, "flow": 120}


def tensor_hash(tensor):
    return sha256(tensor.detach().cpu().contiguous().numpy().tobytes()).hexdigest()


def schedule(steps=STEPS):
    # 余弦 alpha_bar 曲线；限制最后一步 beta，避免除以零。
    grid = torch.arange(steps + 1, dtype=torch.float64) / steps
    curve = torch.cos((grid + 0.008) / 1.008 * pi / 2).square()
    curve = curve / curve[0]
    beta = (1 - curve[1:] / curve[:-1]).clamp(max=0.999).float()
    alpha_bar = torch.cat([torch.ones(1), (1 - beta).cumprod(0)])
    return beta, alpha_bar


def add_noise(clean, noise, level, alpha_bar):
    a = alpha_bar[level, None]
    return a.sqrt() * clean + (1 - a).sqrt() * noise


def flow_pair(clean, noise, time):
    return (1 - time[:, None]) * noise + time[:, None] * clean, clean - noise


class Autoencoder(nn.Module):
    def __init__(self):
        super().__init__()
        self.encoder = nn.Sequential(nn.Linear(64, 32), nn.SiLU(), nn.Linear(32, 8))
        self.decoder = nn.Sequential(nn.Linear(8, 32), nn.SiLU(),
                                     nn.Linear(32, 64), nn.Sigmoid())

    def forward(self, images):
        return self.decoder(self.encoder(images))


class Field(nn.Module):
    def __init__(self):
        super().__init__()
        self.labels = nn.Embedding(10, 16)
        self.network = nn.Sequential(nn.Linear(64 + 9 + 16, 128), nn.SiLU(),
                                     nn.Linear(128, 128), nn.SiLU(),
                                     nn.Linear(128, 64))

    def forward(self, images, time, labels):
        angles = time[:, None] * (2 * pi * 2 ** torch.arange(4))
        clock = torch.cat([time[:, None], angles.sin(), angles.cos()], dim=1)
        return self.network(torch.cat([images, clock, self.labels(labels)], dim=1))


def training_pair(kind, clean, generator, alpha_bar):
    noise = torch.randn(clean.shape, generator=generator)
    if kind == "ddpm":
        level = torch.randint(1, STEPS + 1, (len(clean),), generator=generator)
        return add_noise(clean, noise, level, alpha_bar), level / STEPS, noise
    time = torch.rand(len(clean), generator=generator)
    mixed, velocity = flow_pair(clean, noise, time)
    return mixed, time, velocity


def train(kind, training, validation):
    torch.manual_seed(2026 if kind == "autoencoder" else 2030)
    model = Autoencoder() if kind == "autoencoder" else Field()
    optimizer = torch.optim.Adam(model.parameters(), lr=0.002)
    generator = torch.Generator().manual_seed(3031)
    _, alpha_bar = schedule()
    x, y = training
    vx, vy = validation
    if kind != "autoencoder":
        x, vx = 2 * x - 1, 2 * vx - 1
        validation_pair = training_pair(
            kind, vx, torch.Generator().manual_seed(3030), alpha_bar)
    best, history = float("inf"), []
    for epoch in range(1, EPOCHS[kind] + 1):
        model.train()
        total = 0.0
        for ids in torch.randperm(len(x), generator=generator).split(128):
            optimizer.zero_grad(set_to_none=True)
            if kind == "autoencoder":
                output, target = model(x[ids]), x[ids]
            else:
                mixed, time, target = training_pair(kind, x[ids], generator, alpha_bar)
                output = model(mixed, time, y[ids])
            loss = (output - target).square().mean()
            if not torch.isfinite(loss):
                raise RuntimeError("训练出现非有限损失")
            loss.backward()
            optimizer.step()
            total += loss.item() * len(ids)
        model.eval()
        with torch.no_grad():
            if kind == "autoencoder":
                loss = (model(vx) - vx).square().mean().item()
            else:
                mixed, time, target = validation_pair
                loss = (model(mixed, time, vy) - target).square().mean().item()
        history.append([total / len(x), loss])
        if loss < best:
            best, best_epoch, state = loss, epoch, deepcopy(model.state_dict())
    model.load_state_dict(state)
    model.eval()
    fingerprint = sha256(b"".join(
        value.numpy().tobytes() for value in state.values())).hexdigest()
    return model, {"epoch": best_epoch, "validation_mse": best,
                   "history": history, "parameters": sum(p.numel() for p in model.parameters()),
                   "weights_sha256": fingerprint}


def reverse_step(noisy, epsilon, step, beta, alpha_bar, noise):
    b, current, previous = beta[step - 1], alpha_bar[step], alpha_bar[step - 1]
    clean = ((noisy - (1 - current).sqrt() * epsilon) / current.sqrt()).clamp(-1, 1)
    mean = (previous.sqrt() * b * clean
            + (1 - b).sqrt() * (1 - previous) * noisy) / (1 - current)
    variance = b * (1 - previous) / (1 - current)
    return mean + variance.sqrt() * noise


@torch.no_grad()
def sample(kind, model, labels, steps=32, seed=4040):
    generator = torch.Generator().manual_seed(seed)
    x = torch.randn((len(labels), 64), generator=generator)
    initial_hash = tensor_hash(x)
    snapshots = [x[0].tolist()]
    positions = [0]
    if kind == "ddpm":
        beta, alpha_bar = schedule()
        for step in range(STEPS, 0, -1):
            time = torch.full((len(x),), step / STEPS)
            epsilon = model(x, time, labels)
            noise = torch.randn(x.shape, generator=generator) if step > 1 else torch.zeros_like(x)
            x = reverse_step(x, epsilon, step, beta, alpha_bar, noise)
            if step in [49, 33, 17, 1]:
                snapshots.append(x[0].tolist())
                positions.append(STEPS - step + 1)
    else:
        for step in range(steps):
            time = torch.full((len(x),), step / steps)
            x = x + model(x, time, labels) / steps
            if (step + 1) % (steps // 4) == 0:
                snapshots.append(x[0].tolist())
                positions.append(step + 1)
    return ((x + 1) / 2).clamp(0, 1), {
        "initial_noise_sha256": initial_hash, "snapshots": snapshots,
        "snapshot_steps": positions, "evaluations": STEPS if kind == "ddpm" else steps,
        "raw_outside_fraction": ((x < -1) | (x > 1)).float().mean().item()}


def run():
    started = perf_counter()
    torch.set_num_threads(1)
    torch.use_deterministic_algorithms(True)
    raw, ids = make_data()
    data = [(x.flatten(1).float() / 16, y) for x, y in raw]
    training, validation, test = data
    report = {"split_ids": [index.tolist() for index in ids],
              "environment": {"python": platform.python_version(),
                              "torch": torch.__version__, "numpy": np.__version__,
                              "platform": platform.platform(), "threads": 1},
              "config": {"epochs": EPOCHS, "batch_size": 128, "lr": 0.002,
                         "ddpm_steps": STEPS, "flow_steps": 32,
                         "sample_seed": 4040, "train_seed": 3031,
                         "validation_seed": 3030},
              "source_sha256": sha256(Path(__file__).read_bytes()).hexdigest(),
              "data_sha256": [tensor_hash(x) for x, _ in data]}
    ae, record = train("autoencoder", training, validation)
    with torch.no_grad():
        reconstruction = ae(test[0])
        record["test_mse"] = (reconstruction - test[0]).square().mean().item()
        record["mean_baseline_mse"] = (
            training[0].mean(0) - test[0]).square().mean().item()
        chosen = [int(torch.where(test[1] == label)[0][0]) for label in range(10)]
        record["image_ids"] = ids[2][chosen].tolist()
        record["originals"] = test[0][chosen].tolist()
        record["reconstructions"] = reconstruction[chosen].tolist()
        random_latent = torch.randn((10, 8), generator=torch.Generator().manual_seed(4041))
        record["random_latent_decodes"] = ae.decoder(random_latent).tolist()
    report["autoencoder"] = record
    classifier = LogisticRegression(C=1, max_iter=1000).fit(*[v.numpy() for v in training])
    report["classifier_real_accuracy"] = float(classifier.score(
        test[0].numpy(), test[1].numpy()))
    labels = torch.arange(10).repeat_interleave(10)
    report["sample_labels"] = labels.tolist()
    for kind in ["ddpm", "flow"]:
        model, record = train(kind, training, validation)
        samples, trajectory = sample(kind, model, labels)
        predicted = classifier.predict(samples.numpy())
        record.update(trajectory)
        record.update({"samples": samples.tolist(), "predicted": predicted.tolist(),
                       "condition_correct": int((predicted == labels.numpy()).sum())})
        report[kind] = record
    _, alpha_bar = schedule()
    clean = 2 * test[0][chosen[3]:chosen[3] + 1] - 1
    noise = torch.randn(clean.shape, generator=torch.Generator().manual_seed(4042))
    report["forward"] = {"image_id": int(ids[2][chosen[3]]), "levels": [0, 16, 32, 48, 64],
                         "images": [add_noise(clean, noise, torch.tensor([i]), alpha_bar)[0].tolist()
                                    for i in [0, 16, 32, 48, 64]],
                         "alpha_bar": alpha_bar.tolist()}
    report["elapsed_seconds"] = perf_counter() - started
    return report


def show(report):
    ae = report["autoencoder"]
    print("训练/验证/测试：1077 / 360 / 360；图像：8×8")
    print(f"自编码器：最佳轮次={ae['epoch']}，测试 MSE={ae['test_mse']:.4f}")
    print(f"训练均值图基线 MSE={ae['mean_baseline_mse']:.4f}")
    print(f"代理分类器真实测试准确率={report['classifier_real_accuracy']:.3f}")
    for kind in ["ddpm", "flow"]:
        row = report[kind]
        print(f"{kind}：最佳轮次={row['epoch']}，验证目标 MSE={row['validation_mse']:.4f}，"
              f"条件识别={row['condition_correct']}/100，采样调用={row['evaluations']}")
    print("条件识别只检查生成数字是否被分类器认作指定类别，不代表整体生成质量。")


if __name__ == "__main__":
    parser = ArgumentParser(description=__doc__)
    parser.add_argument("--out", type=Path, help="保存本次完整训练、像素和评价记录")
    args = parser.parse_args()
    result = run()
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n")
    show(result)

"""第 42 章：真实 SD-Turbo 条件生成。默认核对已保存记录；--run 才加载模型。"""

from argparse import ArgumentParser
from hashlib import sha256
import json
from pathlib import Path
import platform
import struct
from time import perf_counter


MODEL = "stabilityai/sd-turbo"
REVISION = "b261bac6fd2cf515557d5d0707481eafa0485ec2"
DEFAULT = Path(__file__).parent / "data" / "sd-turbo" / "report.json"
PROMPT = "A studio photograph of one {} ceramic mug on a plain white table, soft natural light."
CASES = [
    {"id": f"{color}-{seed}-1", "prompt": PROMPT.format(color),
     "seed": seed, "steps": 1}
    for seed in [2026, 2027] for color in ["red", "blue"]
] + [
    {"id": "red-2026-4", "prompt": PROMPT.format("red"), "seed": 2026, "steps": 4},
    {"id": "three-tea-2026-1", "prompt": (
        "A studio photograph of exactly three red ceramic mugs on a white table, "
        "each labeled TEA in clear black letters."), "seed": 2026, "steps": 1},
]


def file_hash(path):
    digest = sha256()
    with Path(path).open("rb") as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def validate(path=DEFAULT):
    report = json.loads(path.read_text())
    assert report["model"] == MODEL and report["revision"] == REVISION
    assert report["source_sha256"] == file_hash(__file__), "记录对应的脚本版本已改变"
    assert report["cases"] == CASES and len(report["outputs"]) == len(CASES)
    for case, row in zip(CASES, report["outputs"]):
        assert all(row[k] == v for k, v in case.items())
        filename = path.parent / row["image"]
        assert filename.name == case["id"] + ".png"
        assert file_hash(filename) == row["image_sha256"]
        header = filename.read_bytes()[:24]
        assert header[:8] == b"\x89PNG\r\n\x1a\n"
        assert struct.unpack(">II", header[16:24]) == (512, 512)
        assert len(row["timesteps"]) == case["steps"] and row["seconds"] > 0
    for seed in [2026, 2027]:
        group = [r for r in report["outputs"] if r["seed"] == seed]
        assert len({r["noise_sha256"] for r in group}) == 1
        assert len({r["initial_latent_sha256"] for r in group}) == 1
    return report


def run(folder):
    # 所有模型推理依赖放在显式运行分支，默认检查只需标准库。
    import accelerate
    import diffusers
    import numpy as np
    import torch
    import transformers
    from diffusers import StableDiffusionPipeline
    from huggingface_hub import snapshot_download

    report_path = folder / "report.json"
    if report_path.exists() or any(folder.glob("*.png")):
        raise ValueError("输出目录已有实验结果，请换一个空目录以保留原始记录")
    folder.mkdir(parents=True, exist_ok=True)
    torch.set_num_threads(4)
    torch.use_deterministic_algorithms(True)
    snapshot = Path(snapshot_download(
        MODEL, revision=REVISION, allow_patterns=[
            "model_index.json", "scheduler/*.json", "text_encoder/config.json",
            "text_encoder/model.fp16.safetensors", "tokenizer/*", "unet/config.json",
            "unet/diffusion_pytorch_model.fp16.safetensors", "vae/config.json",
            "vae/diffusion_pytorch_model.fp16.safetensors", "README.md", "LICENSE.md"]))
    pipeline = StableDiffusionPipeline.from_pretrained(
        snapshot, variant="fp16", torch_dtype=torch.float32,
        local_files_only=True, use_safetensors=True).to("cpu")
    pipeline.set_progress_bar_config(disable=True)
    report = {
        "model": MODEL, "revision": REVISION, "cases": CASES, "outputs": [],
        "source_sha256": file_hash(__file__),
        "environment": {"python": platform.python_version(), "platform": platform.platform(),
                        "torch": torch.__version__, "numpy": np.__version__,
                        "diffusers": diffusers.__version__, "transformers": transformers.__version__,
                        "accelerate": accelerate.__version__, "device": "cpu",
                        "dtype": "float32", "threads": 4},
        "settings": {"height": 512, "width": 512, "guidance_scale": 0.0,
                     "variant": "fp16", "negative_prompt": None},
        "scheduler": type(pipeline.scheduler).__name__,
        "scheduler_config": dict(pipeline.scheduler.config),
        "weights": [{"file": str(p.relative_to(snapshot)), "bytes": p.stat().st_size,
                     "sha256": file_hash(p)} for p in sorted(snapshot.rglob("*.safetensors"))],
        "license_sha256": file_hash(snapshot / "LICENSE.md"),
        "parameters": {name: sum(p.numel() for p in getattr(pipeline, name).parameters())
                       for name in ["text_encoder", "unet", "vae"]},
    }

    def tensor_hash(tensor):
        return sha256(tensor.contiguous().numpy().tobytes()).hexdigest()

    for case in CASES:
        generator = torch.Generator(device="cpu").manual_seed(case["seed"])
        noise = torch.randn((1, 4, 64, 64), generator=generator, dtype=torch.float32)
        pipeline.scheduler.set_timesteps(case["steps"])
        initial = noise * pipeline.scheduler.init_noise_sigma
        started = perf_counter()
        with torch.inference_mode():
            result = pipeline(prompt=case["prompt"], height=512, width=512,
                              num_inference_steps=case["steps"], guidance_scale=0.0,
                              latents=noise.clone(), generator=generator)
        elapsed = perf_counter() - started
        filename = folder / (case["id"] + ".png")
        result.images[0].save(filename)
        row = {**case, "image": filename.name, "image_sha256": file_hash(filename),
               "noise_sha256": tensor_hash(noise),
               "initial_latent_sha256": tensor_hash(initial),
               "init_noise_sigma": float(pipeline.scheduler.init_noise_sigma),
               "timesteps": pipeline.scheduler.timesteps.tolist(), "seconds": elapsed,
               "prompt_tokens": pipeline.tokenizer(case["prompt"]).input_ids,
               "nsfw_content_detected": result.nsfw_content_detected}
        report["outputs"].append(row)
        report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
        print(f"{case['id']}：完成，{elapsed:.2f} 秒", flush=True)
    validate(report_path)
    return report


def show(report):
    print(f"模型：{report['model']}；revision={report['revision']}")
    print("已保存 6 次真实生成：512×512，CPU float32，guidance_scale=0")
    for row in report["outputs"]:
        print(f"{row['id']}：{row['steps']} 步，seed={row['seed']}，图像指纹通过")
    print("同一种子各组的起始噪声一致；默认模式只核对记录，不重新生成。")


if __name__ == "__main__":
    parser = ArgumentParser(description=__doc__)
    parser.add_argument("--run", action="store_true", help="下载/加载固定模型并实际生成")
    parser.add_argument("--out", type=Path, help="真实生成使用的空输出目录")
    parser.add_argument("--record", type=Path, default=DEFAULT, help="默认核对的 JSON 路径")
    args = parser.parse_args()
    if args.run and args.out is None:
        parser.error("--run 需要 --out 指定一个空目录")
    show(run(args.out) if args.run else validate(args.record))

1. 不能。这只说明表示维度减少；原图与潜变量的数值精度、存储格式、压缩编码和模型大小都影响实际空间。
2. 训练只约束经过编码器得到的潜变量。随机先验未必与这些表示的分布一致，解码器可能收到不熟悉的位置。
3. 两个系数是 $\sqrt{0.36}=0.6$ 与 $\sqrt{0.64}=0.8$；结果为 $0.6\times0.5+0.8\times(-1)=-0.5$。
4. 时间说明信号与噪声的比例。训练时噪声由我们抽取，可作监督目标；生成时只有当前状态，网络预测的是噪声估计，不会拿到目标原图对应的“正确答案”。
5. $2+3(1-2)=-1$。当 $s>1$ 时发生外推，结果可以超出两种估计之间的区间。
6. 位置为 $(1-0.25)(-1)+0.25\times3=0$，速度为 4。每步前进 1，依次到 0、1、2、3。这是已知端点的恒定速度例子；实际生成中的速度由网络重新估计。
7. 两种损失预测的量不同，不能直接比较数值；采样调用次数也不同。还要区分条件符合、质量、多样性和计算成本，并使用多个种子。
8. 可以先固定同一个训练好且已按验证规则选中的网络，再改变欧拉步数。固定标签、初始噪声和显示处理，记录全部样本、条件检查、时间和网络调用次数。步数更多只减少某些离散误差，不保证弥补模型学错的速度场。
9. 筛选会隐藏失败率。可以预先固定提示集合、种子与数量要求，把全部输出编号，按颜色、杯子数量、把手结构等项目逐项盲评，保存分歧和原图，避免把一张漂亮样例当作总体表现。
10. 至少保存模型修订、提示、种子、调度器、时间点、精度、环境和原图。单个新种子只能补充一个案例，不能据此判断通用成功率或跨设备复现保证。