Diffusion and Flow World Models › Worked Example: A Bimodal One-Dimensional World

Michael BrenndoerferJune 28, 202615 min read
World Models HandbookAll chapters
Part 5 of 12World-Model Architectures
Chapter 28Diffusion and Flow World Models › Worked Example: A Bimodal One-Dimensional World
Reading section 6 of 9 · Worked Example: A Bimodal One-Dimensional World · 15 minSection 6 of 9 · 15 min

Book contents

Section 6 of 9 · 15 min read

Diffusion and Flow World Models › Worked Example: A Bimodal One-Dimensional WorldOne section · Sections in this chapter6 / 9 ↓
Reading tools Definitions & inline links

Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.

Article links

Make inline references clickable

To make this concrete, we will build a minimal one-dimensional conditional testbed in which the MSE-optimal point estimate lies between two modes. The state sts_t and action ata_t are scalars. Given (st,at)(s_t, a_t), the next state is

st+1=st+at+m⋅Δ+σ ϵ,m∼Uniform{−1,+1},ϵ∼N(0,1),s_{t+1} = s_t + a_t + m \cdot \Delta + \sigma\,\epsilon, \qquad m \sim \mathrm{Uniform}\{-1,+1\},\quad \epsilon\sim\mathcal N(0,1),

with mm and ϵ\epsilon independent, Δ=1.2\Delta = 1.2, and σ=0.25\sigma = 0.25. The hidden variable mm is never observed. Thus p(st+1∣st,at)=12N(st+at−Δ,σ2)+12N(st+at+Δ,σ2)p(s_{t+1}\mid s_t,a_t)=\tfrac12\mathcal N(s_t+a_t-\Delta,\sigma^2)+\tfrac12\mathcal N(s_t+a_t+\Delta,\sigma^2). A single point cannot represent both components. Under squared error, its population optimum is the conditional mean st+ats_t+a_t, halfway between them in a region of extremely low, but nonzero, density. This mirrors the puck example from the introduction: the dynamics are simple conditional on the hidden mode, and the difficulty is that the mode is unknown to the model.

This testbed also separates marginal and conditional diagnostics. In the data generator below, st∼N(0,1)s_t\sim\mathcal N(0,1) and at∼N(0,0.52)a_t\sim\mathcal N(0,0.5^2) independently of mm and ϵ\epsilon. After integrating over st+ats_t+a_t, the analytic marginal is only weakly bimodal: its center is about 0.63%0.63\% below its nearby maxima. A coarse marginal histogram can therefore look like one broad peak even though every fixed-context conditional has two well-separated components. Evaluating only the marginal can hide the conditional ambiguity that matters for prediction.

We begin with imports and dataset generation.

In[7]:
Code
import math

import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
from scipy.stats import wasserstein_distance
from torch.utils.data import DataLoader, TensorDataset

torch.manual_seed(0)
rng = np.random.default_rng(0)

DELTA = 1.2
SIGMA = 0.25
N = 20000

s_t = rng.normal(0.0, 1.0, size=N).astype(np.float32)
a_t = rng.normal(0.0, 0.5, size=N).astype(np.float32)
mode = (rng.integers(0, 2, size=N) * 2 - 1).astype(np.float32)
s_next = (
    s_t + a_t + mode * DELTA + SIGMA * rng.normal(size=N).astype(np.float32)
)

context = torch.from_numpy(np.stack([s_t, a_t], axis=-1))
target = torch.from_numpy(s_next[:, None])

# A held-out batch of "ground truth" next states for context (s_t = 0, a_t = 0).
N_DEMO = 2000
mode_demo = (rng.integers(0, 2, size=N_DEMO) * 2 - 1).astype(np.float32)
s_next_gt = mode_demo * DELTA + SIGMA * rng.normal(size=N_DEMO).astype(
    np.float32
)
Out[8]:
Console
Dataset: 20000 transitions, target shape (20000, 1)

