The three networks that make up a DIAMOND agent.
Parameters
configDIAMONDConfigAttributes
denoiser_Denoiserreward_end_RewardEndModelactor_critic_ActorCriticNotes
Reference: Alonso, Eloi, et al., "Diffusion for World Modeling: Visual Details Matter in Atari", NeurIPS, 2024 (arXiv:2405.12399).
The EDM preconditioners live on this class rather than inside the U-Net, because they are a property of the diffusion, not of the network: the same U-Net under DDPM's parameterisation is the comparison the paper's Section 5.1 runs, and it drifts.
Examples
>>> import lucid
>>> from lucid.models.generative.diamond import DIAMONDConfig, DIAMONDModel
>>> config = DIAMONDConfig(
... sample_size=16, unet_channels=(8, 8), unet_layers=(1, 1),
... reward_channels=(8, 8), reward_layers=(1, 1),
... actor_channels=(8, 8), actor_layers=(1, 1),
... cond_dim=16, reward_cond_dim=8,
... reward_lstm_dim=16, actor_lstm_dim=16, num_actions=4)
>>> model = DIAMONDModel(config).eval()
>>> frames = lucid.randn((2, 4, 3, 16, 16))
>>> actions = lucid.tensor([[0, 1, 2, 3], [1, 1, 0, 2]], dtype=lucid.int64)
>>> with lucid.no_grad():
... nxt = model.imagine_frame(frames, actions)
>>> nxt.shape
(2, 3, 16, 16)Used by 2
Constructors
1Instance methods
9denoise(noised: Tensor, sigma: Tensor, frames: Tensor, actions: Tensor, cond_sigma: Tensor | None = None, quantize: bool = True)Apply — the preconditioned denoiser.
Parameters
noisedTensor(B, C, H, W), the next frame with noise added.sigmaTensor(B,) noise levels.framesTensor(B, L, C, H, W) clean history.actionsTensor(B, L) indices, or (B, L, num_actions) multi-hot.(B,) level to noise the history at. Only meaningful
when the configuration asks for it; None leaves the
history clean, which is what Atari does.quantizebool= True, keyword-onlyFalse: the quantiser has
no gradient, and the loss belongs on the estimate anyway.Returns
Tensor(B, C, H, W) estimate of the clean next frame.
Examples
>>> import lucid
>>> from lucid.models.generative.diamond import DIAMONDConfig, DIAMONDModel
>>> config = DIAMONDConfig(
... sample_size=16, unet_channels=(8, 8), unet_layers=(1, 1),
... reward_channels=(8, 8), reward_layers=(1, 1),
... actor_channels=(8, 8), actor_layers=(1, 1),
... cond_dim=16, reward_cond_dim=8,
... reward_lstm_dim=16, actor_lstm_dim=16, num_actions=4)
>>> model = DIAMONDModel(config).eval()
>>> frames = lucid.randn((2, 4, 3, 16, 16))
>>> actions = lucid.tensor([[0, 1, 2, 3], [1, 1, 0, 2]], dtype=lucid.int64)
>>> noised, sigma = lucid.randn((2, 3, 16, 16)), lucid.tensor([0.5, 2.0])
>>> clean = model.denoise(noised, sigma, frames, actions)
>>> clean.shape
(2, 3, 16, 16)
By default the estimate is put back on the 8-bit grid of [-1, 1],
which has no gradient; training asks for the raw estimate instead:
>>> bool(clean.min() >= -1.0), bool(clean.max() <= 1.0), clean.requires_grad
(True, True, False)
>>> model.denoise(noised, sigma, frames, actions, quantize=False).requires_grad
Trueforward(frames: Tensor, actions: Tensor, next_frame: Tensor, sigma: Tensor | None = None)Train the denoiser on one transition.
Parameters
Returns
DIAMONDOutputLoss, denoised frame, and the noise levels used.
Notes
Reference: Alonso et al., arXiv:2405.12399, Algorithm 1 — sample , noise the target, and take the squared error in pixel space. The EDM weighting is already inside , so no extra loss weight appears here.
imagine_frame(frames: Tensor, actions: Tensor, steps: int | None = None, noise: Tensor | None = None, cond_sigma: float | None = None)Sample the next frame with Euler's method.
Parameters
framesTensor(B, L, C, H, W) clean history.actionsTensor(B, L) indices, or (B, L, num_actions) multi-hot.stepsint or None= Nonecond_sigma(float or None, optional, keyword - only)= None0.005; None leaves
the history clean, which is what Atari does.Returns
Tensor(B, C, H, W).
Notes
Each step moves along , which is the probability-flow ODE written in EDM's variables.
Examples
>>> import lucid
>>> from lucid.models.generative.diamond import DIAMONDConfig, DIAMONDModel
>>> config = DIAMONDConfig(
... sample_size=16, unet_channels=(8, 8), unet_layers=(1, 1),
... reward_channels=(8, 8), reward_layers=(1, 1),
... actor_channels=(8, 8), actor_layers=(1, 1),
... cond_dim=16, reward_cond_dim=8,
... reward_lstm_dim=16, actor_lstm_dim=16, num_actions=4)
>>> model = DIAMONDModel(config).eval()
>>> frames = lucid.randn((2, 4, 3, 16, 16))
>>> actions = lucid.tensor([[0, 1, 2, 3], [1, 1, 0, 2]], dtype=lucid.int64)
>>> noise = lucid.randn((2, 3, 16, 16))
>>> with lucid.no_grad():
... first = model.imagine_frame(frames, actions, noise=noise)
... again = model.imagine_frame(frames, actions, noise=noise)
... single = model.imagine_frame(frames, actions, noise=noise, steps=1)
>>> first.shape
(2, 3, 16, 16)
The sampler solves an ODE, so the starting noise fixes the frame and
a rollout can be reproduced; the number of steps changes it:
>>> bool((first == again).all()), bool((first == single).all())
(True, False)EDM's four scalings at a noise level.
Parameters
sigmaTensor(B,) noise levels.Returns
Notes
Reference: Alonso et al., arXiv:2405.12399, Appendix C, equations 9-12, with .
Examples
>>> import lucid
>>> from lucid.models.generative.diamond import DIAMONDConfig, DIAMONDModel
>>> config = DIAMONDConfig(
... sample_size=16, unet_channels=(8, 8), unet_layers=(1, 1),
... reward_channels=(8, 8), reward_layers=(1, 1),
... actor_channels=(8, 8), actor_layers=(1, 1),
... cond_dim=16, reward_cond_dim=8,
... reward_lstm_dim=16, actor_lstm_dim=16, num_actions=4)
>>> model = DIAMONDModel(config)
>>> c_in, c_out, c_skip, c_noise = model.preconditioners(
... lucid.tensor([0.002, 0.5, 5.0]))
>>> c_in.shape, c_noise.shape
((3, 1, 1, 1), (3,))
The skip hands over to the network as the noise grows, until the
network is asked for the clean frame rather than for the noise:
>>> [round(v, 3) for v in c_skip.reshape(-1).tolist()]
[0.735, 0.424, 0.01]
The low end stops short of 1 because the offset noise is folded
into every level first, so even a clean level is scaled as if it
carried sigma_offset_noise:
>>> data, offset = config.sigma_data, config.sigma_offset_noise
>>> round(data**2 / (data**2 + offset**2), 3)
0.735The noise levels an Euler sampler walks down.
Parameters
stepsintdevicestrReturns
Tensor(steps + 1,) descending to exactly zero.
Notes
Karras et al.'s -schedule, which Algorithm 1 refers to as "the default identity schedule from EDM". ⚠️ DIAMOND's tables give the sampler and the step count but not , or , so those come from EDM itself and are the one place here where a number is inherited rather than cited.
Examples
>>> from lucid.models.generative.diamond import DIAMONDConfig, DIAMONDModel
>>> config = DIAMONDConfig(
... sample_size=16, unet_channels=(8, 8), unet_layers=(1, 1),
... reward_channels=(8, 8), reward_layers=(1, 1),
... actor_channels=(8, 8), actor_layers=(1, 1),
... cond_dim=16, reward_cond_dim=8,
... reward_lstm_dim=16, actor_lstm_dim=16, num_actions=4)
>>> model = DIAMONDModel(config)
>>> schedule = model.sigma_schedule(3, "cpu")
>>> schedule.shape
(4,)
EDM's sigma_max down to its sigma_min, then an exact zero, so
the last Euler step lands on the denoiser's own estimate:
>>> [round(v, 4) for v in schedule.tolist()]
[5.0, 0.2831, 0.002, 0.0]step_actor_critic(frame: Tensor, state: tuple[Tensor, Tensor] | None = None)One step of the actor-critic, typed.
Parameters
Returns
Examples
>>> import lucid
>>> from lucid.models.generative.diamond import DIAMONDConfig, DIAMONDModel
>>> config = DIAMONDConfig(
... sample_size=16, unet_channels=(8, 8), unet_layers=(1, 1),
... reward_channels=(8, 8), reward_layers=(1, 1),
... actor_channels=(8, 8), actor_layers=(1, 1),
... cond_dim=16, reward_cond_dim=8,
... reward_lstm_dim=16, actor_lstm_dim=16, num_actions=4)
>>> model = DIAMONDModel(config).eval()
>>> frame = lucid.randn((2, 3, 16, 16))
>>> with lucid.no_grad():
... logits, value, state = model.step_actor_critic(frame)
... carried, _, _ = model.step_actor_critic(frame, state)
>>> logits.shape, value.shape, state[0].shape
((2, 4), (2,), (2, 16))
The policy has a memory: the same frame read with the carried state
gives different logits than it did from a blank one.
>>> bool((carried == logits).all())
Falsestep_reward_end(frame: Tensor, next_frame: Tensor, action: Tensor, state: tuple[Tensor, Tensor] | None = None)One step of the reward/termination model, typed.
Parameters
frameTensor(B, C, H, W) each — the transition. Both are needed
because a reward is a property of the transition, and the
released encoder reads them stacked.next_frameTensor(B, C, H, W) each — the transition. Both are needed
because a reward is a property of the transition, and the
released encoder reads them stacked.actionTensor(B,) action indices taken from frame.Returns
Examples
>>> import lucid
>>> from lucid.models.generative.diamond import DIAMONDConfig, DIAMONDModel
>>> config = DIAMONDConfig(
... sample_size=16, unet_channels=(8, 8), unet_layers=(1, 1),
... reward_channels=(8, 8), reward_layers=(1, 1),
... actor_channels=(8, 8), actor_layers=(1, 1),
... cond_dim=16, reward_cond_dim=8,
... reward_lstm_dim=16, actor_lstm_dim=16, num_actions=4)
>>> model = DIAMONDModel(config).eval()
>>> frame, after = lucid.randn((2, 3, 16, 16)), lucid.randn((2, 3, 16, 16))
>>> action = lucid.tensor([0, 3], dtype=lucid.int64)
>>> with lucid.no_grad():
... reward, end, state = model.step_reward_end(frame, after, action)
>>> reward.shape, end.shape, [s.shape for s in state]
((2, 3), (2, 2), [(2, 16), (2, 16)])
The three reward classes are the clipped signs -1, 0, +1, in
that order, so their probabilities give an expected reward in
[-1, 1]:
>>> probs = lucid.softmax(reward, dim=-1)
>>> expected = probs[:, 2] - probs[:, 0]
>>> bool((expected.abs() <= 1).all())
Trueupsample(noised: Tensor, sigma: Tensor, low_res: Tensor, previous: Tensor, quantize: bool = True)Apply the upsampler's .
Parameters
noisedTensor(B, C, H*f, W*f) the full-resolution frame with noise.sigmaTensor(B,) noise levels.low_resTensor(B, C, H, W) the frame the world model generated.previousTensor(B, C, H*f, W*f) the previous full-resolution frame, so
the sharpening is temporally consistent rather than
independent per frame.quantizebool= True, keyword-onlyFalse to keep a gradient — the
quantiser has none.Returns
Tensor(B, C, H*f, W*f).
Raises
ValueErrorExamples
>>> import lucid
>>> from lucid.models import diamond_csgo
>>> model = diamond_csgo(
... unet_channels=(8, 8), unet_layers=(1, 1), cond_dim=16,
... attn_depths=(0, 0), upsampler_channels=(4, 4),
... upsampler_layers=(1, 1), upsampler_attn_depths=(0, 0),
... num_actions=4).eval()
>>> low = lucid.randn((1, 3, 30, 56))
>>> noised = lucid.randn((1, 3, 150, 280))
>>> previous = lucid.randn((1, 3, 150, 280))
>>> with lucid.no_grad():
... full = model.upsample(noised, lucid.tensor([1.0]), low, previous)
>>> full.shape
(1, 3, 150, 280)
One denoiser evaluation at one noise level — upsample_frame
walks it down a schedule — quantised back onto [-1, 1]:
>>> bool(full.abs().max() <= 1.0)
Trueupsample_frame(low_res: Tensor, previous: Tensor | None = None, steps: int | None = None, noise: Tensor | None = None)Sample a full-resolution frame from a generated low-resolution one.
Parameters
low_resTensor(B, C, H, W) — what imagine_frame produced.(B, C, H*f, W*f) previous full-resolution frame. Absent
on the first frame of a rollout, where the low-resolution one
scaled up stands in for it.stepsint or None= NoneReturns
Tensor(B, C, H*f, W*f).
Raises
ValueErrorsteps is not
positive.Notes
⚠️ The upsampler appears in the released CS:GO configuration and nowhere in the paper. It is why that experiment is affordable: the world model diffuses at 30x56 and this brings the frame to 150x280, rather than paying full-resolution diffusion for detail a cheaper network can add.
Examples
>>> import lucid
>>> from lucid.models import diamond_csgo
>>> model = diamond_csgo(
... unet_channels=(8, 8), unet_layers=(1, 1), cond_dim=16,
... attn_depths=(0, 0), upsampler_channels=(4, 4),
... upsampler_layers=(1, 1), upsampler_attn_depths=(0, 0),
... num_actions=4).eval()
>>> low = lucid.randn((1, 3, 30, 56))
>>> with lucid.no_grad():
... full = model.upsample_frame(low, steps=1)
>>> full.shape
(1, 3, 150, 280)