Diffusion and Flow World Models › Worked Example: A Bimodal One-Dimensional World
World Models HandbookAll chapters 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 and action are scalars. Given , the next state is
with and independent, , and . The hidden variable is never observed. Thus . A single point cannot represent both components. Under squared error, its population optimum is the conditional mean , 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, and independently of and . After integrating over , the analytic marginal is only weakly bimodal: its center is about 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.
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
)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 and , 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.

The distribution peaks near and 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 , , and . 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.
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 to over steps, which gives , essentially pure noise.
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, epsWe train the denoiser for 30 epochs. Each minibatch samples a random timestep per example and applies the denoising score matching objective.
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))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 ; 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 . Both use the learned denoiser to approximate a reverse generative process. They do not literally invert the independently sampled forward corruptions used during training.
@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 xWe now define and train a mean-square-error baseline for comparison. This is a plain MLP that maps to a point estimate of .
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()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.
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))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 to . The numerical path is deterministic conditional on the sampled Gaussian initial value.
@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 xWe 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.
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)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.

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.
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()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.



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 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.
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))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.

Here the measured flow errors at 4, 8, 16, 32, and 64 evaluations are approximately , , , , and ; the DDIM errors are , , , , and . 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.