The dataset contains 20,000 one-step transitions, and each target is a scalar next state, so the generative models operate on a one-dimensional continuous variable. The conditioning vector is two-dimensional, stacking sts_t and ata_t, and the hidden mode is discarded at training time so that no model can cheat by observing it. This compact implementation runs on CPU with float32 tensors. A device-portable version must place all tensors participating in an operation on compatible devices, align the floating dtype of model parameters, schedules, inputs, and temporary floating tensors, and preserve the required integer or Boolean dtypes of indices and masks. Samples must then move to CPU before NumPy, SciPy, or Matplotlib use. The seeds stabilize this run within a fixed environment, but exact losses and metrics can change across library versions or hardware.

We can see the bimodality directly by plotting the held-out samples for the fixed context.

Out[9]:
Visualization
Histogram with two distinct peaks at plus and minus one point two.
Synthetic-reference conditional next-state distribution p(s_{t+1} | s_t = 0, a_t = 0), estimated from 2,000 held-out simulator samples. The two Gaussian components have means +1.2 and -1.2, separated by 2*Delta = 2.4, and the histogram has a low-density trough at the origin. Analytically, the MSE-optimal point prediction lies at that origin.

The distribution peaks near +1.2+1.2 and −1.2-1.2 and has extremely low density near the origin. A point model optimized by MSE targets the conditional mean at the origin. We will compare that analytic consequence with the empirical behavior of learned diffusion and flow models.

We define a small MLP denoiser conditioned on tt, sts_t, and ata_t. Fixed sinusoidal time features are a common diffusion baseline, although many architectures transform them further or use learned alternatives. Here they give the network a smooth, multi-scale encoding of the timestep so that nearby timesteps produce nearby embeddings, which helps the network share structure across the schedule.

In[10]:
Code
class Denoiser(nn.Module):
    def __init__(self, hidden=128, time_dim=32):
        super().__init__()
        self.time_dim = time_dim
        self.net = nn.Sequential(
            nn.Linear(2 + time_dim + 1, hidden),
            nn.SiLU(),
            nn.Linear(hidden, hidden),
            nn.SiLU(),
            nn.Linear(hidden, 1),
        )

    def time_embed(self, t):
        freq = torch.exp(
            torch.linspace(
                0.0,
                math.log(50.0),
                self.time_dim // 2,
                device=t.device,
                dtype=t.dtype,
            )
        )
        args = t[:, None] * freq[None, :]
        return torch.cat([torch.sin(args), torch.cos(args)], dim=-1)

    def forward(self, x, t, ctx):
        emb = self.time_embed(t)
        h = torch.cat([x, ctx, emb], dim=-1)
        return self.net(h)

The forward diffusion schedule uses a linear beta increase from 10−410^{-4} to 0.050.05 over T=500T = 500 steps, which gives αˉT≈3×10−6\bar\alpha_T \approx 3 \times 10^{-6}, essentially pure noise.

In[11]:
Code
T = 500
betas = torch.linspace(1e-4, 0.05, T)
alphas = 1.0 - betas
alpha_bars = torch.cumprod(alphas, dim=0)


def q_sample(x0, t_idx):
    a_bar = alpha_bars[t_idx].unsqueeze(-1)
    eps = torch.randn_like(x0)
    return torch.sqrt(a_bar) * x0 + torch.sqrt(1 - a_bar) * eps, eps

We train the denoiser for 30 epochs. Each minibatch samples a random timestep per example and applies the denoising score matching objective.

In[12]:
Code
ddpm = Denoiser()
opt_ddpm = torch.optim.Adam(ddpm.parameters(), lr=2e-3)
loader = DataLoader(
    TensorDataset(target, context), batch_size=256, shuffle=True
)

ddpm_losses = []
ddpm.train()
for epoch in range(30):
    running = 0.0
    for x0, ctx in loader:
        t_idx = torch.randint(0, T, (x0.size(0),))
        x_t, eps = q_sample(x0, t_idx)
        t_norm = (t_idx.float() + 1.0) / T
        pred = ddpm(x_t, t_norm, ctx)
        loss = F.mse_loss(pred, eps)
        opt_ddpm.zero_grad()
        loss.backward()
        opt_ddpm.step()
        running += loss.item() * x0.size(0)
    ddpm_losses.append(running / len(target))
