Improving Robot Action Prediction with Semantic Foresight

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

class RobustWAM(nn.Module):
    def __init__(self, K, dim, lambda_align):
        super().__init__()
        # pretrained video generation model (VAE latent space) – placeholder
        self.vgm = torch.hub.load('repo/vgm', 'model')
        # original action Diffusion Transformer (DiT) – unchanged
        self.action_dit = torch.hub.load('repo/action_dit', 'model')
        # frozen DINOv3 teacher for semantic foresight
        self.dino = torch.hub.load('repo/dinov3', 'model')
        self.dino.eval()
        for p in self.dino.parameters():
            p.requires_grad = False
        # learnable query tokens
        self.queries = nn.Parameter(torch.randn(K, dim))
        self.lambda_align = lambda_align
        # simple sinusoidal positional encoding function
        self.pos_enc = lambda pos, d: torch.sin(pos * torch.arange(d).float() * 0.01)

    def forward(self, action_tokens, future_frames, original_wam_loss):
        # prepend K queries to the noised action tokens
        K = self.queries.shape[0]
        # assign temporal positional codes to each query (same step it predicts)
        pos_codes = torch.stack([self.pos_enc(i, self.queries.shape[1]) for i in range(K)])
        queries = self.queries + pos_codes
        seq = torch.cat([queries, action_tokens], dim=0)
        # process through unchanged DiT blocks
        hidden = self.action_dit(seq)
        # extract hidden states corresponding to queries (first K positions)
        query_hidden = hidden[:K]
        # compute frozen DINO CLS embeddings of ground‑truth future frames
        with torch.no_grad():
            dino_emb = self.dino(future_frames)  # shape (K, dim)
        # cosine similarity loss per query
        cos = F.cosine_similarity(query_hidden, dino_emb, dim=-1)
        align_loss = (1 - cos).mean()
        # total loss
        loss = original_wam_loss + self.lambda_align * align_loss
        return loss, hidden

    def inference(self, action_tokens):
        # at inference we drop DINO and the alignment head, keep queries
        K = self.queries.shape[0]
        pos_codes = torch.stack([self.pos_enc(i, self.queries.shape[1]) for i in range(K)])
        queries = self.queries + pos_codes
        seq = torch.cat([queries, action_tokens], dim=0)
        hidden = self.action_dit(seq)
        # return action predictions (e.g., last tokens after queries)
        return hidden[K:]

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

class RobustWAM {
  constructor(K, dim, lambdaAlign) {
    // placeholders for pretrained models
    this.vgm = loadModel('vgm'); // VAE latent space of video generation
    this.actionDit = loadModel('action_dit'); // unchanged DiT
    this.dino = loadModel('dinov3'); // frozen DINOv3 teacher
    this.dino.trainable = false;
    // learnable query tokens
    this.queries = tf.variable(tf.randomNormal([K, dim]));
    this.lambdaAlign = lambdaAlign;
  }

  // simple sinusoidal positional encoding
  posEnc(pos, dim) {
    const i = tf.range(0, dim, 1, 'float32');
    return tf.sin(tf.mul(pos, tf.mul(i, 0.01)));
  }

  // forward pass during training
  async forward(actionTokens, futureFrames, originalWamLoss) {
    const K = this.queries.shape[0];
    // prepend queries with temporal positional codes
    const posCodes = [];
    for (let i = 0; i < K; i++) {
      posCodes.push(this.posEnc(i, this.queries.shape[1]));
    }
    const posStack = tf.stack(posCodes);
    const queries = tf.add(this.queries, posStack);
    const seq = tf.concat([queries, actionTokens], 0);
    // process through unchanged DiT blocks
    const hidden = this.actionDit.apply(seq);
    const queryHidden = hidden.slice([0, 0], [K, -1]);
    // frozen DINO CLS embeddings of ground‑truth future frames
    const dinoEmb = tf.tidy(() => {
      return tf.stopGradient(this.dino.apply(futureFrames));
    });
    // cosine similarity loss per query
    const cos = tf.sum(tf.mul(tf.l2Normalize(queryHidden, -1), tf.l2Normalize(dinoEmb, -1)), -1);
    const alignLoss = tf.mean(tf.sub(1, cos));
    // total loss
    const loss = tf.add(originalWamLoss, tf.mul(this.lambdaAlign, alignLoss));
    return { loss, hidden };
  }

  // inference – drop DINO and alignment head, keep queries
  async inference(actionTokens) {
    const K = this.queries.shape[0];
    const posCodes = [];
    for (let i = 0; i < K; i++) {
      posCodes.push(this.posEnc(i, this.queries.shape[1]));
    }
    const posStack = tf.stack(posCodes);
    const queries = tf.add(this.queries, posStack);
    const seq = tf.concat([queries, actionTokens], 0);
    const hidden = this.actionDit.apply(seq);
    // return action predictions (tokens after the K queries)
    return hidden.slice([K, 0], [-1, -1]);
  }
}

function loadModel(name) {
  // placeholder loader – in practice replace with actual model loading code
  return { apply: (x) => x, trainable: true };
}