Improving Robotic Manipulation with Grounded World Models

# Illustrative sketch (not from the paper)
import torch, torch.nn as nn

class PSGJEPA(nn.Module):
    def __init__(self, latent_dim, proprio_dim):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Conv2d(3, 32, 3, stride=2), nn.ReLU(),
            nn.Flatten(), nn.Linear(32*15*15, latent_dim)
        )
        self.predictor = nn.Sequential(
            nn.Linear(latent_dim + 6, latent_dim), nn.ReLU()
        )
        # static state‑grounding head
        self.state_head = nn.Linear(latent_dim, proprio_dim)
        # transition‑grounding head (predict delta joint angles)
        self.trans_head = nn.Linear(latent_dim*2, proprio_dim)

    def forward(self, img, action, img_next):
        z = self.encoder(img)                     # current latent
        z_next = self.encoder(img_next)           # next latent
        pred_z = self.predictor(torch.cat([z, action], dim=-1))
        # grounding predictions
        state_pred = self.state_head(z)           # proprio from single latent
        trans_pred = self.trans_head(torch.cat([z, z_next], dim=-1))
        return pred_z, state_pred, trans_pred

# ----- training sketch -----
model = PSGJEPA(latent_dim=128, proprio_dim=7)
opt = torch.optim.Adam(model.parameters(), lr=1e-3)

for batch in loader:  # each batch: img, action, img_next, proprio, joint_delta
    pred_z, state_pred, trans_pred = model(batch.img, batch.action, batch.img_next)
    loss_fwd = ((pred_z - model.encoder(batch.img_next))**2).mean()
    loss_state = ((state_pred - batch.proprio)**2).mean()
    loss_trans = ((trans_pred - batch.joint_delta)**2).mean()
    loss = loss_fwd + loss_state + loss_trans
    opt.zero_grad(); loss.backward(); opt.step()

# discard grounding heads for inference
model.state_head = None
model.trans_head = None

// Illustrative sketch (not from the paper)
const tf = require('@tensorflow/tfjs-node');

class PSGJEPA {
  constructor(latentDim, proprioDim) {
    this.encoder = tf.sequential();
    this.encoder.add(tf.layers.conv2d({filters:32, kernelSize:3, strides:2, activation:'relu', inputShape:[64,64,3]}));
    this.encoder.add(tf.layers.flatten());
    this.encoder.add(tf.layers.dense({units:latentDim}));

    this.predictor = tf.sequential();
    this.predictor.add(tf.layers.dense({units:latentDim, activation:'relu', inputShape:[latentDim+6]}));

    // static state‑grounding head
    this.stateHead = tf.layers.dense({units:proprioDim});
    // transition‑grounding head
    this.transHead = tf.layers.dense({units:proprioDim});
  }

  forward(img, action, imgNext) {
    const z = this.encoder.apply(img);
    const zNext = this.encoder.apply(imgNext);
    const predZ = this.predictor.apply(tf.concat([z, action], -1));
    const statePred = this.stateHead.apply(z);
    const transPred = this.transHead.apply(tf.concat([z, zNext], -1));
    return {predZ, statePred, transPred};
  }
}

// ----- training sketch -----
(async () => {
  const model = new PSGJEPA(128, 7);
  const optimizer = tf.train.adam(1e-3);

  for await (const batch of dataLoader()) { // batch: img, action, imgNext, proprio, jointDelta
    optimizer.minimize(() => {
      const {predZ, statePred, transPred} = model.forward(batch.img, batch.action, batch.imgNext);
      const lossFwd = tf.losses.meanSquaredError(model.encoder.apply(batch.imgNext), predZ);
      const lossState = tf.losses.meanSquaredError(batch.proprio, statePred);
      const lossTrans = tf.losses.meanSquaredError(batch.jointDelta, transPred);
      return lossFwd.add(lossState).add(lossTrans);
    });
  }

  // discard grounding heads for inference
  model.stateHead = null;
  model.transHead = null;
})();