Out[13]:
Console
Final denoising loss: 0.2118

The reported denoising loss is an optimization diagnostic, not a calibrated likelihood or a guarantee of sample quality. It need not reach zero because, after corruption, a particular noise realization is not uniquely determined by (xt,t,c)(x_t,t,c); this ambiguity is especially relevant at low-noise timesteps where the observation remains strongly tied to the unknown clean sample.

Two samplers are useful. Full DDPM ancestral sampling visits every timestep and injects stochastic noise; DDIM sampling follows a selected timestep sequence and is deterministic given its initial noise when η=0\eta=0. Both use the learned denoiser to approximate a reverse generative process. They do not literally invert the independently sampled forward corruptions used during training.

In[14]:
Code
@torch.no_grad()
def ddpm_sample(ctx):
    x = torch.randn(ctx.size(0), 1)
    for i in reversed(range(T)):
        t_norm = torch.full((ctx.size(0),), (i + 1.0) / T)
        eps = ddpm(x, t_norm, ctx)
        a_bar = alpha_bars[i]
        mean = (x - betas[i] / torch.sqrt(1 - a_bar) * eps) / torch.sqrt(
            alphas[i]
        )
        if i > 0:
            x = mean + torch.sqrt(betas[i]) * torch.randn_like(x)
        else:
            x = mean
    return x


@torch.no_grad()
def ddim_sample(ctx, n_steps=50):
    step_indices = torch.linspace(T - 1, 0, n_steps).long().tolist()
    x = torch.randn(ctx.size(0), 1)
    for j, i in enumerate(step_indices):
        t_norm = torch.full((ctx.size(0),), (i + 1.0) / T)
        eps = ddpm(x, t_norm, ctx)
        a_bar = alpha_bars[i]
        x0_hat = (x - torch.sqrt(1 - a_bar) * eps) / torch.sqrt(a_bar)
        next_i = step_indices[j + 1] if j + 1 < len(step_indices) else -1
        a_bar_next = alpha_bars[next_i] if next_i >= 0 else torch.tensor(1.0)
        x = torch.sqrt(a_bar_next) * x0_hat + torch.sqrt(1 - a_bar_next) * eps
    return x

We now define and train a mean-square-error baseline for comparison. This is a plain MLP that maps (st,at)(s_t, a_t) to a point estimate of st+1s_{t+1}.

In[15]:
Code
mse_model = nn.Sequential(
    nn.Linear(2, 128),
    nn.SiLU(),
    nn.Linear(128, 128),
    nn.SiLU(),
    nn.Linear(128, 1),
)
opt_mse = torch.optim.Adam(mse_model.parameters(), lr=2e-3)

for epoch in range(30):
    for x0, ctx in loader:
        pred = mse_model(ctx)
        loss = F.mse_loss(pred, x0)
        opt_mse.zero_grad()
        loss.backward()
        opt_mse.step()

with torch.no_grad():
    mse_pred_at_origin = mse_model(torch.zeros(1, 2)).item()
Out[16]:
Console
MSE prediction at (s_t=0, a_t=0): -0.020

The learned point estimate should land near the analytic conditional mean at the origin, in the low-density trough between the modes. This is not a bug: it is the population-optimal answer for squared error, even though it is unrepresentative of a typical draw from the conditional distribution.

Now we turn to flow matching. The model class is identical, but the corruption path, time sampling, and regression target change.

In[17]:
Code
flow = Denoiser()
opt_flow = torch.optim.Adam(flow.parameters(), lr=2e-3)

flow_losses = []
flow.train()
for epoch in range(30):
    running = 0.0
    for x1, ctx in loader:
        z0 = torch.randn_like(x1)
        t = torch.rand(x1.size(0), 1)
        x_t = (1.0 - t) * z0 + t * x1
        target_v = x1 - z0
        pred_v = flow(x_t, t.squeeze(-1), ctx)
        loss = F.mse_loss(pred_v, target_v)
        opt_flow.zero_grad()
        loss.backward()
        opt_flow.step()
        running += loss.item() * x1.size(0)
    flow_losses.append(running / len(target))
