{ "cells": [ { "cell_type": "markdown", "id": "196a0a09", "metadata": {}, "source": "# Lesson 65: Video Diffusion and Generation\n\nLesson 62's diffusion models learn to reverse a noising process one still image at a time. This lesson extends that same denoising recipe to video: generate a whole clip at once while keeping it temporally consistent — the new failure mode a video generator has that an image generator never faces is **flicker**, where independently-denoised frames disagree slightly from one frame to the next, so a shape's position or appearance jitters instead of moving smoothly. The fix has two parts: architecture (a temporal-attention layer, plus a frame-position embedding so the network knows *which* frame of the clip it is denoising) and sampling (correlating, rather than fully independently drawing, the random noise injected into each frame at every step). Both parts turn out to matter, and this lesson shows why with a concrete, measurable failure mode along the way." }, { "cell_type": "code", "id": "a765a545", "source": "import numpy as np\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport matplotlib.pyplot as plt", "metadata": {}, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "5854718e", "source": "## A tiny synthetic video dataset\n\nReuse Lesson {{lesson:diffusion_models}}'s shape generator, but now give the shape a constant velocity: each clip is a short sequence of frames in which a plus or circle translates by a fixed `(vx, vy)` offset per frame. \"Smooth, consistent motion\" is exactly the property a good video generator needs to reproduce, and exactly what flicker destroys — so a dataset where motion is literally constant velocity by construction makes it easy to check, quantitatively, whether a generated clip moved smoothly or not.", "metadata": {} }, { "cell_type": "code", "id": "a4d35160", "source": "SIZE = 16\nT = 60\nNF = 4 # frames per clip\n\nbetas = torch.linspace(1e-4, 0.02, T)\nalphas = 1.0 - betas\nalpha_bars = torch.cumprod(alphas, dim=0)\n\ndef make_image(shape_type, cx, cy, size=SIZE):\n img = np.zeros((size, size), dtype=np.float32)\n if shape_type == 'plus':\n img[cy-1:cy+2, cx-3:cx+4] = 1.0\n img[cy-3:cy+4, cx-1:cx+2] = 1.0\n else:\n yy, xx = np.mgrid[0:size, 0:size]\n img[((xx-cx)**2 + (yy-cy)**2) <= 9] = 1.0\n return img * 2 - 1 # scale to [-1, 1], same convention as Lesson {{lesson:diffusion_models}}\n\ndef make_clip(rng, n_frames=NF, size=SIZE):\n shape_type = rng.choice(['plus', 'circle'])\n vx, vy = 0, 0\n while vx == 0 and vy == 0: # exclude the degenerate \"doesn't move\" case\n vx, vy = rng.integers(-1, 2), rng.integers(-1, 2)\n cx0, cy0 = rng.integers(7, 9), rng.integers(7, 9)\n frames = [make_image(shape_type, cx0 + vx * i, cy0 + vy * i, size) for i in range(n_frames)]\n return np.stack(frames, axis=0) # (NF, H, W)\n\ndef make_clip_dataset(rng, n, n_frames=NF, size=SIZE):\n return np.array([make_clip(rng, n_frames, size) for _ in range(n)], dtype=np.float32)\n\nrng = np.random.default_rng(3)\nclips = make_clip_dataset(rng, 500)\nclips_t = torch.tensor(clips) # (N, NF, H, W)\n\nfig, axes = plt.subplots(2, NF, figsize=(7, 3.5))\nfor row in range(2):\n for i in range(NF):\n axes[row, i].imshow(clips[row, i], cmap='gray'); axes[row, i].axis('off')\n if row == 0:\n axes[row, i].set_title(f't={i}', fontsize=8)\nplt.suptitle('Two training clips: a shape moving at constant velocity')\nplt.show()", "metadata": {}, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "45f28f4b", "source": "## From a single-frame denoiser to a clip-level denoiser\n\n`DenoiseNet` below is Lesson {{lesson:diffusion_models}}'s own per-frame denoiser, unchanged — it will serve as the frame-independent baseline. `VideoDenoiseNet` adds two things on top of the same spatial backbone: a **temporal-attention** layer (Lesson {{lesson:attention_mechanism}}'s attention mechanism, repurposed so that instead of mixing information across *positions in a sequence*, it mixes information across *frames at the same pixel location* — every spatial location's feature vector becomes a query that attends to that same location's feature vectors in the other `NF` frames), and a **frame-position embedding** (the same sinusoidal-embedding idea as the diffusion timestep `t`, but indexing \"which of the `NF` frames is this\" instead of \"how noisy is this\"). The embedding matters more than it looks: without it, every frame gets an *identical* input representation whenever the noise happens to be correlated across frames, so a purely symmetric network has no way to tell frame 0 from frame 3 and nothing stops it from collapsing to the trivial, degenerate \"solution\" of generating the exact same frame four times (zero flicker, but also zero motion). The frame-position embedding breaks that symmetry, giving the temporal-attention layers an actual position to reason about.\n\nReal video diffusion models usually **factorize** spatial and temporal mixing into two separate passes, rather than one joint attention over every pixel of every frame at once, since joint attention's cost grows with the *product* of spatial and temporal length; this lesson's architecture already reflects that factorization, with ordinary spatial convolutions handling within-frame structure and `TemporalAttention` handling across-frame structure separately.", "metadata": {} }, { "cell_type": "code", "id": "81804666", "source": "class TimeEmbedding(nn.Module):\n def __init__(self, dim=32):\n super().__init__()\n self.dim = dim\n\n def forward(self, t):\n half = self.dim // 2\n freqs = torch.exp(-np.log(10000) * torch.arange(half).float() / half)\n args = t[:, None].float() * freqs[None, :]\n return torch.cat([torch.sin(args), torch.cos(args)], dim=-1)\n\nclass DenoiseNet(nn.Module):\n \"\"\"Lesson {{lesson:diffusion_models}}'s per-frame denoiser, unmodified -- the frame-independent baseline.\"\"\"\n def __init__(self, ch=48, time_dim=32):\n super().__init__()\n self.time_embed = TimeEmbedding(time_dim)\n self.time_mlp = nn.Linear(time_dim, ch)\n self.conv1 = nn.Conv2d(1, ch, 3, padding=1)\n self.conv2 = nn.Conv2d(ch, ch, 3, padding=1)\n self.conv3 = nn.Conv2d(ch, 1, 3, padding=1)\n\n def forward(self, x, t):\n temb = self.time_mlp(self.time_embed(t))[:, :, None, None]\n h = F.relu(self.conv1(x) + temb)\n h = F.relu(self.conv2(h) + temb)\n return self.conv3(h)\n\nclass TemporalAttention(nn.Module):\n \"\"\"Lesson {{lesson:attention_mechanism}}'s attention, applied along the frame axis: each spatial location's\n feature vector attends across that same location in all NF frames.\"\"\"\n def __init__(self, ch, n_frames):\n super().__init__()\n self.n_frames = n_frames\n self.q = nn.Linear(ch, ch)\n self.k = nn.Linear(ch, ch)\n self.v = nn.Linear(ch, ch)\n self.scale = ch ** -0.5\n\n def forward(self, h):\n BF, ch, Hh, Ww = h.shape # BF = batch_size * n_frames\n B = BF // self.n_frames\n x = h.view(B, self.n_frames, ch, Hh, Ww).permute(0, 3, 4, 1, 2).reshape(B * Hh * Ww, self.n_frames, ch)\n q, k, v = self.q(x), self.k(x), self.v(x)\n attn = torch.softmax((q @ k.transpose(-1, -2)) * self.scale, dim=-1)\n out = attn @ v\n out = out.reshape(B, Hh, Ww, self.n_frames, ch).permute(0, 3, 4, 1, 2).reshape(BF, ch, Hh, Ww)\n return h + out # residual, same pattern as Lesson {{lesson:transformer_architecture}}'s encoder block\n\nclass VideoDenoiseNet(nn.Module):\n def __init__(self, ch=48, time_dim=32, n_frames=NF):\n super().__init__()\n self.n_frames = n_frames\n self.time_embed = TimeEmbedding(time_dim)\n self.time_mlp = nn.Linear(time_dim, ch)\n self.frame_embed = nn.Embedding(n_frames, ch) # tells the network WHICH frame of the clip this is\n self.conv1 = nn.Conv2d(1, ch, 3, padding=1)\n self.temporal_attn1 = TemporalAttention(ch, n_frames)\n self.conv2 = nn.Conv2d(ch, ch, 3, padding=1)\n self.temporal_attn2 = TemporalAttention(ch, n_frames)\n self.conv3 = nn.Conv2d(ch, 1, 3, padding=1)\n\n def forward(self, x, t, frame_idx):\n # x: (B*NF, 1, H, W), t: (B*NF,), frame_idx: (B*NF,) with values in [0, NF)\n temb = self.time_mlp(self.time_embed(t))[:, :, None, None]\n femb = self.frame_embed(frame_idx)[:, :, None, None]\n h = F.relu(self.conv1(x) + temb + femb)\n h = self.temporal_attn1(h)\n h = F.relu(self.conv2(h) + temb + femb)\n h = self.temporal_attn2(h)\n return self.conv3(h)", "metadata": {}, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "a6c4358f", "source": "## Training: frame-independent vs. clip-aware\n\nBoth models are trained with the identical noise-prediction objective as Lesson {{lesson:diffusion_models}}. The baseline is trained on every frame of every clip, pooled together as if they were unrelated still images — it never sees two frames of the same clip together, so it has no way to learn any relationship between them. The video model is trained clip-by-clip: all `NF` frames of a clip are denoised together, at the *same* randomly-sampled noise level `t`, with each frame also given its own `frame_idx` so `VideoDenoiseNet` can learn frame-position-dependent content.", "metadata": {} }, { "cell_type": "code", "id": "94ae59bb", "source": "def train_baseline(clips_t, epochs=800, bs=64, seed=0):\n torch.manual_seed(seed)\n model = DenoiseNet()\n opt = torch.optim.Adam(model.parameters(), lr=0.001)\n frames_flat = clips_t.reshape(-1, 1, SIZE, SIZE) # every frame of every clip, treated as an independent image\n n = frames_flat.shape[0]\n for epoch in range(epochs):\n idx = np.random.default_rng(epoch).permutation(n)[:bs]\n x0 = frames_flat[idx]\n t = torch.randint(0, T, (x0.shape[0],))\n noise = torch.randn_like(x0)\n ab = alpha_bars[t][:, None, None, None]\n x_t = torch.sqrt(ab) * x0 + torch.sqrt(1 - ab) * noise\n pred = model(x_t, t)\n loss = F.mse_loss(pred, noise)\n opt.zero_grad(); loss.backward(); opt.step()\n return model, loss.item()\n\ndef train_video(clips_t, epochs=2500, bs_clips=16, seed=0):\n torch.manual_seed(seed)\n model = VideoDenoiseNet()\n opt = torch.optim.Adam(model.parameters(), lr=0.001)\n n = clips_t.shape[0]\n frame_idx_single = torch.arange(NF)\n for epoch in range(epochs):\n idx = np.random.default_rng(epoch).permutation(n)[:bs_clips]\n B = idx.shape[0]\n x0 = clips_t[idx].reshape(-1, 1, SIZE, SIZE) # (B*NF, 1, H, W)\n frame_idx = frame_idx_single.repeat(B) # [0,1,2,3, 0,1,2,3, ...]\n t_clip = torch.randint(0, T, (B,))\n t = t_clip.repeat_interleave(NF) # same noise LEVEL across a clip's frames\n noise = torch.randn_like(x0) # still an independent noise DRAW per frame\n ab = alpha_bars[t][:, None, None, None]\n x_t = torch.sqrt(ab) * x0 + torch.sqrt(1 - ab) * noise\n pred = model(x_t, t, frame_idx)\n loss = F.mse_loss(pred, noise)\n opt.zero_grad(); loss.backward(); opt.step()\n return model, loss.item()\n\nbaseline_model, bl_loss = train_baseline(clips_t)\nvideo_model, vid_loss = train_video(clips_t)\nprint(f'baseline (frame-independent) final loss: {bl_loss:.3f}')\nprint(f'video (temporal-attention) final loss: {vid_loss:.3f}')", "metadata": {}, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "ed89ee7d", "source": "## The flicker failure mode\n\nSampling mirrors training's asymmetry. The baseline reverses the diffusion process for each frame completely independently — its own starting noise, its own fresh random noise injected at every one of the `T` reverse steps — so nothing connects what frame 0 looks like to what frame 1 looks like, beyond both being *some* plausible shape. The video model's sampler instead draws **mostly-shared noise**: at every step, a single shared noise tensor is mixed with a small independent-per-frame component (`SHARE = 0.9` below), rather than being either fully independent (no correlation at all) or fully shared (which, perhaps surprisingly, is *also* wrong — with literally identical noise and no frame-position embedding, every frame receives an identical input at every step and the reverse process provably produces four pixel-identical frames: zero flicker, but also zero motion, a degenerate \"solution\" that technically minimizes flicker metrics while completely failing at the actual task. Try it: remove the frame-position embedding and set `SHARE = 1.0`, and check that every generated frame is bit-for-bit identical). Mostly-shared noise keeps the frames' trajectories correlated enough to suppress flicker while leaving enough independent signal, combined with the frame-position embedding, for the network to still generate genuinely different (and smoothly evolving) content per frame.\n\nTo measure flicker numerically, track each frame's shape centroid and compute the **jerk**: the mean magnitude of the frame-to-frame change in velocity (the second difference of centroid position). Real clips move at constant velocity by construction, so their jerk is exactly 0; a generator that produces smooth, physically plausible motion should score close to 0 too, while one that effectively redraws an unrelated shape each frame will have large, essentially random jerk. (Jerk alone can't distinguish \"smooth motion\" from \"no motion at all\" — a frozen clip also scores 0 — so the code below also reports each clip's mean frame-to-frame **speed**, to confirm the video model is actually moving the shape, not just holding it still.)", "metadata": {} }, { "cell_type": "code", "id": "df3aad62", "source": "@torch.no_grad()\ndef sample_baseline_clip(model, seed, n_frames=NF):\n \"\"\"Each frame is its own completely independent reverse-diffusion run.\"\"\"\n torch.manual_seed(seed)\n x = torch.randn(n_frames, 1, SIZE, SIZE)\n for t in reversed(range(T)):\n t_batch = torch.full((n_frames,), t, dtype=torch.long)\n pred_noise = model(x, t_batch)\n alpha_t, alpha_bar_t, beta_t = alphas[t], alpha_bars[t], betas[t]\n mean = (1 / torch.sqrt(alpha_t)) * (x - (beta_t / torch.sqrt(1 - alpha_bar_t)) * pred_noise)\n x = mean + torch.sqrt(beta_t) * torch.randn_like(x) if t > 0 else mean\n return x.squeeze(1)\n\nSHARE = 0.9 # fraction of each noise draw's variance shared across frames; the rest is independent per frame\n\ndef mixed_noise(n_frames):\n shared = torch.randn(1, 1, SIZE, SIZE)\n indep = torch.randn(n_frames, 1, SIZE, SIZE)\n return np.sqrt(SHARE) * shared + np.sqrt(1 - SHARE) * indep\n\n@torch.no_grad()\ndef sample_video_clip(model, seed, n_frames=NF):\n \"\"\"Temporal-attention + frame-position-aware architecture, PLUS mostly-shared noise.\"\"\"\n torch.manual_seed(seed)\n frame_idx = torch.arange(n_frames)\n x = mixed_noise(n_frames)\n for t in reversed(range(T)):\n t_batch = torch.full((n_frames,), t, dtype=torch.long)\n pred_noise = model(x, t_batch, frame_idx)\n alpha_t, alpha_bar_t, beta_t = alphas[t], alpha_bars[t], betas[t]\n mean = (1 / torch.sqrt(alpha_t)) * (x - (beta_t / torch.sqrt(1 - alpha_bar_t)) * pred_noise)\n x = mean + torch.sqrt(beta_t) * mixed_noise(n_frames) if t > 0 else mean\n return x.squeeze(1)\n\ndef frame_centroid(frame):\n ys, xs = np.where(frame.numpy() > 0)\n return None if len(xs) == 0 else np.array([xs.mean(), ys.mean()])\n\ndef clip_jerk(clip):\n centroids = [frame_centroid(f) for f in clip]\n if any(c is None for c in centroids):\n return None\n vels = [centroids[i + 1] - centroids[i] for i in range(len(centroids) - 1)]\n jerks = [np.linalg.norm(vels[i + 1] - vels[i]) for i in range(len(vels) - 1)]\n return np.mean(jerks)\n\ndef clip_speed(clip):\n \"\"\"mean distance moved between consecutive frames -- near 0 would mean a frozen clip.\"\"\"\n centroids = [frame_centroid(f) for f in clip]\n if any(c is None for c in centroids):\n return None\n return np.mean([np.linalg.norm(centroids[i + 1] - centroids[i]) for i in range(len(centroids) - 1)])\n\nreal_jerks = [clip_jerk(torch.tensor(c)) for c in clips[:30] if clip_jerk(torch.tensor(c)) is not None]\n\nn_eval = 30\nbaseline_jerks, video_jerks, baseline_speeds, video_speeds = [], [], [], []\nfor seed in range(n_eval):\n bclip, vclip = sample_baseline_clip(baseline_model, seed=1000 + seed), sample_video_clip(video_model, seed=2000 + seed)\n bj, vj = clip_jerk(bclip), clip_jerk(vclip)\n bs, vs = clip_speed(bclip), clip_speed(vclip)\n if bj is not None: baseline_jerks.append(bj)\n if vj is not None: video_jerks.append(vj)\n if bs is not None: baseline_speeds.append(bs)\n if vs is not None: video_speeds.append(vs)\n\nprint(f'real training clips: jerk = {np.mean(real_jerks):.3f}')\nprint(f'baseline (frame-independent sampling): jerk = {np.mean(baseline_jerks):.3f}, speed = {np.mean(baseline_speeds):.3f} (n={len(baseline_jerks)}/{n_eval})')\nprint(f'video model (temporal attn + mostly-shared noise): jerk = {np.mean(video_jerks):.3f}, speed = {np.mean(video_speeds):.3f} (n={len(video_jerks)}/{n_eval})')", "metadata": {}, "execution_count": null, "outputs": [] }, { "cell_type": "code", "id": "fa30e345", "source": "bclip_show = sample_baseline_clip(baseline_model, seed=1000)\nvclip_show = sample_video_clip(video_model, seed=2000)\n\nfig, axes = plt.subplots(2, NF, figsize=(7, 3.5))\nfor i in range(NF):\n axes[0, i].imshow(bclip_show[i], cmap='gray'); axes[0, i].axis('off')\n axes[1, i].imshow(vclip_show[i], cmap='gray'); axes[1, i].axis('off')\n axes[0, i].set_title(f't={i}', fontsize=8)\nplt.suptitle('Baseline (top, flickers) vs. temporal video model (bottom, moves smoothly)')\nplt.show()", "metadata": {}, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "ae4f4264", "source": "The video model's jerk is a clear fraction of the baseline's, while its speed confirms the shape is genuinely moving (not frozen) — both halves of the fix are doing real work, but a small tiny-model-on-tiny-data gap remains between the video model's jerk and the real clips' exact 0, since 2,500 toy-scale training steps on 500 clips can't perfectly learn \"constant velocity.\" (Exercise 2 asks you to probe how much each half of the fix is pulling its own weight.)\n\n## Conditioning and rollout: how these models are actually used\n\nA model that can only generate an unconditional 4-frame clip of a moving blob isn't yet useful on its own. Two conditioning recipes turn it into something closer to a real product:\n\n- **Image-to-video**: fix the first frame to a given real image (e.g. concatenate it as an extra input channel, or cross-attend to it the way Lesson {{lesson:diffusion_models}}'s note on Stable Diffusion cross-attends to a text embedding) and only denoise the remaining frames — this is the \"animate this photo\" capability, and it reuses exactly the same noise-prediction network, just with one frame's content supplied instead of sampled.\n- **Text-to-video**: condition every frame's denoising step on a text embedding (Lesson {{lesson:vision_language_models}}'s CLIP text encoder, or a dedicated language model), via cross-attention, exactly as in Lesson {{lesson:diffusion_models}}'s text-to-image discussion — the only new piece here is applying that same conditioning consistently across all frames, so the generated motion matches the prompt throughout the clip, not just in one frame.\n\nBoth conditioning recipes generate a clip of a *fixed* length. Real video models need to go well beyond `NF` frames, and no single joint denoising pass scales to minutes of footage. The standard solution is **autoregressive rollout**: generate a short chunk of frames (as this lesson does), then generate the *next* chunk conditioned on the last few already-generated frames (image-to-video conditioning, applied to the model's own output) instead of starting fresh — repeating this produces an arbitrarily long video, at the cost of slowly accumulating drift from one chunk to the next, since every chunk's conditioning is itself a generated (not real) clip.\n\n### Exercises\n\n1. Set `SHARE = 1.0` in `mixed_noise` and re-sample from the already-trained `video_model` (which already includes `frame_embed`). Is every frame identical, or does the model still generate genuinely different content per frame? Now build and train a second copy of `VideoDenoiseNet` with the `+ femb` terms removed from `forward` (so it never learns to use frame position at all), and sample from *that* model, also with `SHARE = 1.0`. Which single piece — removing the frame-position embedding, or using fully shared noise — is actually responsible for the degenerate frozen-clip solution described above, and which one turns out not to matter much on its own?\n2. The video model's fix has two independent parts: the architecture (`TemporalAttention` + `frame_embed`) and the sampling trick (mostly-shared noise, `SHARE = 0.9`). Sample from the trained `video_model` using fully *independent* per-frame noise (`SHARE = 0.0`) instead, and separately, sample from the trained `baseline_model` using `mixed_noise` instead of independent noise (ignoring that it was never trained with frame indices, so just drop the `frame_idx` argument it doesn't accept). Which change moves the jerk metric more on its own?\n3. Increase `vx, vy`'s range in `make_clip` to allow speeds of 2 pixels/frame instead of 1, and retrain both models. Does the video model's jerk advantage over the baseline hold up at the faster, more dramatic motion, or does it degrade — and if it degrades, why might faster motion be a harder case for a fixed `NF`-frame temporal window and a fixed `SHARE` to track well?", "metadata": {} } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.x" } }, "nbformat": 4, "nbformat_minor": 5 }