Skip to content

Quick start

This page trains a small world model from pixels and uses it to reach goal images. It takes a few minutes on a laptop CPU.

1. Collect data

Environments in torch_world.envs are batched and render RGB images. collect plays random actions and returns episodes as tensors.

import torch
from torch.utils.data import DataLoader

from torch_world.datasets import TrajectoryDataset
from torch_world.envs import PointMass

torch.manual_seed(0)
data = PointMass(batch_size=2048, size=32).collect(16)
# data["pixels"]  (2048, 16, 3, 32, 32)
# data["actions"] (2048, 16, 2)   actions[:, t] is applied at states[:, t]
# data["states"]  (2048, 16, 2)

loader = DataLoader(TrajectoryDataset(data, window=4), batch_size=128, shuffle=True, drop_last=True)

TrajectoryDataset cuts every episode into sliding windows of window steps.

2. Build a model

LeWM needs an encoder that maps one image (N, C, H, W) to one embedding (N, D). CLSEncoder is a ViT whose CLS token does exactly that.

from torch_world.models import LeWM
from torch_world.nn import CLSEncoder

encoder = CLSEncoder(img_size=32, patch_size=8, embed_dim=64, depth=3, num_heads=4)
model = LeWM(encoder, action_dim=2, embed_dim=64, history=3, depth=3, num_heads=4, head_dim=16,
             mlp_dim=256, proj_dim=256, lamb=0.01)

3. Train

Every model returns a dict with a scalar "loss" and its components.

import itertools

from torch_world.utils import recompute_batchnorm

opt = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-3)
for epoch in range(40):
    for batch in loader:
        out = model(batch["pixels"], batch["actions"])  # {"loss", "pred_loss", "sigreg_loss"}
        opt.zero_grad()
        out["loss"].backward()
        opt.step()

recompute_batchnorm(model, itertools.islice(loader, 50), lambda b: model(b["pixels"], b["actions"]))
model.eval()

recompute_batchnorm re-estimates the BatchNorm statistics of the projectors before evaluation; see Training.

4. Plan

CEM searches for action sequences whose imagined final latent is closest to the latent of a goal image. In closed loop, execute the first action and plan again.

from torch_world.envs import run_policy
from torch_world.planning import CEM

env = PointMass(batch_size=64, size=32)
env.reset()
goals = env.sample_goals((0.2, 0.5))
with torch.no_grad():
    goal = model.encode(env.render(goals)[:, None])[:, 0]

planner = CEM(horizon=5, action_dim=2, num_samples=512, num_elites=48, iterations=10, bounds=(-1, 1))
paths = run_policy(env, lambda obs: planner(model, model.encode(obs[:, None]), goal)[:, 0], steps=20)

success = ((paths[:, -1] - goals).norm(dim=-1) < 0.1).float().mean()

Next

  • Concepts: the two base classes and the shape conventions.
  • Models: every model, what it predicts and how to call it.
  • LeWM notebook: this page with figures, probes and results.