Out[18]:
Console
Final flow-matching loss: 1.6424

The flow-matching loss is another training diagnostic. Its numerical value should not be compared directly with the denoising loss because the objectives use different targets, scales, and timestep weighting. Here, separate instances of the same small MLP architecture optimize the two objectives well enough for the distributional diagnostics below.

Sampling from the flow model uses forward Euler integration from t=0t = 0 to t=1t = 1. The numerical path is deterministic conditional on the sampled Gaussian initial value.

In[19]:
Code
@torch.no_grad()
def flow_sample(ctx, steps=20):
    x = torch.randn(ctx.size(0), 1)
    dt = 1.0 / steps
    for i in range(steps):
        t = torch.full((ctx.size(0),), i * dt)
        v = flow(x, t, ctx)
        x = x + v * dt
    return x

We now generate fixed-context samples using three procedures: ancestral DDPM and DDIM sampling from the learned diffusion model, and Euler sampling from the learned flow model. The moment summary includes all three, while the histogram directly compares the synthetic reference, ancestral DDPM, and flow samples. All sampling happens before plotting.

In[20]:
Code
demo_ctx = torch.zeros(NUM_SAMPLES := 2000, 2)

torch.manual_seed(1)
samples_ddpm = ddpm_sample(demo_ctx)
samples_ddim = ddim_sample(demo_ctx, n_steps=40)
samples_flow = flow_sample(demo_ctx, steps=20)
Out[21]:
Console
DDPM sample mean -0.060, std 1.208
DDIM sample mean +0.001, std 1.157
Flow sample mean +0.050, std 1.230

The printed means and standard deviations provide a first check of symmetry and spread, but they cannot by themselves establish bimodality: a broad unimodal distribution can share both moments. The histogram below is the more relevant diagnostic because it shows whether probability mass concentrates near both component means.

We visualize three sampled distributions on the same axes together with the MSE point estimate. The DDIM sampler is used again for the rollout and step-sensitivity diagnostics below; the 40-step DDIM samples generated above are used only in the moment summary.

Out[22]:
Visualization
Overlaid density histograms for the synthetic reference, ancestral DDPM, and 20-step Euler flow samples all place mass near minus 1.2 and plus 1.2; a dashed MSE prediction line lies near zero between the peaks.
Overlaid density histograms of synthetic-reference samples, DDPM ancestral samples, and flow-matching samples with 20 Euler steps for the fixed context (s_t = 0, a_t = 0). The MSE point estimate is a dashed vertical line near the origin. In this trained run, both generative models place mass near both component means, whereas the point baseline lies in the low-density middle.

In this seeded run, DDPM and flow matching place samples near both synthetic-reference peaks, while the MSE baseline lies near zero in a very low-density region. The general lesson is narrower than a claim of universal superiority: a point prediction cannot represent a multimodal conditional, whereas a well-trained generative model can represent multiple alternatives.

Distribution recovery is only half the story. We next illustrate recursive rollout behavior by repeatedly calling each per-state model on its own output. The hidden mode in this simulator is resampled independently at every transition, so the six-step process has many possible mode sequences rather than one persistent left/right branch.

In[23]:
Code
actions_seq = [0.6, -0.2, 0.4, -0.1, 0.5, -0.3]
H = len(actions_seq)


def rollout(sample_fn, seed, n_rolls=8):
    trajectories = []
    for k in range(n_rolls):
        torch.manual_seed(seed + k)
        s = torch.zeros(1, 1)
        traj = [s.item()]
        for a in actions_seq:
            ctx = torch.cat([s, torch.full((1, 1), a)], dim=-1)
            s = sample_fn(ctx)
            traj.append(s.item())
        trajectories.append(np.array(traj))
    return np.stack(trajectories)


traj_ddim = rollout(lambda c: ddim_sample(c, n_steps=40), seed=10)
traj_flow = rollout(lambda c: flow_sample(c, steps=20), seed=10)


