Flow Matching (Rectified Flow) - Generative Models
Implement Flow Matching (Rectified Flow) in Python with NumPy. medium exercise with examples, hints, and instant feedback. No setup or signup required.
Conditional Flow Matching (Rectified Flow)
What you'll build
The training target and loss of flow matching with straight paths, and an Euler sampler that integrates the learned velocity field from noise to data.
Background
Flow matching (Lipman et al., 2023) and rectified flow (Liu et al., 2023) train a network v_θ(x, t) to be a velocity field whose ODE carries noise to data. With straight-line paths between a noise sample and a data sample, training is plain regression, with no noise schedule or variance bookkeeping. Stable Diffusion 3 and Flux are trained this way.
Formula (training)
Sample noise x0 ~ N(0, I), a data point x1, and t ~ U(0, 1) per example:
x_t = (1 − t) · x0 + t · x1 # point on the straight path
target = d x_t / dt = x1 − x0 # its constant velocity
loss = mean( (v_pred − target)² ) # mean over every element of the batch
Here t = 0 is noise and t = 1 is data (some papers use the opposite convention). t has shape (B,): reshape it to (B, 1, ..., 1) so it broadcasts over the remaining dimensions.
Algorithm (sampling with Euler steps)
- Start from
x = x0and setdt = 1 / n_steps. - For
i = 0, ..., n_steps − 1:t = i · dt(a Python float),x ← x + dt · velocity_fn(x, t). - Return
x.
If the learned field is exactly the straight-line velocity, a single Euler step lands on the data: that is the appeal of straight (rectified) paths.
Inputs
x0,x1,v_pred: float arrays(B, ...)of the same shapet: float array(B,)with values in [0, 1]velocity_fn(x, t): callable returning an array shaped likex;n_steps: int ≥ 1
Expected return format
flow_matching_loss(x0, x1, t, v_pred)returns a dict:'x_t': interpolated points, same shape asx0'target': target velocityx1 − x0, same shape asx0'loss': MSE betweenv_predandtarget(Python float)
euler_step(x_t, v, dt)returnsx_t + dt · veuler_sample(x0, velocity_fn, n_steps)returns the finalx(same shape asx0)
Constraints
NumPy only; the Euler loop is over steps, never over elements.
Follow-up
- How is this related to diffusion? (With Gaussian paths, flow matching recovers a probability-flow ODE of a diffusion model.)
- What does "reflow" do to make one-step sampling work for a real, imperfect model?
Examples
Input: x0 = [[0, 0], [1, 1]], x1 = [[2, 4], [3, -1]], t = [0.25, 0.5] Output: x_t = [[0.5, 1.0], [2.0, 0.0]], target = [[2, 4], [2, -2]]
Row 0 is a quarter of the way to its data point, row 1 halfway
Input: euler_sample(np.array([2.0]), lambda x, t: -x, 4) Output: array([0.6328])
Each step multiplies x by (1 − 1/4), so 2 · 0.75⁴ = 0.6328 (the exact ODE solution is 2·e⁻¹ = 0.7358)
Practice in Python
import numpy as np
def flow_matching_loss(x0, x1, t, v_pred):
"""
Conditional flow matching with straight paths.
Args:
x0: (B, ...) noise samples
x1: (B, ...) data samples
t: (B,) times in [0, 1]
v_pred: (B, ...) predicted velocities at (x_t, t)
Returns:
dict with 'x_t', 'target' (same shape as x0) and 'loss' (float)
"""
# Your code here
pass
def euler_step(x_t, v, dt):
"""One Euler step: x_t + dt * v"""
# Your code here
pass
def euler_sample(x0, velocity_fn, n_steps):
"""Integrate dx/dt = velocity_fn(x, t) from t = 0 to 1 with n_steps Euler steps."""
# Your code here
pass
The interactive editor loads with Python and NumPy. Run your implementation against 5 tests, or use the progressive hints.