版权属于：
人工智能炼丹君
作品采用：
《
署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0)
》许可协议授权
- 首页
- /
- AIGC Fundamentals
- /
- 正文
AIGC 基本功|自回归视频生成与 Forcing 范式-Forcing
自回归视频生成与 Forcing 范式：流式出帧的账，和曝光偏差的坑
所属方向：生成范式 ｜ 难度：进阶 ｜ 前置知识：DDPM 训练目标与采样流程（$\bar{\alpha}$ 记号、DDIM 单步更新直接沿用）、DiT 架构拆解（潜空间 patch 化成 token 的约定）
关键词：自回归视频、Diffusion Forcing、Self-Forcing、teacher forcing、曝光偏差、KV cache、流式生成
01. 为什么需要它
你盯着一个视频生成产品等了 3 秒，第一帧才出来——不是网速问题，是范式问题。
本文先以全序列双向扩散为对照：21 个潜帧（对应约 5 秒成片）拼成一条 3 万多 token 的序列，帧和帧之间双向可见，整段一起降噪。代价是「流式」这个词跟它彻底无缘——最后一帧没去噪完，第一帧就不能给你看。这不是工程优化能修的：双向注意力在数学上要求未来帧也参与当前帧的去噪，未来帧不存在，这一步就算不了。
我按 Wan2.1-T2V-1.3B 的真实拓扑和 Self-Forcing 的默认配置，把这笔账算成了 MAC 数（第 06 节有完整过程）：在假设有效算力 400 TFLOPS、只计 Transformer 的理想算术估算中，5 秒视频的全序列扩散首帧约 2.83 秒；换成因果自回归加 KV cache，0.20 秒就能吐出第一帧——账本中的 13.98 倍差距（不是实测 GPU 延迟）。而总算力几乎没变（比值 0.875，自回归反而略省）。
但把「一帧一帧往后生成」这条路走通，会立刻撞上一个语言模型社区早就算过的老账：训练时模型看的上下文是真值帧（teacher forcing），推理时上下文只能换成模型自己的输出。这两件事不是一回事，误差会顺着自回归链条滚雪球——这就是曝光偏差（exposure bias）。
Forcing 系列就是围着这两个问题打转的：Diffusion Forcing 把「下一帧预测」和「扩散去噪」缝成一个目标，让每帧带独立噪声级；Self-Forcing 干脆在训练时就用模型自己的 rollout 当上下文，把训练分布直接掰成测试分布。这篇把两个问题都算成数：流式的账用真实拓扑算，曝光偏差的坑用一个能完整复现的玩具实验拆开看。
02. 最小可用理解
把三种范式并排放，差别其实只有一句话：每一帧被允许看什么、以什么噪声级被看到。
这张图要看什么：上行是全序列扩散——所有帧共用同一个噪声级、一起降、双向注意力互相可见，代价是 4 步全部算完才有第一帧；中行是 teacher forcing——只有当前帧带噪声，上下文全是浅色真值帧，而推理时真值帧不存在，误差就从这里的「换上下文」进来；下行是因果自回归采样；训练时使用自身 rollout 对应 Self-Forcing，而独立噪声的 Diffusion Forcing 训练上下文仍来自数据——历史帧干净地躺在 KV cache 里，当前 block 从高噪声一路降到干净，其中“训练用自身 rollout”是 Self-Forcing 的额外设计。
三个要点：
- 全序列扩散：质量上限高（未来帧参与去噪），双向联合去噪通常须整块完成才能交付；可变长度、滑窗或块间生成需要额外设计，不能称为所有此类模型永远不能流式。
- Diffusion Forcing：给每帧分配独立的噪声级 $k^{(t)}$。训练目标退化成「对每一帧做条件去噪」，采样时可以逐帧（逐 block）从高噪声降到干净再吐出去——这就是「next-token prediction 与全序列扩散的合流」这句话的数学含义。
- Self-Forcing：承认上下文永远是模型自己的输出，于是在训练时就真的 rollout——上一帧生成完、进 KV cache、下一帧从它出发，整段视频算一个整体损失，而不是逐帧各自算。
顺带把「Forcing」这个词的出处交代掉。它来自语言模型的 teacher forcing：训练 RNN 语言模型时，每一步的输入都用数据集里的真值 token「强行喂入」（force），而不是模型上一步的输出。这个约定让训练可以并行、梯度稳定，但也埋下了训练-测试分布错位的种子——语言模型社区管它叫 exposure bias，几十年里试过 scheduled sampling、DAgger 各种解法。Forcing 系列的谱系就是围绕这个错位逐步收紧的过程：Diffusion Forcing 先把「下一帧预测」的接口扩展成「每帧带独立噪声级的条件去噪」，让自回归的骨架里能塞进扩散的表达力；Self-Forcing 再把训练时的上下文从真值换成模型自己的 rollout，直接对齐两个分布。视频把这个问题变得更尖锐：语言模型一步错一个 token，视频一步错的是一整帧，而且帧与帧之间还有时间维度上的积分效应（下一节的主角）。
03. 数学推导
3.1 训练目标与推理目标的错位
为展示上下文错位，可用一个简化的逐帧去噪回归目标：
$$\mathcal L_{TF}=E_{x,k,\epsilon}\sum_t\|x_t-f_\theta(z_t(k_t),k_t,x_{<t}^{gt})\|^2.$$
推理时把真值历史换成模型生成历史 $\hat x_{<t}$，但通常不在推理现场优化损失。用同样的去噪回归在生成历史上评测，只是本文的受控诊断量，不是视频生成质量的完整定义；任意生成视频与某条真值逐帧 MSE 也不是 Self-Forcing 的训练目标。
训练和评测使用同一个 $f_\theta$ 时，上下文从数据历史换成模型历史即可造成分布偏移。充分容量下精确学习每个条件分布可得到正确联合分布；现实中的估计误差、有限容量和长时 rollout 会放大这一差异。
3.2 曝光偏差为什么会滚雪球：慢变量与快变量
把误差沿着自回归链条往前传一步。设第 $t$ 帧的上下文误差是 $e_t$，对很多动力系统可以近似成线性递推 $e_t = A e_{t-1} + \delta_t$，其中 $\delta_t$ 是本帧新引入的误差。下面标量方差式假设初始误差为零、增量独立同分布且零均值；一般矩阵非正规性、相关误差和模型偏差会改变增长。关键量是传递矩阵的谱半径。按谱半径把状态拆成两块看：
慢变量（谱半径 $\lambda_{\text{slow}}$ 接近 1，近似积分环节）：
$$e_t = \lambda_{\text{slow}} e_{t-1} + \delta_t \quad \Rightarrow \quad \mathrm{Var}[e_T] = \mathrm{Var}[\delta] \cdot \frac{1 - \lambda_{\text{slow}}^{2T}}{1 - \lambda_{\text{slow}}^{2}} \quad \xrightarrow{T \to \infty} \quad \frac{\mathrm{Var}[\delta]}{1 - \lambda_{\text{slow}}^{2}}$$
$\lambda_{\text{slow}} = 0.97$ 时系数是 $1 / (1 - 0.9409) \approx 16.9$，误差方差被放大约 17 倍，而且随长度单调增长——这就是「滚雪球」的数学形态。
快变量（谱半径 $\rho_{\text{fast}}$ 明显小于 1，收缩映射）：
$$e_t = \rho_{\text{fast}} e_{t-1} + \eta_t \quad \Rightarrow \quad \mathrm{Var}[e^{\text{fast}}] \approx \frac{\mathrm{Var}[\eta]}{1 - \rho_{\text{fast}}^{2}}, \quad \rho_{\text{fast}} < 1$$
这是一个有界的常数抬升：旧误差每帧被乘上 $\rho_{\text{fast}}$ 衰减掉，只有本帧新增的误差活着。$\rho_{\text{fast}} = 0.85$ 时系数约 3.6 倍，早期仍随长度增大，但更快接近上界。
所以「曝光偏差有多严重」这个问题没有单一答案：在 $|\lambda|<1$ 的独立增量模型中，两者都有界，只是慢变量更晚饱和；$\lambda=1$ 方差才线性增长，$|\lambda|>1$ 才可能指数发散。这也是为什么第 08 节的误解二里，光看「平均误差涨了几倍」会得出错误结论。
3.3 Diffusion Forcing：每帧一个独立噪声级
标准全序列噪声训练通常共享 $k$；Diffusion Forcing 对各帧独立采噪声级，并把带噪历史一起作为条件。例如噪声预测形式为：
$$\mathcal L_{DF}=E_{x,k_{1:T},\epsilon_{1:T}}\sum_t w(k_t)\|\epsilon_t-\epsilon_\theta(z_{\le t},k_{\le t})\|^2,$$
其中 $z_t=\sqrt{\bar\alpha_{k_t}}x_t+\sqrt{1-\bar\alpha_{k_t}}\epsilon_t$；也可等价改写成带相应权重的 $x_0$ 预测。历史 $z_{<t}$ 通常也带各自的噪声，不能只在公式里写干净的 $x_{<t}$。
统一噪声级可恢复共享噪声训练形式，但不会自动把因果遮罩变成双向注意力；把历史噪声降到 0、仅对当前块去噪，则给出自回归采样接口。当前块仍从高噪声逐步降噪，不是“把当前噪声取最小就等于 next-token prediction”。
3.4 Self-Forcing：在训练时就把分布掰过来
Self-Forcing 第 3.3 节 用自身自回归 rollout 产生整段视频，再做视频级分布匹配。以 DMD 为例，目标可写为
$$\mathcal L_{SF}=E_k\big[D_{KL}(p_{\theta,k}(x^{1:T})\|p_{data,k}(x^{1:T}))\big].$$
这里比较生成与数据的加噪联合分布，借助教师/学生 score 估计更新；论文还考察 SiD、GAN 损失。它不是逐帧配对 MSE，也不是 DAgger 的同义词。二者都关注自生成上下文，但经典 DAgger 需要专家对学习器访问状态提供标签，Self-Forcing 不以这种标签循环定义。
本文下面的线性回归 toy 只诊断上下文分布错位，并做 DAgger 风格重拟合；它没有实现 DMD，不能用它的失败证明 Self-Forcing 失败。
3.5 梯度怎么穿过 rollout
Self-Forcing 为控制显存，每个序列随机选一个去噪退出步，只保留最终被选步骤的反传，并阻断先前帧 KV cache 的梯度；这比笼统说“回传最近几帧”更准确。普通一阶反传并不自动产生 Hessian 的二阶交叉项。
本文 toy 的重拟合则是收集自身 rollout 特征，再做岭回归并用 line search 混合权重；它不对 rollout 链做可微反传。line search 和发散保护只约束这个实验，不能外推成 Self-Forcing 的固有不稳定性。
04. 代码实现
完整实验在 forcing_lab.py（附录有全文，numpy 单文件可跑）。设计原则是让三种范式唯一的差别就是特征函数：
- 动力系统：3 维慢变量 $u_t = \Lambda u_{t-1} + G w_{t-1} + 0.02 \xi$（谱半径 0.97）+ 3 维快变量 $w_t = \tanh(A_w w_{t-1} + B_w c) + 0.1 \xi$（谱半径 0.85），再加一个每条序列固定、模型可见的条件向量 $c$（类比文本 embedding——没有它第一帧不可预测，实验就没法做）。
- 去噪：4 步 DDIM（eta=0），噪声步 $[1000, 750, 500, 250]$，线性 beta 表。特征维数两范式完全相同（$F = 3D + 4 + 1 = 23$），线性回归头（岭回归闭式解）——给定数据可求确定的回归解，但数据、噪声、特征和模型假设仍决定结果，不能把差异全部归因于范式。
- 训练集 6000 条 × 72 帧，测试集 1500 条；训练视野 24 帧，外推到 72 帧。
公平性是这么保证的：同一个回归头 $W$ 拟合三版——全序列版用双向特征整段拟合；teacher forcing 版用因果特征、上下文取真值；自回归评测用的模型与 teacher forcing 版共享同一组参数，差别只在评测时喂什么上下文。所以下表里 rollout 和 teacher forcing 之间 8.3 倍的差距里没有任何训练差异的成分，全部来自「上下文是谁生成的」。
核心代码（节选自 forcing_lab.py，去掉 Q4 受控实验分支）：
def feat_causal(z_t, hist, c):
if len(hist) == 0:
prev = np.zeros(D)
past_mean = np.zeros(D)
else:
prev = hist[-1]
past_mean = np.asarray(hist[-4:]).mean(axis=0)
return np.concatenate([z_t, prev, past_mean, c, np.ones(1)])
def feat_bidir(z_seq, t, c):
zp = z_seq[t - 1] if t > 0 else np.zeros(D)
zn = z_seq[t + 1] if t + 1 < len(z_seq) else np.zeros(D)
return np.concatenate([z_seq[t], zp, zn, c, np.ones(1)])
def sample_causal(W, rng, T_out, c, steps=STEPS, ctx_ts=0, hist_true=None):
"""ctx_ts: 把刚生成的帧按这个时间步加噪后才写进历史。
hist_true 给定时上下文永远取真值 = teacher forcing 评测。"""
out = np.empty((T_out, D))
hist = []
for t in range(T_out):
a0 = abar_of(steps[0])
z = np.sqrt(1.0 - a0) * rng.standard_normal(D)
for i, ts in enumerate(steps):
if hist_true is not None:
ctx = hist_true[:t]
else:
ctx = np.array(hist) if len(hist) else np.zeros((0, D))
xhat = feat_causal(z, ctx, c) @ W
if i + 1 < len(steps):
z = ddim_step(z, xhat, abar_of(ts), abar_of(steps[i + 1]))
out[t] = xhat
hist.append(xhat.copy())
return out
这张图要看什么：左图三条曲线在训练视野边界（第 24 帧，竖虚线）附近的分叉——teacher forcing 平在 0.01 附近，但它测试时不可得；自回归 rollout 从第 1 帧起就比 teacher forcing 高一截，越过视野后继续单调上爬。右图把 rollout 的误差拆成快慢两块：慢变量（红）在所测 72 帧窗口内继续增大，不能据此证明无限长度发散，快变量（绿）基本是平的——3.2 节的预测被数据证实。
真实输出（逐帧 MSE，每维平均平方误差）：
| 帧号 | 全序列扩散 | teacher forcing | 自回归 rollout |
|---|---|---|---|
| 0 | 0.36699 | 0.32465 | 0.32471 |
| 1 | 0.33968 | 0.02699 | 0.27550 |
| 11 | 0.20119 | 0.01126 | 0.17954 |
| 23（视野边界前最后一帧） | 0.24698 | 0.01152 | 0.22402 |
| 47 | —（非流式） | 0.01152 | 0.41608 |
| 71 | — | 0.01152 | 0.60288 |
三个范式在视野内的总账：全序列扩散 0.23990，teacher forcing 0.02536，自回归 rollout 0.20941。teacher forcing 比 rollout 低 8.3 倍，这 8.3 倍全是曝光偏差——同一个模型、同一套参数，只是上下文来源不同。
把 rollout 的误差按 3.2 节拆开：
| teacher forcing | rollout | 放大倍数 | 行为 | |
|---|---|---|---|---|
| 快变量 w | 0.03208 | 0.13269 | 4.14× | 恒定抬升，不随长度涨 |
| 慢变量 u | 0.01864 | 0.28614 | 15.35× | 积累，理论预测约 16.9× |
慢变量 15.35 倍对上理论值 16.9 倍，量级和方向都对得上（实测略低是因为有限长度截断了增长：“翻倍长度”若来自有限区间拟合，不能当成有界理论模型的渐近性质）。越过训练视野后慢变量误差是视野内的 2.45 倍，快变量只有 1.01 倍——扩散模型 rollout 的伤害是有结构的，不是均匀糊在所有维度上。
DAgger 轮次（Q3）的实现细节值得交代：每轮用当前权重 rollout 出上下文，重新收集特征-标签对，在岭回归闭式解上以步长 $\alpha = 0.3$ 混合新权重；然后做步长折半的 line search——实际输出若为 $\alpha=0.01875$，对应从 0.3 连续减半四次；应以脚本打印的接受步长为准，说明更新方向已接近退化；外推末帧超过 round0 的 50 倍触发发散保护。Q5 的训练上下文噪声扫描是同一套循环的外层：给训练时的真值上下文按 $k \in [0, 400]$ 个噪声步加噪（用 3.3 节同一个 $\bar{\alpha}$ 表映射），再在 TF 与 rollout 两端评测——06 节的精度换稳定性曲线就是这么来的。
05. 工业级实现对照
玩具里的一切，在 Self-Forcing 的官方实现（pipeline/causal_inference.py 的 CausalInferencePipeline，以 2026-09 时的代码为准）里都有对应物，而且配置文件里每一条都能对上：
| 玩具实验 | Self-Forcing 真实配置（configs/self_forcing_dmd.yaml） |
|---|---|
| 4 步去噪 $[1000, 750, 500, 250]$ | denoising_step_list: [1000, 750, 500, 250] |
| 每个 block 3 帧 | num_frame_per_block: 3 |
| 条件向量 $c$ | conditional_dict（文本编码，cross-attention 消费） |
| 训练用自己 rollout 的上下文 | 训练管线 rollout + KV cache，distribution_loss: dmd |
| 底座 | Wan2.1-T2V-14B，学习率 2.0e-06 |
推理循环的骨架是按 block 走的，每个 block 3 帧，5 次前向：
- Step 3.1 空间去噪：当前 block 的帧从噪声步 1000 开始，沿
denoising_step_list降到 250，共 4 次前向。每次前向都通过 KV cache 读到全部历史 token，当前帧的 key/value 也写进缓存。 - Step 3.2 记录去噪输出
denoised_pred。 - Step 3.3 刷缓存：这是整个管线最值得盯的一步——前 4 次前向里写进 KV cache 的是带噪声输入的 key/value，和「历史是干净的」这个推理前提不符。所以代码用干净的
denoised_pred在context_noise时间步上重跑一次前向，把缓存里这个 block 的条目覆盖掉。这就是 Q5 实验里「训练上下文噪声」在推理侧的镜像：缓存里的历史按多脏的口径存，下游就按什么口径消费。 - Step 3.4 起始帧号前移 3 帧潜帧，进入下一个 block。
底座 Wan2.1-T2V-1.3B 的拓扑是账本的地基：dim=1536、30 层、12 头（head_dim=128）、FFN 8960、patch (1,2,2)、VAE stride (4,8,8)。Self-Forcing 默认视频形状 [1, 21, 16, 60, 104]——21 个潜帧，每帧 $(60/2)\times(104/2)=1560$ 个 token；16 是通道数，进入每个 patch 的特征维度，不再乘进 token 数（60、104 是潜空间宽高，空间 patch 是 $2 \times 2$），整段 32760 个 token，约 81 个像素帧，16 fps 下 5.06 秒。训练侧的规模感：基于 CausVid 蒸馏，600 次迭代、64 张 H100、2 小时以内。
双向底座怎么改成因果的？ Wan2.1 的 DiT 本来是双向注意力，改造没有动预训练权重的语义：把 self-attention 换成「空间维双向 + 时间维因果」的混合遮罩，rollout 时每个 block 的 token 先以 query 身份读完整个缓存，再把自己的 key/value 写进去；第 0 帧没有历史，等价于一次图像生成。文本条件走 cross-attention——它的 K/V 只有 512 个 T5 token，缓存恒定 0.088 GB，一次算好全程复用，跟帧数无关。训练侧在 rollout 前向之上叠 DMD 蒸馏损失（distribution_loss: dmd），让少步学生的分布对齐教师——这就是 600 次迭代能收敛的原因：监督信号来自蒸馏，不是从零拟合数据分布。
为什么缓存刷新那一步不能省？ 4 步去噪的每次前向都会把「带噪声输入」的 key/value 写进缓存，而下游 block 读缓存时的前提是「历史是干净的」。Step 3.3 用干净输出在 context_noise 时间步重刷一遍，本质是把「缓存里的历史有多脏」从「去噪过程的残留」变成一个显式超参——第 06 节 Q4 实验会证明模型对这个超参极其敏感：干净上下文训练的模型，推理时给上下文加 50 步噪声，误差就从 0.025 涨到 0.047。
06. 代价与边界
流式不是免费的，把账算干净。
这张图要看什么：左图的台阶是每个 block 算完才吐 12 个像素帧（3 潜帧），出帧节奏整体贴着 16 fps 的实时线走，橙色竖线是全序列扩散一次性交付的时刻；中图和右图是一对警告——不滚动缓存时显存和单 block 时间都随长度线性涨，9 帧滚动窗口把两者同时钉成常数。
口径先交代清楚（逐项对齐公开配置，代码在附录 streaming_ledger.py）：每层每 token 的 MAC = 注意力分数（对每个可见 key 做两次内积，$2 \times n_{\text{key}} \times d$）+ 四个注意力投影（$4 d^2$）+ FFN（$2 \times d \times 8960$）+ 一次 cross-attention（对 512 个文本 token 的分数与投影）。KV cache 每潜帧的字节数是 $2 \times 30 \times 1560 \times 1536 \times 2 \text{ B} \approx 0.268$ GB——K、V 两份，乘 30 层、每帧 1560 token、$d = 1536$、bf16 两字节——21 帧合计 5.624 GB。时间按 H100 bf16 有效 400 TFLOPS 折算（1 MAC = 2 FLOP），只算 transformer 主体，不含 VAE 解码与文本编码。
算力账（400 TFLOPS 有效算力折算，MAC 计入注意力 QKV 与投影）：
| 全序列扩散 | 因果 + KV cache | |
|---|---|---|
| 单步去噪 MAC | 1.4142e+14 | 首块 8.09e+12 → 末块 2.02e+13 |
| 4 步总 MAC | 5.6567e+14 | 4.9514e+14（比值 0.875） |
| 总时间 | 2.828 s | 2.476 s |
| 首帧延迟 | 2.828 s | 0.202 s（13.98×） |
| 稳态节奏 | 一次性交付 | 每块 505.1 ms，实时预算 750 ms，余量 1.48× |
两点容易被误读：其一，自回归并不显著省总算力（0.875 倍，注意力遮罩少算的钱被 KV cache 刷新的额外前向花掉了大半），它买到的是首帧延迟和出帧节奏；其二，若每层稠密保存 32760×32760×12 个 bf16 分数，约 25.76 GB（24.0 GiB）（32760 token 的平方 × 12 头 × bf16），所以 FlashAttention 不是优化项是必需品（前置阅读见那篇）。
显存账（bf16 KV cache，30 层全量）：21 帧全量 5.624 GB，9 帧滚动窗口 2.410 GB（43%）。不滚动的话，生成到 336 潜帧（336 个潜帧按 4 倍时间解码约 1341 帧，在 16 fps 下约 83.8 秒）时单 block 要 5.8 秒、缓存 89.98 GB——两头都爆炸；9 帧滚动窗口下恒定 303.2 ms、2.41 GB。「越生成越慢」不是自回归的本质属性，是「不滚动」这个实现选择的属性。
质量与稳定性的账，用两个受控实验说：
这张图要看什么：左图是曝光偏差的「单位换算」——把 rollout 的误差水平放到「干净模型 + 给真值上下文加噪」的曲线上插值，等效于给上下文加了约 160 个噪声步（$\bar{\alpha}$ 约 0.76）；右图是精度换稳定性的折中——训练时给上下文加的噪声从 0 加到 250，外推末帧误差先降后平，但视野内精度从 0.025 恶化到 0.096。
- 等效噪声级 ≈ 160 步：自回归 rollout 的伤害，等价于把干净的真值上下文往里掺这么多噪声。这给了当前模型、当前噪声表与误差指标下的诊断刻度，不能不经校准跨模型比较，而不是一句「会变差」。
- 训练上下文加噪的最优点很小：扫描 $k \in [0, 400]$，视野内误差在 $k=50$ 处最优（0.20516，对照 $k=0$ 的 0.20914），$k=250$ 时视野内恶化到 0.34373 但外推末帧确实最稳（0.49690 对 0.60284）。加噪换来的鲁棒性是真的，但精度代价涨得比稳定性收益快，别一上来就拉满。
- 训练噪声级和推理噪声级必须配套（Q4 受控实验）：干净上下文训练的模型，推理时给上下文加 50 步噪声，误差就从 0.025 涨到 0.047——模型对「上下文多脏」这件事的敏感性是训练时铸死的，这也解释了 Step 3.3 为什么非刷缓存不可。
- 一个诚实说明：这个玩具里全序列扩散的逐帧误差（0.24）反而比自回归 rollout（0.21）高——线性回归头太弱，双向注意力没占到便宜。真实系统的质量取决于架构、训练与蒸馏，双向上下文本身不构成质量必胜保证，所以这条不能外推成「因果不亏质量」，只能说质量差不是自回归路线的主要障碍，曝光偏差和工程复杂度才是。
07. 经典论文脉络
先把「为什么不用更老的解法」说掉。Scheduled sampling（按概率把训练输入从真值换成模型自己的输出）在语言模型上就有分布畸变的老毛病：混着喂会让模型面对一个训练里从未出现过的「半真半假」分布；搬到扩散模型上问题更糟，因为上下文还带着噪声级这个第二维度——真值帧和自生成帧在不同的 $k$ 下混在一起，畸变是二维的。GAN 式判别器（让判别器区分真值轨迹和 rollout 轨迹）能补分布层面的监督，但训练不稳、和扩散目标叠加的工程成本高。DAgger 路线的好处是监督信号始终来自真值（专家），模型只是把「自己会走到的地方」纳入训练分布，不需要引入新网络。Self-Forcing 采用自 rollout 的视频分布匹配，与 DAgger 共享关注分布偏移的动机，但目标与监督信号不同。
- Diffusion Forcing（arXiv 2407.01392，NeurIPS 2024）：把下一帧预测和全序列扩散统一成「每帧独立噪声级的条件去噪」，证明了两者是同一个目标的两个端点。后续的流式视频模型（含游戏引擎式的实时生成）基本都沿用「逐 block 从噪声降干净 + 历史进缓存」的采样骨架。
- CausVid：把双向扩散模型蒸馏成因果自回归的 few-step 生成器，证明「因果化 + 步数蒸馏」可以叠加。Self-Forcing 的训练基建直接继承自它。
- Self-Forcing（arXiv 2506.08009）：指出曝光偏差在视频扩散上的具体形态，给出「训练时 rollout + KV cache + 视频级整体损失」的解法。600 次迭代、64 H100、2 小时以内，这是这条路线「能用普通实验室的预算续命」的直接证据。
- 同方向的中文篇目：DDIM 采样器（本文的 4 步去噪就是它）、流匹配（另一条「少步数化」的路线）、潜空间扩散（21 潜帧从哪来）。
08. 常见误解
误解一：「自回归省算力。」 总 MAC 比值 0.875——基本不省。注意力遮罩省下的 FLOPs，被 Step 3.3 的缓存刷新前向（每块 5 次前向对 4 次）吃掉了大半。自回归买到的是 13.98 倍的首帧延迟和贴着实时线的出帧节奏，把「省算力」当卖点去汇报会翻车。
误解二：「曝光偏差就是误差随长度无限涨。」 本例慢变量系数 0.97 仍小于 1，理想独立增量模型有界；快变量更快饱和。实际学习器可因模型偏差或不稳定闭环继续增长，需另测有效误差传播，不能把快慢直接等同于有界/发散。看「平均逐帧误差涨了几倍」会把 15.35 倍和 4.14 倍混成一个数，既高估也低估——先拆维度再下结论。
误解三：「在自生成上下文上逐帧重拟合，轮次越多就越稳。」 看 DAgger 循环只盯视野内逐帧损失时发生了什么：
这张图要看什么：蓝色柱（训练视野内 MSE）从 0.209 逐轮降到 0.187，看起来一路向好；红色柱（外推第 72 帧 MSE，对数轴）从 0.60 涨到 0.83 再爆到 97.15——第 2 轮的外推末帧是第 0 轮的 161 倍。视野内的改进是真的，但把权重推离了能外推的区域。这说明本 toy 的视野内回归目标不能保证长时稳定；Self-Forcing 用联合分布匹配，不能把本 toy 当作论文算法消融或 DAgger 的普遍反例。我的实验里第 2 轮被发散保护停掉；换成对整段（含外推段）算损失的变体，两轮内慢变量降 11.9%、快变量降 8.5%，外推同步受控。
误解四：「训练时给上下文加噪越多越鲁棒。」 右图（06 节）的折中曲线明确说不是，Q5 的原始数据在这里：
| 训练上下文噪声步 $k$ | TF 评测 | 自回归视野内 | 外推 t=71 |
|---|---|---|---|
| 0 | 0.02536 | 0.20914 | 0.60284 |
| 50 | 0.02792 | 0.20516 | 0.51928 |
| 150 | 0.05050 | 0.27835 | 0.51929 |
| 250 | 0.09561 | 0.34373 | 0.49690 |
| 400 | 0.15762 | 0.29314 | 0.54894 |
本次扫描中 $k=250$ 的末帧 MSE 最低，但 rollout 视野内 MSE 0.34373 是 $k=50$ 的约 1.68 倍、$k=0$ 的约 1.64 倍；另一个指标 TF MSE 为 0.09561，较 $k=0$ 的 0.02536 恶化约 3.77 倍。两列不能混算——加噪不是只影响 rollout 端。结合 Q4（模型对上下文噪声级的敏感度在训练时铸死），「加噪」本身是一种要配套的契约，不是免费的保险。
误解五：「因果注意力必须丢质量。」 在这个玩具里没有丢（见 06 节诚实说明）；真实系统里质量损失主要来自曝光偏差与蒸馏误差，而不是「看不见未来」本身——CausVid 与 Self-Forcing 的生成质量支撑这一点。
09. 动手验证
三个脚本都在本文附录，numpy 单文件，无 GPU 依赖：
# 主实验：三种范式 + 快慢拆解 + DAgger 轮次 + 上下文噪声扫描（约 3~5 分钟）
python forcing_lab.py
# 流式账本：按 Wan2.1-1.3B 真实拓扑算 MAC / 显存 / 出帧节奏（纯算术，秒级）
python streaming_ledger.py
# 复现本文全部 5 张图
python make_figures.py对着输出核对三件事：
forcing_lab.py的 Q1 表里，teacher forcing 视野内均值应为 0.02536，自回归 rollout 应为 0.20941（比值约 8.3）；Q3 应打印「外推末帧已发散」并停止。streaming_ledger.py的首帧延迟比应为 13.98，KV cache 全量应为 5.624 GB。改任何拓扑参数（层数、头数、潜帧数）都应按比例传导——比如把潜帧数从 21 改成 42，全量缓存应精确翻倍到 11.25 GB。- 去 Self-Forcing 仓库打开
configs/self_forcing_dmd.yaml，核对denoising_step_list与num_frame_per_block是否与本文 05 节的表一致（上游若重构，以仓库为准）。 - 把
forcing_lab.py里的LAM_SLOW从 0.97 调到 0.85 重跑：慢变量的放大倍数应从 15.35 回落到 4 倍上下（理论值 $1/(1 - 0.85^2) \approx 3.6$），快变量几乎不动——这是 3.2 节「谱半径决定一切」最直接的验证。 - 把
streaming_ledger.py的LAT_FRAMES从 21 改成 42：全量 KV cache 应精确翻倍到 11.25 GB，首块延迟不变——「流式成本与已生成长度解耦」在代码里就是这么体现的。
10. 延伸阅读
- Diffusion Forcing: Next-token Prediction Meets Full-Sequence Diffusion（arXiv 2407.01392，NeurIPS 2024）
- Self Forcing: Bridging the Train-Test Gap in Autoregressive Video Diffusion（arXiv 2506.08009）
- 代码：guandeh17/Self-Forcing（
pipeline/causal_inference.py、configs/self_forcing_dmd.yaml） - 本系列相关篇目：KV cache、FlashAttention、DDIM 采样器、流匹配
下一篇自然是「少步数蒸馏」（DMD / 蒸馏到 1~4 步）——Forcing 解决了「怎么流式地生成」，蒸馏解决「每块少算几步」，两者拼起来才是实时视频生成的完整拼图。
读原论文的建议顺序：先读 Self-Forcing 的第 3 节（推理管线那五步，对照本文 05 节的表），再回头读 Diffusion Forcing 的第 3 节（目标函数的统一形式，对照本文 3.3 节）——反过来读会被「每帧独立噪声级」的抽象描述卡住，先看到具体的采样循环再回头抽象，会顺很多。玩具实验（04 节）建议在读论文之前跑一遍，数字先在手里，论文里每一句关于 exposure bias 的表述都能对号入座。
附录：完整代码
09 节用到的脚本全文如下（forcing_lab.py、streaming_ledger.py、make_figures.py）。复制到本地存成同名文件，按各脚本开头的依赖说明准备环境后即可运行。
forcing_lab.py
# -*- coding: utf-8 -*-
"""
forcing_lab.py —— 自回归视频生成 / Forcing 范式的玩具实验（只依赖 numpy）
跑法： python forcing_lab.py # 全量
python forcing_lab.py --fast # 小样本快速迭代
python forcing_lab.py --rounds 4 # 多做几轮 DAgger 风格 toy 重拟合
python forcing_lab.py --ctx 0 100 250 # 只扫这几个上下文噪声级
数据是一段"带条件的 6 维视频"，分成两块，故意让它们的时间尺度不同：
u（前 3 维）慢变量：u_t = Lam u_{t-1} + G w_{t-1} + 噪声，Lam 的谱半径 0.97
—— 一条会被积分记住的轨迹，误差会累积（相机轨迹、主体位置）
w（后 3 维）快变量：w_t = tanh(Aw w_{t-1} + Bw c) + 噪声
—— 收缩模态，误差会被动力学忘掉（纹理、局部细节）
输出顺序与正文表格一一对应：
Q1 三种范式的逐帧误差：全序列扩散 / teacher forcing / 自回归 rollout
Q2 曝光偏差拆成两块看：快变量是恒定抬升，慢变量是随帧号累积
Q3 DAgger 风格线性重拟合；不是 Self-Forcing 的 DMD 实现
Q4 上下文噪声 context_noise：训练时用什么噪声级，推理时就得用什么噪声级
"""
import argparse
import os
import numpy as np
# ═══════════════════════════════ 0. 配置 ═══════════════════════════════
DU = 3 # 慢变量 u 的维度（会被积分记住的分量）
DW = 3 # 快变量 w 的维度（收缩的分量）
D = DU + DW # 每帧"潜向量"的总维度
DC = 4 # 条件向量维度（类比文本 embedding）
T_TRAIN = 24 # 训练时的序列长度（horizon）
T_ROLL = 72 # 推理外推到 3 倍长度
N_STEPS_DIFF = 1000
STEPS = [1000, 750, 500, 250] # 对齐 Self-Forcing 的 4 步去噪
LAM_SLOW = 0.97 # 慢变量的自回归系数（谱半径）
RHO_FAST = 0.85 # 快变量转移矩阵的谱半径
G_SCALE = 0.09 # 快变量驱动慢变量的耦合强度
SIG_U = 0.02 # 慢变量的过程噪声
SIG_W = 0.10 # 快变量的过程噪声
LAMBDA = 3e-3 # ridge 系数（压住上下文权重，防止 rollout 重训时正反馈发散）
ALPHA = 0.3 # toy 重拟合 每轮的权重更新步长（1.0 = 完全换成新解）
SEED_SYS = 20250608
CACHE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "forcing_lab_cache.npz")
# ═══════════════════════ 1. 噪声表：线性 beta 调度（DDPM 原版）═══════════════════════
def build_schedule(n_steps=N_STEPS_DIFF, beta0=1e-4, beta1=0.02):
beta = np.linspace(beta0, beta1, n_steps)
return np.concatenate([[1.0], np.cumprod(1.0 - beta)]) # 下标 = 时间步 t
ABAR = build_schedule()
def abar_of(timestep):
t = min(max(int(round(timestep)), 0), N_STEPS_DIFF)
return float(ABAR[t])
# ═══════════════════════ 2. 数据：慢变量 + 快变量 + 条件 ═══════════════════════
def _ortho(rng, n, rho):
q, _ = np.linalg.qr(rng.standard_normal((n, n)))
return rho * q
def make_system():
rng = np.random.default_rng(SEED_SYS)
return dict(
Lam=_ortho(rng, DU, LAM_SLOW),
G=rng.standard_normal((DU, DW)) * G_SCALE,
Aw=_ortho(rng, DW, RHO_FAST),
Bw=rng.standard_normal((DW, DC)) * 0.9,
H=rng.standard_normal((DU, DC)) * 0.9,
)
def sample_sequences(rng, S, n, T):
"""采样 n 条长度为 T 的序列。c 是每条序列固定、模型可见的条件（类比文本）。"""
c = rng.standard_normal((n, DC))
cb = c @ S["Bw"].T # (n, DW)
hb = c @ S["H"].T # (n, DU)
u = np.tanh(hb) + SIG_U * rng.standard_normal((n, DU))
w = np.tanh(cb) + SIG_W * rng.standard_normal((n, DW))
out = np.empty((n, T, D))
for t in range(T):
out[:, t, :DU] = u
out[:, t, DU:] = w
w_new = np.tanh(w @ S["Aw"].T + cb) + SIG_W * rng.standard_normal((n, DW))
u_new = u @ S["Lam"].T + w @ S["G"].T + SIG_U * rng.standard_normal((n, DU))
u, w = u_new, w_new
return out, c
# ═══════════════════════════ 3. 两种特征（因果 / 双向）═══════════════════════════
#
# 两边都是"自己的观测 + 两个邻居/汇总向量 + 条件 + 偏置"，维数完全相同 = 3D+DC+1。
# 差别只在"能看哪几帧"——这个差别就是两种范式的全部定义：
# 因果 causal：上一帧的上下文 + 过去 4 帧均值（只看过去）
# 双向 bidir ：前一帧 + 后一帧的噪声观测（能看未来）
F_DIM = 3 * D + DC + 1
def feat_causal(z_t, hist, c):
if len(hist) == 0:
prev = np.zeros(D)
past_mean = np.zeros(D)
else:
prev = hist[-1]
past_mean = np.asarray(hist[-4:]).mean(axis=0)
return np.concatenate([z_t, prev, past_mean, c, np.ones(1)])
def feat_bidir(z_seq, t, c):
zp = z_seq[t - 1] if t > 0 else np.zeros(D)
zn = z_seq[t + 1] if t + 1 < len(z_seq) else np.zeros(D)
return np.concatenate([z_seq[t], zp, zn, c, np.ones(1)])
def fit_ridge(X, Y, lam=LAMBDA):
F = X.shape[1]
return np.linalg.solve(X.T @ X + lam * np.eye(F), X.T @ Y)
# ═══════════════════════════ 4. 采样器 ════════════════════════════
def ddim_step(z, xhat, a, a_next):
"""DDIM (eta=0)：固定噪声估计，只把噪声水平降到下一档。"""
eps = (z - np.sqrt(a) * xhat) / np.sqrt(max(1e-12, 1.0 - a))
return np.sqrt(a_next) * xhat + np.sqrt(1.0 - a_next) * eps
def sample_bidir(W, z_init, c, steps=STEPS):
"""全序列扩散：整条序列一起走 4 步，每步所有帧互相可见。非流式。"""
T = z_init.shape[0]
z = z_init.copy()
xhat = np.zeros_like(z)
for i, ts in enumerate(steps):
a = abar_of(ts)
xhat = np.stack([feat_bidir(z, t, c) for t in range(T)]) @ W
if i + 1 < len(steps):
z = ddim_step(z, xhat, a, abar_of(steps[i + 1]))
return xhat
def sample_causal(W, rng, T_out, c, steps=STEPS, ctx_ts=0, hist_true=None,
ctx_rng=None, collect=False, x_true=None):
"""因果自回归 rollout。
ctx_ts 把刚生成的帧按这个时间步加噪后才写进历史（对应 Self-Forcing 里
用 context_noise 刷新 KV cache 那一步）。0 = 干净上下文。
hist_true 给定时上下文永远取真值 —— teacher forcing 评测（测试时不可得）。
ctx_rng 给定时，连真值上下文也按 ctx_ts 加噪（Q4 用的受控实验）。
"""
a_ctx = abar_of(ctx_ts)
out = np.empty((T_out, D))
hist = []
rows_X, rows_Y = [], []
for t in range(T_out):
a0 = abar_of(steps[0])
z = rng.standard_normal(D) if a0 < 1e-8 else (
np.sqrt(a0) * (x_true[t] if x_true is not None else np.zeros(D))
+ np.sqrt(1.0 - a0) * rng.standard_normal(D))
for i, ts in enumerate(steps):
a = abar_of(ts)
if hist_true is not None:
ctx = hist_true[:t]
if ctx_rng is not None and a_ctx < 1.0 - 1e-12 and len(ctx):
ctx = (np.sqrt(a_ctx) * ctx
+ np.sqrt(1 - a_ctx) * ctx_rng.standard_normal(ctx.shape))
else:
ctx = np.array(hist) if len(hist) else np.zeros((0, D))
feat = feat_causal(z, ctx, c)
xhat = feat @ W
if collect and x_true is not None:
rows_X.append(feat.copy())
rows_Y.append(x_true[t].copy())
if i + 1 < len(steps):
z = ddim_step(z, xhat, a, abar_of(steps[i + 1]))
out[t] = xhat
if hist_true is None:
if a_ctx >= 1.0 - 1e-12:
hist.append(xhat.copy())
else:
hist.append(np.sqrt(a_ctx) * xhat
+ np.sqrt(1.0 - a_ctx) * rng.standard_normal(D))
if collect:
return out, np.array(rows_X), np.array(rows_Y)
return out
# ═══════════════════════════ 5. 数据集构造 ════════════════════════════
def build_gt_dataset(seqs, conds, rng, levels=STEPS, ctx_ts=0):
"""teacher forcing 数据集：上下文用真值帧（可按 ctx_ts 加噪）。"""
n, T, _ = seqs.shape
a_ctx = abar_of(ctx_ts)
Xs, Ys = [], []
for t in range(T):
if t == 0:
prev, pm = np.zeros((n, D)), np.zeros((n, D))
else:
prev = seqs[:, t - 1, :]
pm = seqs[:, max(0, t - 4):t, :].mean(axis=1)
if a_ctx < 1.0 - 1e-12:
prev = np.sqrt(a_ctx) * prev + np.sqrt(1 - a_ctx) * rng.standard_normal((n, D))
pm = np.sqrt(a_ctx) * pm + np.sqrt(1 - a_ctx) * rng.standard_normal((n, D))
for ts in levels:
a = abar_of(ts)
z = np.sqrt(a) * seqs[:, t, :] + np.sqrt(1.0 - a) * rng.standard_normal((n, D))
Xs.append(np.concatenate([z, prev, pm, conds, np.ones((n, 1))], axis=1))
Ys.append(seqs[:, t, :])
return np.concatenate(Xs, 0), np.concatenate(Ys, 0)
def build_bidir_dataset(seqs, conds, rng, levels=STEPS):
n, T, _ = seqs.shape
Xs, Ys = [], []
for ts in levels:
a = abar_of(ts)
z = np.sqrt(a) * seqs + np.sqrt(1.0 - a) * rng.standard_normal((n, T, D))
for t in range(T):
zp = z[:, t - 1, :] if t > 0 else np.zeros((n, D))
zn = z[:, t + 1, :] if t + 1 < T else np.zeros((n, D))
Xs.append(np.concatenate([z[:, t, :], zp, zn, conds, np.ones((n, 1))], axis=1))
Ys.append(seqs[:, t, :])
return np.concatenate(Xs, 0), np.concatenate(Ys, 0)
# ═══════════════════════════ 6. 度量 ════════════════════════════
def mse_curve(pred, truth, sl=None):
"""每帧每维的平均平方误差。sl 是维度切片，用来把快慢变量分开看。"""
e = (pred - truth) if sl is None else (pred[..., sl] - truth[..., sl])
return (e ** 2).mean(axis=(0, -1))
USL, WSL = slice(0, DU), slice(DU, D)
def ctx_gain(W):
"""模型对上下文的依赖强度：上下文两块特征对应权重的 Frobenius 范数。"""
return float(np.linalg.norm(W[D:3 * D, :]))
def doubling_length(curve, t0=2, t1=None):
t1 = t1 or len(curve) - 1
y = np.log(np.maximum(curve[t0:t1 + 1], 1e-12))
b = np.polyfit(np.arange(t0, t1 + 1), y, 1)[0]
return float(np.log(2.0) / b) if b > 0 else float("inf")
def fmt_row(tag, cur, cur_u, cur_w, T=T_TRAIN):
return (f" {tag:<12}{cur[:T].mean():>12.5f}{cur_u[:T].mean():>12.5f}"
f"{cur_w[:T].mean():>12.5f}{cur[min(23, len(cur) - 1)]:>12.5f}"
f"{cur[min(47, len(cur) - 1)]:>12.5f}{cur[-1]:>12.5f}")
# ═══════════════════════════ 7. 主流程 ════════════════════════════
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--rounds", type=int, default=3)
ap.add_argument("--n-train", type=int, default=6000)
ap.add_argument("--n-test", type=int, default=1500)
ap.add_argument("--alpha", type=float, default=ALPHA)
ap.add_argument("--fast", action="store_true")
ap.add_argument("--ctx", type=int, nargs="*",
default=[0, 50, 100, 150, 200, 250, 300, 400])
args = ap.parse_args()
if args.fast:
args.n_train, args.n_test = 2500, 500
S = make_system()
seqs_train, c_train = sample_sequences(np.random.default_rng(11), S,
args.n_train, max(T_TRAIN, T_ROLL))
seqs_test, c_test = sample_sequences(np.random.default_rng(12), S,
args.n_test, max(T_TRAIN, T_ROLL))
var_u = seqs_test[..., USL].var(axis=(0, 1)).mean()
var_w = seqs_test[..., WSL].var(axis=(0, 1)).mean()
print("=" * 78)
print(f"数据：u_t = Lam u_{{t-1}} + G w_{{t-1}} + {SIG_U} xi （慢变量，谱半径 {LAM_SLOW}）")
print(f" w_t = tanh(Aw w_{{t-1}} + Bw c) + {SIG_W} xi （快变量，谱半径 {RHO_FAST}）")
print(f"D={D}（u:{DU} + w:{DW}） DC={DC} 训练集 {seqs_train.shape} 测试集 {seqs_test.shape}")
print(f"慢变量边缘方差 {var_u:.4f} 快变量边缘方差 {var_w:.4f}")
print(f"训练视野 T={T_TRAIN}，外推到 T={T_ROLL}；去噪步 {STEPS}")
print(f"特征维数 F = 3D+DC+1 = {F_DIM}（因果与双向完全相同）")
print(f"最后一步的输入噪声级 abar({STEPS[-1]}) = {abar_of(STEPS[-1]):.4f}")
print("=" * 78)
# ── 基线 0：全序列扩散（非因果，整段一起解）─────────────────────
Xb, Yb = build_bidir_dataset(seqs_train[:, :T_TRAIN], c_train, np.random.default_rng(21))
Wb = fit_ridge(Xb, Yb)
rng_b = np.random.default_rng(31)
pred_bi = np.stack([sample_bidir(Wb, rng_b.standard_normal((T_TRAIN, D)), c_test[i])
for i in range(args.n_test)])
cur_bi = mse_curve(pred_bi, seqs_test[:, :T_TRAIN])
cur_bi_u = mse_curve(pred_bi, seqs_test[:, :T_TRAIN], USL)
cur_bi_w = mse_curve(pred_bi, seqs_test[:, :T_TRAIN], WSL)
# ── 基线 1：teacher forcing（上下文用真值帧）────────────────────
Xg, Yg = build_gt_dataset(seqs_train[:, :T_TRAIN], c_train, np.random.default_rng(22))
W_tf = fit_ridge(Xg, Yg)
rng_tf = np.random.default_rng(32)
pred_tf = np.stack([sample_causal(W_tf, rng_tf, T_TRAIN, c_test[i],
hist_true=seqs_test[i, :T_TRAIN])
for i in range(args.n_test)])
cur_tf = mse_curve(pred_tf, seqs_test[:, :T_TRAIN])
cur_tf_u = mse_curve(pred_tf, seqs_test[:, :T_TRAIN], USL)
cur_tf_w = mse_curve(pred_tf, seqs_test[:, :T_TRAIN], WSL)
# ── 基线 2：同一个权重做自回归 rollout ──────────────────────────
rng_sf = np.random.default_rng(33)
pred_sf0 = np.stack([sample_causal(W_tf, rng_sf, T_ROLL, c_test[i], ctx_ts=0)
for i in range(args.n_test)])
cur_sf0 = mse_curve(pred_sf0, seqs_test[:, :T_ROLL])
cur_sf0_u = mse_curve(pred_sf0, seqs_test[:, :T_ROLL], USL)
cur_sf0_w = mse_curve(pred_sf0, seqs_test[:, :T_ROLL], WSL)
print("\n【Q1】三种范式的逐帧 MSE（每维平均平方误差）")
print(f"{'帧号':>6}{'全序列扩散':>14}{'teacher forcing':>16}{'自回归 rollout':>16}")
for t in [0, 1, 2, 3, 5, 7, 11, 15, 19, 23]:
print(f"{t:>6}{cur_bi[t]:>14.5f}{cur_tf[t]:>16.5f}{cur_sf0[t]:>16.5f}")
print("\n【Q2】曝光偏差拆成快慢两块看")
print(f" {'':<12}{'视野内MSE':>12}{'慢变量u':>12}{'快变量w':>12}"
f"{'t=23':>12}{'t=47':>12}{'t=71':>12}")
print(fmt_row("全序列", cur_bi, cur_bi_u, cur_bi_w) + " 非流式，长度固定")
print(fmt_row("teacher", cur_tf, cur_tf_u, cur_tf_w) + " 上下文用真值，测试时不可得")
print(fmt_row("rollout0", cur_sf0, cur_sf0_u, cur_sf0_w))
print(f"\n 快变量 w：teacher {cur_tf_w[:T_TRAIN].mean():.5f} -> "
f"rollout {cur_sf0_w[:T_TRAIN].mean():.5f} "
f"放大 {cur_sf0_w[:T_TRAIN].mean() / cur_tf_w[:T_TRAIN].mean():.2f} 倍（恒定抬升）")
print(f" 慢变量 u：teacher {cur_tf_u[:T_TRAIN].mean():.5f} -> "
f"rollout {cur_sf0_u[:T_TRAIN].mean():.5f} "
f"放大 {cur_sf0_u[:T_TRAIN].mean() / cur_tf_u[:T_TRAIN].mean():.2f} 倍")
print(f" 慢变量误差 t=1 -> t=23 增长 "
f"{cur_sf0_u[23] / max(cur_sf0_u[1], 1e-12):.2f} 倍，"
f"翻倍长度 {doubling_length(cur_sf0_u, 1, T_TRAIN - 1):.2f} 帧")
print(f" 快变量误差 t=1 -> t=23 变化 "
f"{cur_sf0_w[23] / max(cur_sf0_w[1], 1e-12):.3f} 倍（不累积）")
print(f" 越过训练视野（t>={T_TRAIN}）后：慢变量是视野内的 "
f"{cur_sf0_u[T_TRAIN:].mean() / cur_sf0_u[:T_TRAIN].mean():.2f} 倍，"
f"快变量 {cur_sf0_w[T_TRAIN:].mean() / cur_sf0_w[:T_TRAIN].mean():.2f} 倍")
print(f" 模型对上下文的依赖强度 |W_ctx| = {ctx_gain(W_tf):.4f}")
# ── Q3：Self-Forcing 轮次 ─────────────────────────────────────
print(f"\n【Q3】Self-Forcing 轮次（每轮用自己 rollout 的上下文重训，步长 alpha={args.alpha}）")
print(f" {'':<12}{'视野内MSE':>12}{'慢变量u':>12}{'快变量w':>12}"
f"{'t=23':>12}{'t=47':>12}{'t=71':>12}")
W = W_tf.copy()
curves, us, ws = {"round0": cur_sf0}, {"round0": cur_sf0_u}, {"round0": cur_sf0_w}
rng_roll = np.random.default_rng(41)
rng_eval = np.random.default_rng(61)
n_sub = min(3000, args.n_train)
def eval_rollout(Wm, rng):
p = np.stack([sample_causal(Wm, rng, T_ROLL, c_test[i], ctx_ts=0)
for i in range(args.n_test)])
return p, mse_curve(p, seqs_test[:, :T_ROLL])
# 注意：这里故意只按训练视野内的指标接受更新。Self-Forcing 的论文强调用
# "视频级整体损失"，下面会看到只盯视野内会发生什么。
best_score = cur_sf0[:T_TRAIN].mean()
tail0 = cur_sf0[-1]
for r in range(1, args.rounds + 1):
Xs, Ys = [], []
for i in range(n_sub):
_, Xr, Yr = sample_causal(W, rng_roll, T_TRAIN, c_train[i], ctx_ts=0,
collect=True, x_true=seqs_train[i, :T_TRAIN])
Xs.append(Xr)
Ys.append(Yr)
W_new = fit_ridge(np.concatenate(Xs, 0), np.concatenate(Ys, 0))
# 折半线搜索：rollout 重训容易形成正反馈，只接受真的变好的更新
picked = None
a = args.alpha
for _ in range(5):
W_try = (1.0 - a) * W + a * W_new
rng_try = np.random.default_rng(900 + r * 10 + int(a * 100))
p_try, c_try = eval_rollout(W_try, rng_try)
sc = c_try[:T_TRAIN].mean()
if np.isfinite(sc) and sc < best_score:
picked = (W_try, p_try, c_try, sc, a)
best_score = sc
break
a *= 0.5
if picked is None:
print(f" round {r}: 折半线搜索到 alpha={a:.4f} 仍没有变好的更新，提前停止")
break
if picked[2][-1] > 50.0 * tail0:
print(f" round {r}: 外推末帧已发散到 {picked[2][-1]:.3g}（round0 的 "
f"{picked[2][-1] / tail0:.1f} 倍），停止")
curves[f"round{r}"] = picked[2]
us[f"round{r}"] = mse_curve(picked[1], seqs_test[:, :T_ROLL], USL)
ws[f"round{r}"] = mse_curve(picked[1], seqs_test[:, :T_ROLL], WSL)
break
W = picked[0]
curves[f"round{r}"] = picked[2]
us[f"round{r}"] = mse_curve(picked[1], seqs_test[:, :T_ROLL], USL)
ws[f"round{r}"] = mse_curve(picked[1], seqs_test[:, :T_ROLL], WSL)
print(fmt_row(f"round{r}", curves[f"round{r}"], us[f"round{r}"], ws[f"round{r}"])
+ f" alpha={picked[3]:.4f}")
print(f"\n {'':<12}{'视野内MSE':>12}{'慢变量u':>12}{'快变量w':>12}"
f"{'t=23':>12}{'t=47':>12}{'t=71':>12}")
print(fmt_row("teacher", cur_tf, cur_tf_u, cur_tf_w))
print(fmt_row("全序列", cur_bi, cur_bi_u, cur_bi_w))
keys = ["round0"] + [k for k in curves if k != "round0"]
for k in keys:
print(fmt_row(k, curves[k], us[k], ws[k]))
best = min(keys, key=lambda k: curves[k][:T_TRAIN].mean())
print(f"\n 最好一轮 {best}：快变量 {ws['round0'][:T_TRAIN].mean():.5f} -> "
f"{ws[best][:T_TRAIN].mean():.5f}"
f"（降 {100 * (1 - ws[best][:T_TRAIN].mean() / ws['round0'][:T_TRAIN].mean()):.1f}%），"
f"慢变量 {us['round0'][:T_TRAIN].mean():.5f} -> "
f"{us[best][:T_TRAIN].mean():.5f}"
f"（降 {100 * (1 - us[best][:T_TRAIN].mean() / us['round0'][:T_TRAIN].mean()):.1f}%）")
# ── Q4：上下文噪声必须与训练对齐 ──────────────────────────────
print("\n【Q4】上下文噪声 context_noise：训练用什么级，推理就得用什么级")
print(" （受控实验：上下文一律取真值帧，只改加噪级别，把曝光偏差排除掉）")
W_noisy = fit_ridge(*build_gt_dataset(seqs_train[:, :T_TRAIN], c_train,
np.random.default_rng(23), ctx_ts=250))
print(f" {'训练噪声步':<12}{'推理噪声步':<12}{'abar':>8}{'视野内MSE':>12}{'慢变量u':>12}")
ctx_rows = {}
for tag, Wm in [("clean(0)", W_tf), ("noisy(250)", W_noisy)]:
row = {}
for cts in args.ctx:
rng_e = np.random.default_rng(70)
p = np.stack([sample_causal(Wm, rng_e, T_TRAIN, c_test[i], ctx_ts=cts,
hist_true=seqs_test[i, :T_TRAIN], ctx_rng=rng_e)
for i in range(args.n_test)])
cc = mse_curve(p, seqs_test[:, :T_TRAIN])
cu = mse_curve(p, seqs_test[:, :T_TRAIN], USL)
row[cts] = cc
print(f" {tag:<12}{cts:<12}{abar_of(cts):>8.4f}{cc.mean():>12.5f}{cu.mean():>12.5f}")
ctx_rows[tag] = row
print(f" -> {tag} 的最优推理噪声步 = {min(row, key=lambda k: row[k].mean())}")
print("\n 同样的扫描放到自回归 rollout 上（此时上下文是模型自己的输出）")
row_sf = {}
for cts in args.ctx:
rng_e = np.random.default_rng(80)
p = np.stack([sample_causal(W, rng_e, T_TRAIN, c_test[i], ctx_ts=cts)
for i in range(args.n_test)])
cc = mse_curve(p, seqs_test[:, :T_TRAIN])
row_sf[cts] = cc
print(f" {'selfroll':<12}{cts:<12}{abar_of(cts):>8.4f}{cc.mean():>12.5f}"
f"{mse_curve(p, seqs_test[:, :T_TRAIN], USL).mean():>12.5f}")
print(f" -> 自回归 rollout 的最优推理噪声步 = {min(row_sf, key=lambda k: row_sf[k].mean())}")
# ── Q5：等效噪声级 + 训练时给上下文加噪换鲁棒性 ──────────────────
print("\n【Q5】曝光偏差的等效噪声级，以及训练时给上下文加噪能不能换来自回归的鲁棒性")
line_tf = np.array([ctx_rows["clean(0)"][k].mean() for k in args.ctx])
target = cur_sf0[:T_TRAIN].mean()
grid = np.array(args.ctx, dtype=float)
eq = float(np.interp(target, line_tf, grid)) if line_tf[-1] > target else float("nan")
print(f" 自回归 rollout 的视野内 MSE = {target:.5f}")
if np.isfinite(eq):
print(f" 在 Q4 那条『干净模型 + 加噪上下文』曲线上插值，它等效于给真值上下文加 "
f"{eq:.1f} 个时间步的噪声（abar 约 {abar_of(eq):.4f}）")
else:
print(" 超出 Q4 的扫描范围，无法插值出等效噪声级")
print(f"\n {'训练ctx噪声':<14}{'TF评测':>12}{'自回归(视野内)':>16}{'自回归t=71':>14}{'慢变量u':>12}")
q5 = {}
for k in args.ctx:
Wk = fit_ridge(*build_gt_dataset(seqs_train[:, :T_TRAIN], c_train,
np.random.default_rng(24), ctx_ts=k))
rng_a = np.random.default_rng(90)
p_tf = np.stack([sample_causal(Wk, rng_a, T_TRAIN, c_test[i],
hist_true=seqs_test[i, :T_TRAIN])
for i in range(args.n_test)])
rng_b2 = np.random.default_rng(91)
p_sf = np.stack([sample_causal(Wk, rng_b2, T_ROLL, c_test[i], ctx_ts=0)
for i in range(args.n_test)])
c_tfk = mse_curve(p_tf, seqs_test[:, :T_TRAIN]).mean()
c_sfk = mse_curve(p_sf, seqs_test[:, :T_ROLL])
c_sfu = mse_curve(p_sf, seqs_test[:, :T_ROLL], USL)
q5[k] = (c_tfk, c_sfk[:T_TRAIN].mean(), c_sfk[-1], c_sfu[:T_TRAIN].mean())
print(f" {k:<14}{c_tfk:>12.5f}{c_sfk[:T_TRAIN].mean():>16.5f}"
f"{c_sfk[-1]:>14.5f}{c_sfu[:T_TRAIN].mean():>12.5f}")
best_k = min(q5, key=lambda k: q5[k][1])
print(f" -> 自回归 rollout 最好的训练上下文噪声步 = {best_k}"
f"（视野内 {q5[best_k][1]:.5f}，对照 k=0 的 {q5[args.ctx[0]][1]:.5f}）")
cache_path = CACHE.replace(".npz", "_fast.npz") if args.fast else CACHE
np.savez(cache_path,
cur_bi=cur_bi, cur_bi_u=cur_bi_u, cur_bi_w=cur_bi_w,
cur_tf=cur_tf, cur_tf_u=cur_tf_u, cur_tf_w=cur_tf_w,
var_u=np.array([var_u]), var_w=np.array([var_w]),
**{f"cur_{k}": v for k, v in curves.items()},
**{f"u_{k}": v for k, v in us.items()},
**{f"w_{k}": v for k, v in ws.items()},
ctx_clean=np.array([ctx_rows["clean(0)"][k].mean() for k in args.ctx]),
ctx_noisy=np.array([ctx_rows["noisy(250)"][k].mean() for k in args.ctx]),
ctx_sf=np.array([row_sf[k].mean() for k in args.ctx]),
ctx_list=np.array(args.ctx),
q5_tf=np.array([q5[k][0] for k in args.ctx]),
q5_sf=np.array([q5[k][1] for k in args.ctx]),
q5_sf_end=np.array([q5[k][2] for k in args.ctx]),
q5_sf_u=np.array([q5[k][3] for k in args.ctx]),
eq_noise=np.array([eq]))
print(f"\n[cached] {cache_path}")
if __name__ == "__main__":
main()streaming_ledger.py
# -*- coding: utf-8 -*-
"""
streaming_ledger.py —— 流式生成 vs 全序列扩散的算力 / 显存账本（纯算术，numpy 只用来算）
跑法： python streaming_ledger.py
拓扑全部取自公开配置，不是估的：
Wan2.1-T2V-1.3B（wan/configs/wan_t2v_1_3B.py）：
dim=1536, ffn=8960, num_heads=12, num_layers=30, patch_size=(1,2,2), vae_stride=(4,8,8)
Self-Forcing（configs/self_forcing_dmd.yaml）：
image_or_video_shape=[1,21,16,60,104]
denoising_step_list=[1000,750,500,250] # 4 步
num_frame_per_block=3
Self-Forcing（pipeline/causal_inference.py）：
每个 block 走完 4 步之后，还要用 context_noise 时间步再跑一次 forward 刷新 KV cache
口径：MAC = 一次乘加（1 MAC = 2 FLOP）。时间按 H100 bf16 有效算力 400 TFLOPS 折算，
只算 transformer 主体，不含 VAE 解码与文本编码。
"""
import numpy as np
# ── 拓扑 ──────────────────────────────────────────────────────────
D = 1536 # dim
FFN = 8960 # ffn_dim
LAYERS = 30
HEADS = 12
D_HEAD = D // HEADS
N_TEXT = 512 # T5 文本 token 数（cross-attn 的 K/V 长度）
LAT_FRAMES = 21 # 潜空间帧数
LAT_H, LAT_W = 60, 104
PATCH = (1, 2, 2)
TOK_PER_FRAME = (LAT_H // PATCH[1]) * (LAT_W // PATCH[2]) # 1560
N_TOK = LAT_FRAMES * TOK_PER_FRAME # 32760
BLOCK = 3 # num_frame_per_block
N_STEPS = 4 # len(denoising_step_list)
EXTRA_KV_REFRESH = 1 # 每个 block 结束后刷新 KV cache 的那次 forward
PIXEL_PER_LATENT = 4 # vae_stride[0]，21 潜帧 ≈ 81 像素帧
FPS = 16
BYTES = 2 # bf16
EFF_TFLOPS = 400e12
def mac_per_layer(n_new, n_key):
"""一层 transformer 的 MAC。n_new = 本次参与计算的 token 数，n_key = 可见的 key 数。"""
attn = 2.0 * n_new * n_key * D # QK^T + AV
proj = 4.0 * n_new * D * D # q, k, v, o
ffn = 2.0 * n_new * D * FFN
cross = 2.0 * n_new * N_TEXT * D + 2.0 * n_new * D * D # 注意力 + q/o 投影
return attn + proj + ffn + cross
def gbyte(x_bytes):
return x_bytes / 1024.0 ** 3
def sec(mac):
return 2.0 * mac / EFF_TFLOPS
def main():
print("=" * 78)
print("拓扑：Wan2.1-T2V-1.3B + Self-Forcing 默认配置")
print(f" dim={D} ffn={FFN} layers={LAYERS} heads={HEADS} head_dim={D_HEAD}")
print(f" 潜空间 {LAT_FRAMES}x16x{LAT_H}x{LAT_W} patch={PATCH} -> "
f"每帧 {TOK_PER_FRAME} token，整段 {N_TOK} token")
print(f" block={BLOCK} 帧，每 block {N_STEPS} 步去噪 + {EXTRA_KV_REFRESH} 次 KV cache 刷新")
print(f" 21 潜帧 ≈ {LAT_FRAMES * PIXEL_PER_LATENT - 3} 像素帧 @ {FPS}fps ≈ "
f"{(LAT_FRAMES * PIXEL_PER_LATENT - 3) / FPS:.2f} 秒")
print(f" 时间按 {EFF_TFLOPS/1e12:.0f} TFLOPS 有效算力折算（1 MAC = 2 FLOP）")
print("=" * 78)
# ── A. 全序列扩散：4 步，每步整段双向 ────────────────────────────
mac_full_step = LAYERS * mac_per_layer(N_TOK, N_TOK)
mac_full = N_STEPS * mac_full_step
print("\n【A】全序列扩散（非流式，整段一起解）")
print(f" 单步 MAC {mac_full_step:.4e} 时间 {sec(mac_full_step)*1000:.1f} ms")
print(f" 4 步合计 MAC {mac_full:.4e} 时间 {sec(mac_full):.3f} s")
print(f" 第一帧延迟 必须等 {N_STEPS} 步全部算完 = {sec(mac_full):.3f} s")
attn_bytes_full = HEADS * N_TOK * N_TOK * BYTES
print(f" 注意力矩阵要是真存下来：{HEADS} 头 x {N_TOK}^2 x {BYTES}B = "
f"{gbyte(attn_bytes_full):.1f} GB（所以必须 FlashAttention）")
# ── B. 因果自回归 + KV cache ────────────────────────────────────
n_blocks = LAT_FRAMES // BLOCK
print(f"\n【B】因果自回归 + KV cache（{n_blocks} 个 block，每 block {BLOCK} 帧）")
print(f" {'block':>6}{'新token':>9}{'可见key':>9}{'单步MAC':>14}{'block合计MAC':>16}{'时间(ms)':>11}")
mac_blocks = []
for i in range(n_blocks):
n_new = BLOCK * TOK_PER_FRAME
n_key = (i + 1) * BLOCK * TOK_PER_FRAME
m_step = LAYERS * mac_per_layer(n_new, n_key)
n_fwd = N_STEPS + EXTRA_KV_REFRESH
# 4 步去噪：每步的 key 数就是 n_key（含当前 block 内部的因果可见部分）
m_block = n_fwd * m_step
mac_blocks.append(m_block)
print(f" {i:>6}{n_new:>9}{n_key:>9}{m_step:>14.4e}{m_block:>16.4e}"
f"{sec(m_block)*1000:>11.1f}")
mac_blocks = np.array(mac_blocks)
mac_causal = mac_blocks.sum()
print(f" 合计 MAC {mac_causal:.4e} 时间 {sec(mac_causal):.3f} s")
print(f" 第一帧延迟 = 第 0 个 block = {sec(mac_blocks[0])*1000:.1f} ms"
f"（比全序列快 {sec(mac_full)/sec(mac_blocks[0]):.1f} 倍）")
print(f" 稳态每个 block {sec(mac_blocks[-1])*1000:.1f} ms，"
f"实时预算 {BLOCK * PIXEL_PER_LATENT / FPS * 1000:.0f} ms "
f"-> 余量 {BLOCK * PIXEL_PER_LATENT / FPS / sec(mac_blocks[-1]):.2f} 倍")
# ── C. 总账对比 ────────────────────────────────────────────────
print(f"\n【C】总账（{LAT_FRAMES} 潜帧）")
print(f" 全序列扩散 MAC {mac_full:.4e} 时间 {sec(mac_full):.3f} s "
f"首帧延迟 {sec(mac_full):.3f} s")
print(f" 因果+KVcache MAC {mac_causal:.4e} 时间 {sec(mac_causal):.3f} s "
f"首帧延迟 {sec(mac_blocks[0]):.3f} s")
print(f" 总算力比 因果 / 全序列 = {mac_causal / mac_full:.3f}")
print(f" 首帧延迟比 全序列 / 因果 = {sec(mac_full) / sec(mac_blocks[0]):.2f}")
# ── D. KV cache 显存 ───────────────────────────────────────────
print("\n【D】KV cache 显存（bf16）")
kv_all = 2 * LAYERS * N_TOK * D * BYTES
print(f" 完整 21 帧 2 x {LAYERS} x {N_TOK} x {D} x {BYTES}B = {gbyte(kv_all):.3f} GB")
for w in [3, 6, 9, 12, 21]:
kv_w = 2 * LAYERS * w * TOK_PER_FRAME * D * BYTES
print(f" 滚动窗口 {w:>2} 帧 {gbyte(kv_w):.3f} GB （占完整缓存 {w/LAT_FRAMES*100:.0f}%）")
cross_cache = 2 * LAYERS * N_TEXT * D * BYTES
print(f" cross-attn 缓存（文本 {N_TEXT} token，只算一次）{gbyte(cross_cache):.3f} GB")
# ── E. 无限长：不滚动缓存会怎样 ─────────────────────────────────
print("\n【E】继续往下生成（不滚动缓存 vs 滚动窗口 9 帧）")
print(f" {'已生成潜帧':>10}{'无滚动:单block时间(ms)':>24}{'滚动9帧(ms)':>16}{'无滚动:累计GB':>16}")
cum = 0.0
for nlat in [21, 42, 84, 168, 336]:
i = nlat // BLOCK - 1
n_new = BLOCK * TOK_PER_FRAME
n_key = (i + 1) * BLOCK * TOK_PER_FRAME
m_noroll = (N_STEPS + EXTRA_KV_REFRESH) * LAYERS * mac_per_layer(n_new, n_key)
m_roll = (N_STEPS + EXTRA_KV_REFRESH) * LAYERS * mac_per_layer(
n_new, min(n_key, 9 * TOK_PER_FRAME))
cum = 2 * LAYERS * nlat * TOK_PER_FRAME * D * BYTES
print(f" {nlat:>10}{sec(m_noroll)*1000:>24.1f}{sec(m_roll)*1000:>16.1f}"
f"{gbyte(cum):>16.3f}")
# ── F. 出帧节奏 ────────────────────────────────────────────────
print("\n【F】出帧节奏（每 block 出 3 潜帧 = 12 像素帧）")
t_full = sec(mac_full)
t_blocks = np.cumsum(sec(mac_blocks))
print(f" 全序列：t={t_full:.3f} s 时一次性拿到全部 {LAT_FRAMES * PIXEL_PER_LATENT - 3} 帧")
for i, tb in enumerate(t_blocks):
print(f" 因果 ：t={tb:.3f} s 时拿到第 {(i+1)*BLOCK*PIXEL_PER_LATENT-3:>3} 像素帧")
print(f" 实时预算：{FPS} fps 下应该在 "
f"{np.arange(1, n_blocks+1)*BLOCK*PIXEL_PER_LATENT/FPS} 秒处出帧")
print("\n[cached] 数值直接被 make_figures.py 引用（本文件被 import 时用函数取）")
def ledger_numbers():
"""给 make_figures.py 用的结构化数值。"""
n_blocks = LAT_FRAMES // BLOCK
mac_blocks = []
for i in range(n_blocks):
n_new = BLOCK * TOK_PER_FRAME
n_key = (i + 1) * BLOCK * TOK_PER_FRAME
mac_blocks.append((N_STEPS + EXTRA_KV_REFRESH) * LAYERS * mac_per_layer(n_new, n_key))
mac_blocks = np.array(mac_blocks, dtype=float)
mac_full = N_STEPS * LAYERS * mac_per_layer(N_TOK, N_TOK)
return dict(
mac_full=float(mac_full), mac_causal=float(mac_blocks.sum()),
mac_blocks=mac_blocks, t_full=sec(mac_full), t_blocks=np.cumsum(sec(mac_blocks)),
kv_all_gb=gbyte(2 * LAYERS * N_TOK * D * BYTES),
kv_per_frame_gb=gbyte(2 * LAYERS * TOK_PER_FRAME * D * BYTES),
n_blocks=n_blocks, block=BLOCK, lat_frames=LAT_FRAMES,
tok_per_frame=TOK_PER_FRAME, n_tok=N_TOK,
attn_gb_if_materialized=gbyte(HEADS * N_TOK * N_TOK * BYTES),
pixel_frames=LAT_FRAMES * PIXEL_PER_LATENT - 3, fps=FPS,
)
if __name__ == "__main__":
main()
print()
for k, v in ledger_numbers().items():
if not isinstance(v, np.ndarray):
print(f" {k:>28} = {v}")make_figures.py
# -*- coding: utf-8 -*-
"""
make_figures.py —— 本文配图（matplotlib，中文字体 PingFang SC）
跑法： python make_figures.py
所有数字都从 forcing_lab_cache.npz（forcing_lab.py 实跑产出）和
streaming_ledger.ledger_numbers() 里取，不手写，避免图文漂移。
注意：matplotlib 的 mathtext 标签一律写 raw 字符串，且反斜杠后只跟字母
（beta 这种字母命令没问题，逗号、百分号等非字母不要跟在反斜杠后面）。
"""
import os
import numpy as np
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle, FancyArrow
HERE = os.path.dirname(os.path.abspath(__file__))
FIGDIR = os.path.join(os.path.dirname(HERE), "figures")
CACHE = os.path.join(HERE, "forcing_lab_cache.npz")
import forcing_lab as FL
import streaming_ledger as SL
plt.rcParams["font.sans-serif"] = ["PingFang SC", "Arial Unicode MS", "Heiti TC"]
plt.rcParams["axes.unicode_minus"] = False
# 颜色常量（写图注前先对照这里，别凭印象写颜色）
C_BLUE = "#1f6feb" # 自回归 rollout
C_TEAL = "#0f9b8e" # teacher forcing
C_ORANGE = "#e07b00" # 全序列扩散
C_RED = "#c0392b" # 慢变量 u / 恶化
C_GREEN = "#2d8a4e" # 快变量 w / 改善
C_GRAY = "#8a8a8a" # 参照线、未生成
C_DARK = "#3a3a3a" # 噪声最重的格子
C_LIGHT = "#f2f2f2" # 干净的格子
def load():
return dict(np.load(CACHE, allow_pickle=True))
# ══════════════════════════ 图 1：三种范式的噪声级与因果结构 ══════════════════════════
def _box(ax, x, y, w, h, level, edge=C_GRAY, lw=0.8, ls="-"):
"""level: 1 = 干净（浅），0 = 纯噪声（深）"""
facecolor = str(max(0.0, min(1.0, float(level))))
ax.add_patch(Rectangle((x, y), w, h, facecolor=facecolor, edgecolor=edge,
linewidth=lw, linestyle=ls, zorder=2))
def fig_paradigm(path):
n = 12
bw, bh = 0.86, 0.62
fig, axes = plt.subplots(3, 1, figsize=(11.0, 6.6))
fig.subplots_adjust(left=0.06, right=0.97, top=0.93, bottom=0.07, hspace=0.34)
# (a) 全序列扩散：整段一起降噪，帧与帧之间双向可见
ax = axes[0]
steps_lv = [0.06, 0.30, 0.62, 0.92]
for r, lv in enumerate(steps_lv):
y = (len(steps_lv) - 1 - r) * 1.0
for i in range(n):
_box(ax, i * 1.07, y, bw, bh, lv)
ax.annotate("", xy=(n * 1.07 - 0.1, y + bh / 2), xytext=(0.05, y + bh / 2),
arrowprops=dict(arrowstyle="<->", color=C_ORANGE, lw=1.8))
ax.text(-0.35, y + bh / 2, r"去噪步 %d" % (r + 1), ha="right", va="center",
fontsize=10, color=C_DARK)
ax.text(n * 1.07 / 2, len(steps_lv) + 0.15,
"所有帧同一个噪声级，一起降；双向注意力，帧间互相可见",
ha="center", fontsize=11, color=C_ORANGE)
ax.text(n * 1.07 / 2, -0.55, "代价：本图需 4 步整段去噪才交付第一帧；可用长度由架构与训练共同限制",
ha="center", fontsize=10, color=C_GRAY)
# (b) teacher forcing：真值上下文，只看过去
ax = axes[1]
y = 0.0
cur = 7
for i in range(n):
if i < cur:
_box(ax, i * 1.07, y, bw, bh, 0.95, edge=C_TEAL, lw=1.2)
elif i == cur:
_box(ax, i * 1.07, y, bw, bh, 0.10)
else:
_box(ax, i * 1.07, y, bw, bh, 1.0, edge=C_GRAY, lw=0.8, ls="--")
for i in range(cur):
ax.annotate("", xy=(cur * 1.07 + bw * 0.5, y + bh * 0.45),
xytext=(i * 1.07 + bw * 0.5, y + bh * 0.45),
arrowprops=dict(arrowstyle="->", color=C_TEAL, lw=1.2,
connectionstyle="arc3,rad=-0.25"))
ax.text(-0.35, y + bh / 2, "训练时", ha="right", va="center", fontsize=10, color=C_DARK)
ax.text(n * 1.07 / 2, y + 1.15,
"上下文永远取真值帧（浅色），只有当前帧（深色）带噪声",
ha="center", fontsize=11, color=C_TEAL)
ax.text(n * 1.07 / 2, y - 0.52,
"代价：推理时真值帧不存在，上下文换成模型自己的输出 —— 这就是曝光偏差",
ha="center", fontsize=10, color=C_GRAY)
# (c) Self-Forcing 风格的生成上下文：自己的输出进 KV cache，逐 block 走
ax = axes[2]
done, blk = 6, 3
for i in range(n):
if i < done:
_box(ax, i * 1.07, y, bw, bh, 0.95, edge=C_BLUE, lw=1.2)
elif i < done + blk:
_box(ax, i * 1.07, y, bw, bh, 0.10, edge=C_BLUE, lw=1.2)
else:
_box(ax, i * 1.07, y, bw, bh, 1.0, edge=C_GRAY, lw=0.8, ls="--")
for i in range(done):
ax.annotate("", xy=(done * 1.07 + bw * 0.5, y + bh * 0.45),
xytext=(i * 1.07 + bw * 0.5, y + bh * 0.45),
arrowprops=dict(arrowstyle="->", color=C_BLUE, lw=1.2,
connectionstyle="arc3,rad=-0.25"))
ax.add_patch(Rectangle((-0.05, y - 0.42), done * 1.07 + 0.05, 0.26,
facecolor=C_BLUE, alpha=0.15, edgecolor=C_BLUE, lw=1.0))
ax.text(done * 1.07 / 2, y - 0.29, "KV cache（已生成的帧，不再重算）",
ha="center", va="center", fontsize=9.5, color=C_BLUE)
ax.text(-0.35, y + bh / 2, "推理时", ha="right", va="center", fontsize=10, color=C_DARK)
ax.text(n * 1.07 / 2, y + 1.15,
"自回归推理：历史为已生成的干净帧，当前 block 从噪声降到干净",
ha="center", fontsize=11, color=C_BLUE)
ax.text(n * 1.07 / 2, y - 0.82,
"Self-Forcing 训练也用自己的 rollout 上下文，减轻训练与推理的分布差异",
ha="center", fontsize=10, color=C_GRAY)
for ax in axes:
ax.set_xlim(-2.4, n * 1.07 + 0.3)
ax.set_ylim(-1.05, len(steps_lv) + 0.45 if ax is axes[0] else 1.55)
ax.axis("off")
axes[0].set_title("三种上下文组织示意：可见范围、噪声级与历史来源",
fontsize=13, pad=10)
fig.savefig(path, dpi=140)
plt.close(fig)
# ══════════════════════════ 图 2：误差随帧号怎么长 ══════════════════════════
def fig_rollout(d, path):
T = FL.T_TRAIN
fig, axes = plt.subplots(1, 2, figsize=(12.4, 4.7))
ax = axes[0]
xs = np.arange(T)
ax.plot(xs, d["cur_bi"][:T], color=C_ORANGE, lw=2.0, label="全序列扩散（非流式）")
ax.plot(xs, d["cur_tf"][:T], color=C_TEAL, lw=2.0, label="teacher forcing（真值上下文）")
xr = np.arange(len(d["cur_round0"]))
ax.plot(xr, d["cur_round0"], color=C_BLUE, lw=2.0, label="自回归 rollout（自己的输出）")
ax.axvline(T - 0.5, color=C_GRAY, ls="--", lw=1.2)
ax.text(T - 0.4, ax.get_ylim()[1] * 0.55, "训练视野\nT=24", fontsize=9.5, color=C_GRAY)
ax.set_yscale("log")
ax.set_xlabel("帧号 t")
ax.set_ylabel("每维平均平方误差（对数轴）")
ax.set_title("(a) 三种范式的逐帧误差", fontsize=12)
ax.legend(fontsize=9.5, loc="lower right")
ax.grid(alpha=0.25, ls=":")
ax = axes[1]
ax.plot(xr, d["u_round0"], color=C_RED, lw=2.0, label=r"慢变量 u（会被积分记住）")
ax.plot(xr, d["w_round0"], color=C_GREEN, lw=2.0, label=r"快变量 w（收缩模态）")
ax.axvline(T - 0.5, color=C_GRAY, ls="--", lw=1.2)
ax.text(T - 0.4, 0.10, "训练视野\nT=24", fontsize=9.5, color=C_GRAY)
ax.set_yscale("log")
ax.set_xlabel("帧号 t")
ax.set_ylabel("每维平均平方误差（对数轴）")
ax.set_title("(b) 自回归 rollout 拆成快慢两块", fontsize=12)
ax.legend(fontsize=9.5, loc="upper left")
ax.grid(alpha=0.25, ls=":")
fig.suptitle("本玩具系统中，慢变量的误差在越过训练视野后明显增长",
fontsize=13)
fig.tight_layout(rect=[0, 0, 0.99, 0.94])
fig.savefig(path, dpi=140)
plt.close(fig)
# ══════════════════════════ 图 3：Self-Forcing 轮次的取舍 ══════════════════════════
def fig_rounds(d, path):
keys = sorted([k[4:] for k in d if k.startswith("cur_round")],
key=lambda s: int(s.replace("round", "")))
lbl = [("round0" if k == "round0" else k) for k in keys]
inh = [float(d[f"cur_{k}"][:FL.T_TRAIN].mean()) for k in keys]
tail = [float(d[f"cur_{k}"][-1]) for k in keys]
x = np.arange(len(keys))
fig, ax = plt.subplots(figsize=(9.6, 4.6))
ax2 = ax.twinx()
b1 = ax.bar(x - 0.19, inh, 0.36, color=C_BLUE, label="训练视野内 MSE（左轴）")
b2 = ax2.bar(x + 0.19, tail, 0.36, color=C_RED, label="外推第 72 帧 MSE（右轴，对数）")
for b in b1:
ax.text(b.get_x() + b.get_width() / 2, b.get_height(),
f"{b.get_height():.3f}", ha="center", va="bottom", fontsize=9)
for b in b2:
ax2.text(b.get_x() + b.get_width() / 2, b.get_height(),
f"{b.get_height():.2f}", ha="center", va="bottom", fontsize=9,
color=C_RED)
ax.set_xticks(x)
ax.set_xticklabels([k.replace("round", "第 ") + " 轮" if k != "round0" else "第 0 轮"
for k in keys])
ax.set_ylabel("训练视野内 MSE", color=C_BLUE)
ax2.set_ylabel("外推第 72 帧 MSE", color=C_RED)
ax2.set_yscale("log")
ax.set_ylim(0, max(inh) * 1.35)
ax2.set_ylim(min(tail) * 0.5, max(tail) * 3.0)
ax.grid(alpha=0.25, ls=":", axis="y")
ax.set_title("玩具 DAgger 式重训：视野内改善，长程误差增加（非论文 Self-Forcing）",
fontsize=13)
h1, l1 = ax.get_legend_handles_labels()
h2, l2 = ax2.get_legend_handles_labels()
ax.legend(h1 + h2, l1 + l2, fontsize=10, loc="upper left")
fig.tight_layout()
fig.savefig(path, dpi=140)
plt.close(fig)
# ══════════════════════════ 图 4：上下文噪声的取舍 ══════════════════════════
def fig_context(d, path):
ks = [int(v) for v in d["ctx_list"]]
ab = [FL.abar_of(k) for k in ks]
tf = [float(v) for v in d["q5_tf"]]
end = [float(v) for v in d["q5_sf_end"]]
inh = [float(v) for v in d["q5_sf"]]
fig, axes = plt.subplots(1, 2, figsize=(12.4, 4.7))
ax = axes[0]
ax.plot(ks, d["ctx_clean"], "o-", color=C_TEAL, lw=2.0,
label="干净模型 + 推理时给上下文加噪")
ax.axhline(float(d["cur_round0"][:FL.T_TRAIN].mean()), color=C_BLUE, ls="--", lw=1.6,
label="自回归 rollout 的实际水平")
ax.axvline(float(d["eq_noise"][0]), color=C_RED, ls=":", lw=1.8)
ax.text(float(d["eq_noise"][0]) + 8, ax.get_ylim()[1] * 0.62,
f"等效噪声级\n约 {float(d['eq_noise'][0]):.0f} 步", fontsize=9.5, color=C_RED)
ax.set_xlabel("上下文噪声时间步")
ax.set_ylabel("视野内 MSE")
ax.set_title("(a) 曝光偏差有多大：换算成等效噪声级", fontsize=12)
ax.legend(fontsize=9.5)
ax.grid(alpha=0.25, ls=":")
ax = axes[1]
ax.plot(ks, inh, "o-", color=C_BLUE, lw=2.0, label="训练视野内 MSE（精度）")
ax.set_xlabel("训练时给上下文加的噪声时间步 k")
ax.set_ylabel("训练视野内 MSE", color=C_BLUE)
ax2 = ax.twinx()
ax2.plot(ks, end, "s-", color=C_RED, lw=2.0, label="外推第 72 帧 MSE（稳定性）")
ax2.set_ylabel("外推第 72 帧 MSE", color=C_RED)
ax.set_title("(b) 训练时给上下文加噪：精度换稳定性", fontsize=12)
h1, l1 = ax.get_legend_handles_labels()
h2, l2 = ax2.get_legend_handles_labels()
ax.legend(h1 + h2, l1 + l2, fontsize=9.5, loc="center right")
ax.grid(alpha=0.25, ls=":")
fig.suptitle(f"上下文噪声：abar 从 {ab[0]:.2f} 降到 {ab[-1]:.2f}，"
f"本实验中增加上下文噪声可降低长程误差，但影响视野内精度", fontsize=13)
fig.tight_layout(rect=[0, 0, 0.99, 0.93])
fig.savefig(path, dpi=140)
plt.close(fig)
# ══════════════════════════ 图 5：流式账本 ══════════════════════════
def fig_ledger(path):
L = SL.ledger_numbers()
fig, axes = plt.subplots(1, 3, figsize=(15.6, 4.6))
# (a) 出帧节奏
ax = axes[0]
pix = (np.arange(1, L["n_blocks"] + 1) * L["block"] * 4 - 3)
ax.step(np.concatenate([[0], L["t_blocks"]]),
np.concatenate([[0], pix]), where="post", color=C_BLUE, lw=2.2,
label="因果自回归 + KV cache")
ax.plot([L["t_full"], L["t_full"]], [0, L["pixel_frames"]], color=C_ORANGE, lw=2.2,
label="全序列扩散（一次性出全部帧）")
tt = np.linspace(0, L["pixel_frames"] / L["fps"], 50)
ax.plot(tt, tt * L["fps"], color=C_GRAY, ls="--", lw=1.4, label="实时预算 16 fps")
ax.set_xlabel("时间 (s)")
ax.set_ylabel("已生成的像素帧")
ax.set_title("(a) 出帧节奏", fontsize=12)
ax.legend(fontsize=9.5)
ax.grid(alpha=0.25, ls=":")
ax.set_ylim(0, L["pixel_frames"] * 1.05)
# (b) KV cache 显存
ax = axes[1]
nf = np.arange(1, 169)
ax.plot(nf, nf * L["kv_per_frame_gb"], color=C_BLUE, lw=2.2, label="KV cache（不滚动）")
ax.axhline(9 * L["kv_per_frame_gb"], color=C_GREEN, ls="--", lw=1.6,
label="滚动窗口 9 帧的上限")
ax.axhline(L["kv_all_gb"], color=C_GRAY, ls=":", lw=1.4, label="21 帧整段")
ax.set_xlabel("已生成的潜帧数")
ax.set_ylabel("KV cache 显存 (GB)")
ax.set_title("(b) 不滚动缓存，显存线性涨", fontsize=12)
ax.legend(fontsize=9.5)
ax.grid(alpha=0.25, ls=":")
# (c) 单 block 计算时间
ax = axes[2]
nkey = np.arange(1, 169) * SL.TOK_PER_FRAME
t_noroll = 2.0 * (SL.N_STEPS + SL.EXTRA_KV_REFRESH) * SL.LAYERS * np.array(
[SL.mac_per_layer(SL.BLOCK * SL.TOK_PER_FRAME, k) for k in nkey]) / SL.EFF_TFLOPS
t_roll = 2.0 * (SL.N_STEPS + SL.EXTRA_KV_REFRESH) * SL.LAYERS * np.array(
[SL.mac_per_layer(SL.BLOCK * SL.TOK_PER_FRAME,
min(k, 9 * SL.TOK_PER_FRAME)) for k in nkey]) / SL.EFF_TFLOPS
ax.plot(np.arange(1, 169), t_noroll * 1000, color=C_RED, lw=2.2, label="不滚动（线性变慢）")
ax.plot(np.arange(1, 169), t_roll * 1000, color=C_GREEN, lw=2.2, label="滚动窗口 9 帧（恒定）")
ax.axhline(SL.BLOCK * 4 / SL.FPS * 1000, color=C_GRAY, ls="--", lw=1.4,
label="实时预算 750 ms / block")
ax.set_xlabel("已生成的潜帧数")
ax.set_ylabel("单个 block 的计算时间 (ms)")
ax.set_title("(c) 不滚动缓存，每个 block 越来越慢", fontsize=12)
ax.legend(fontsize=9.5)
ax.grid(alpha=0.25, ls=":")
fig.suptitle("400 TFLOPS 假设下的算术估算：首帧交付更早；缓存与历史计算仍有代价",
fontsize=13)
fig.tight_layout(rect=[0, 0, 0.99, 0.93])
fig.savefig(path, dpi=140)
plt.close(fig)
def main():
os.makedirs(FIGDIR, exist_ok=True)
d = load()
fig_paradigm(os.path.join(FIGDIR, "paradigm.png"))
fig_rollout(d, os.path.join(FIGDIR, "rollout_error.png"))
fig_rounds(d, os.path.join(FIGDIR, "selfforcing_rounds.png"))
fig_context(d, os.path.join(FIGDIR, "context_noise.png"))
fig_ledger(os.path.join(FIGDIR, "streaming_ledger.png"))
for f in sorted(os.listdir(FIGDIR)):
p = os.path.join(FIGDIR, f)
print(f" {f:<28} {os.path.getsize(p)/1024:.0f} KB")
if __name__ == "__main__":
main()
评论 (0)