def rollout_mse(n_rolls=8):
    trajectories = []
    for k in range(n_rolls):
        s = torch.zeros(1, 1)
        traj = [s.item()]
        for a in actions_seq:
            ctx = torch.cat([s, torch.full((1, 1), a)], dim=-1)
            s = mse_model(ctx).detach()
            traj.append(s.item())
        trajectories.append(np.array(traj))
    return np.stack(trajectories)


traj_mse = rollout_mse()
Out[24]:
Console
Rollout tensor shapes: DDIM (8, 7), flow (8, 7), MSE (8, 7)

The shape is (8, 7) for each model: eight separately initialized rollout runs and seven states including the initial state. The generative rows use distinct seeds, while the deterministic MSE rows are identical.

We plot the trajectories from both generative models and the MSE baseline.

Out[25]:
Visualization
Eight DDIM trajectories share an initial state and diverge across six transitions, ending at seven plotted state positions.
Eight seeded DDIM rollouts for the same action sequence. Because each per-state call draws a new conditional sample, successive states can land near different learned mixture components, producing diverse recursive paths.
Eight flow-matching trajectories share an initial state and diverge across six transitions, ending at seven plotted state positions.
Eight seeded flow-matching rollouts for the same action sequence. Diversity comes from a fresh Gaussian initial value at every sampled transition.
Eight identical MSE trajectories overlap as one line across six transitions and seven plotted state positions.
The deterministic MSE baseline produces one repeated rollout of one-step point predictions and therefore does not display conditional sample diversity.

The generative rollouts diverge because each per-state call draws another conditional sample, while all eight deterministic MSE traces overlap. These paths are a behavior illustration, not evidence that the learned models reproduce the true joint rollout distribution. Indeed, because the simulator resamples mm at each transition, a rollout can switch branch sign; joint horizon consistency would need a separate trajectory-level evaluation.

We close the experiment with a sampling-step sensitivity check. For this implementation, both DDIM and Euler use one network evaluation per step, so step count equals NFE. We measure Wasserstein-1 distance to the finite synthetic-reference sample under one fixed seed; this is an illustrative diagnostic, not a benchmark.

In[26]:
Code
step_options = [4, 8, 16, 32, 64]
flow_errors = []
ddim_errors = []

for n in step_options:
    torch.manual_seed(2)
    fs = flow_sample(demo_ctx, steps=n).numpy().ravel()
    torch.manual_seed(2)
    ds = ddim_sample(demo_ctx, n_steps=n).numpy().ravel()
    flow_errors.append(wasserstein_distance(fs, s_next_gt))
    ddim_errors.append(wasserstein_distance(ds, s_next_gt))
Out[27]:
Console
Flow W1 by steps: ['0.209', '0.185', '0.159', '0.156', '0.158']
DDIM W1 by steps: ['1.139', '0.313', '0.150', '0.104', '0.087']

The resulting quality-versus-steps curves show a practical difference between these trained DDIM and Euler-flow implementations.

Out[28]:
Visualization
Line chart where Euler flow has lower Wasserstein error at four and eight evaluations, the methods cross near sixteen, and DDIM is lower thereafter.
Seeded Wasserstein-1 distance to 2000 synthetic-reference samples versus network evaluations (4, 8, 16, 32, 64). Euler flow is lower at 4 and 8 evaluations, the curves are close at 16, and DDIM is lower at 16, 32, and 64 for this trained run.

Here the measured flow errors at 4, 8, 16, 32, and 64 evaluations are approximately 0.2090.209, 0.1850.185, 0.1590.159, 0.1560.156, and 0.1580.158; the DDIM errors are 1.1391.139, 0.3130.313, 0.1500.150, 0.1040.104, and 0.0870.087. Euler flow is markedly better at the two coarsest budgets, the curves cross around 16 evaluations, and DDIM is better at the larger budgets. This single seeded, one-dimensional experiment supports a local quality-versus-NFE comparison only; it does not establish a general ordering between flow and diffusion samplers. It is not a latency measurement: deployment requires end-to-end timing on the target control stack.