Transformers and Neural State-Space Architectures › Code Implementation

Michael BrenndoerferJune 26, 202614 min read
World Models HandbookAll chapters
Part 5 of 12World-Model Architectures
Chapter 26Transformers and Neural State-Space Architectures › Code Implementation
Reading section 7 of 10 · Code Implementation · 14 minSection 7 of 10 · 14 min

Book contents

Section 7 of 10 · 14 min read

Transformers and Neural State-Space Architectures › Code ImplementationOne section · Sections in this chapter7 / 10 ↓
Reading tools Definitions & inline links

Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.

Article links

Make inline references clickable

Now we build a synthetic, partially observable linear-Gaussian system: a two-dimensional hidden state, scalar control, and noisy scalar observation. A single scalar reading does not identify the state, so observation and control history can improve prediction. A correctly specified Kalman filter would use dynamics- and covariance-dependent prediction/correction updates, generally with time-varying gains before steady state. Because the dynamics are coupled and oscillatory and controls are consequential, a plain running average is not an established approximation to that filter. Environment structure, model capacity, and optimization all influence the comparison below.

We train two models on the same data with the same next-observation objective: a tiny causal transformer with one self-attention block, and a tiny diagonal SSM with a stable, input-independent learned recurrence (a time-invariant member of the family, since its decay, input matrix, and output matrix do not depend on the token, unlike the selective SSMs described above). Then we compare one-step accuracy against recursive rollout error under held-out action sequences, distinguishing open-loop prediction quality from the compounding behavior an agent would experience. Keeping the models tiny is a deliberate methodological choice: at roughly 3,300 parameters for the transformer and 1,200 for the SSM, we can inspect the mechanisms directly. Their capacities still differ, so the comparison does not isolate architecture from scale.

First, the system and dataset. We simulate trajectories of

st+1=Ast+But+wt,yt+1=Cst+1+vt+1,s_{t+1} = A s_t + B u_t + w_t, \qquad y_{t+1} = C s_{t+1} + v_{t+1},

where:

  • st∈R2s_t \in \mathbb{R}^2: the latent state, hidden from the learner
  • utu_t: the scalar control input chosen by the experimenter
  • yty_t: the noisy scalar observation
  • A,B,CA, B, C: the transition, input, and observation matrices, matching the system used in the simulation code below
  • wtw_t: process noise added to the state transition
  • vt+1v_{t+1}: observation noise added to the sensor reading at time t+1t+1

The observation matrix CC projects the two-dimensional state onto a single scalar with a fixed linear combination of the two components, so the observation is a lossy, noise-corrupted shadow of the state. The eigenvalues of AA have magnitude slightly below one, so the latent dynamics are stable and gently contract, and the off-diagonal entries couple the two channels so that neither component evolves independently. Tokens are (yt,ut)(y_t, u_t) pairs and the target at position tt is yt+1y_{t+1}, so every model is trained with teacher forcing under the causal objective from the previous section. Position 0 is a start-of-sequence exception: the simulator never emits y0y_0, so the first token carries a placeholder zero observation paired with u0u_0: it is (0,u0)(0, u_0), not a real observation pair, and genuine observation-control pairs begin at position 1.

In[4]:
Code
import numpy as np
import torch
import torch.nn as nn

torch.manual_seed(0)
rng = np.random.default_rng(0)
A_sys = np.array([[0.97, 0.10], [-0.10, 0.96]])
B_sys = np.array([[0.6], [0.1]])
C_sys = np.array([[1.0, 0.6]])
T = 32


def simulate(n_traj, rng):
    us = rng.normal(size=(n_traj, T, 1))
    xs = np.zeros((n_traj, T + 1, 2))
    ys = np.zeros((n_traj, T + 1, 1))
    for t in range(T):
        xs[:, t + 1] = (
            xs[:, t] @ A_sys.T
            + us[:, t] * B_sys.T
            + rng.normal(0, 0.02, size=(n_traj, 2))
        )
        ys[:, t + 1] = xs[:, t + 1] @ C_sys.T + rng.normal(
            0, 0.1, size=(n_traj, 1)
        )
    return ys, us


train_ys, train_us = simulate(512, rng)
test_ys, test_us = simulate(128, rng)


def make_tensors(ys, us):
    x = torch.tensor(
        np.concatenate([ys[:, :T], us], axis=2), dtype=torch.float32
    )
    target = torch.tensor(ys[:, 1:], dtype=torch.float32)
    return x, target


x_train, y_train = make_tensors(train_ys, train_us)
x_test, y_test = make_tensors(test_ys, test_us)

Apart from the zero start marker, each token carries a scalar observation and its aligned action; the target is the next observation. Optimal conditional prediction can use history without recovering the hidden state exactly, and fresh process and sensor noise prevent perfect forecasts. Training and test trajectories use the same simulator and independent control/noise laws. Recursive evaluation changes observation feedback after its prefix; its start position and prefix-length distribution also differ from the pooled all-position teacher-forced score. We therefore compare conditioning regimes without claiming to isolate every causal source of their difference.

Next, the two architectures. The transformer uses a single-head causal attention block, learned positional embeddings, a feed-forward sublayer, and a linear head. The SSM uses a real diagonal recurrence with input-independent BB and CC. We clamp each sigmoid decay to [ϵ,1−ϵ][\epsilon,1-\epsilon], with ϵ=10−4\epsilon=10^{-4}, so sigmoid saturation cannot produce a unit multiplier in this float32 implementation. With bounded finite writes, the uniformly contractive recurrence has a bounded state in exact arithmetic. Finite-precision overflow is still possible at extreme scales; the bound is not a guarantee for other dtypes, unbounded inputs, nonfinite parameters, or the entire training process. A learned skip connection passes the current embedding to the output without routing it through recurrent memory.

In[5]:
Code
class TinyTransformer(nn.Module):
    def __init__(self, d_in=2, d_model=16, nhead=1, d_ff=32, max_len=64):
        super().__init__()
        self.embed = nn.Linear(d_in, d_model)
        self.pos = nn.Parameter(torch.randn(1, max_len, d_model) * 0.02)
        self.attn = nn.MultiheadAttention(d_model, nhead, batch_first=True)
        self.mlp = nn.Sequential(
            nn.Linear(d_model, d_ff), nn.GELU(), nn.Linear(d_ff, d_model)
        )
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.head = nn.Linear(d_model, 1)

    def forward(self, x):
        L = x.size(1)
        h = self.embed(x) + self.pos[:, :L]
        mask = torch.triu(torch.ones(L, L, dtype=torch.bool), diagonal=1)
        a, w = self.attn(
            h,
            h,
            h,
            attn_mask=mask,
            need_weights=True,
            average_attn_weights=True,
        )
        h = self.norm1(h + a)
        h = self.norm2(h + self.mlp(h))
        return self.head(h), w


class TinySSM(nn.Module):
    def __init__(self, d_in=2, d_state=16, d_model=32):
        super().__init__()
        self.embed = nn.Linear(d_in, d_model)
        self.logit_decay = nn.Parameter(torch.full((d_state,), 2.2))
        self.B = nn.Parameter(torch.randn(d_state, d_model) * 0.1)
        self.C = nn.Parameter(torch.randn(d_model, d_state) * 0.1)
        self.skip = nn.Parameter(torch.zeros(d_model))
        self.head = nn.Linear(d_model, 1)

    def forward(self, x):
        z = self.embed(x)
        decay = torch.sigmoid(self.logit_decay).clamp(1e-4, 1 - 1e-4)
        h = torch.zeros(z.size(0), self.B.size(0))
        outs = []
        for t in range(z.size(1)):
            h = decay * h + z[:, t] @ self.B.T
            outs.append(h @ self.C.T + self.skip * z[:, t])
        return self.head(torch.stack(outs, dim=1)), None


transformer = TinyTransformer()
ssm = TinySSM()

A few implementation details connect directly to the chapter's formalism. In the transformer, the mask is built with torch.triu above the diagonal, so position ii is forbidden from attending to positions j>ij > i, the causal constraint rendered as a boolean matrix. The forward method returns the attention weights as well as the predictions, which we will use later for the diagnostic plot; average_attn_weights=True averages over heads into a single weight matrix; the model as defined uses a single head (nhead=1), so this averaging is a no-op and the returned matrix is that head's attention directly. In the SSM, the loop over time steps is the explicit recurrence ht=Aˉht−1+Bˉxth_t = \bar{A} h_{t-1} + \bar{B} x_t, written out step by step for clarity rather than hidden inside a parallel scan; on this short sequence the loop is harmless, but on longer sequences, parallel scan or convolution can avoid this Python-level sequential loop. For fixed state size, a work-efficient scan has linear total work and logarithmic parallel depth, while FFT convolution takes O(Tlog⁡T)O(T\log T) work. Both models are small (roughly 3,300 and 1,200 parameters respectively), but their unequal capacities remain a confound. Both use full-batch Adam and the same learning rate and epoch count; matching that recipe does not guarantee equally effective optimization.

In[6]:
Code
def train_model(model, x, target, epochs=80, lr=3e-3):
    opt = torch.optim.Adam(model.parameters(), lr=lr)
    losses = []
    for _ in range(epochs):
        opt.zero_grad()
        pred, _ = model(x)
        loss = torch.mean((pred - target) ** 2)
        loss.backward()
        opt.step()
        losses.append(loss.item())
    return losses


loss_tr = train_model(transformer, x_train, y_train)
loss_ssm = train_model(ssm, x_train, y_train)

with torch.no_grad():
    pred_tr, _ = transformer(x_test)
    pred_ssm, _ = ssm(x_test)
one_step_tr = torch.mean((pred_tr - y_test) ** 2).item()
one_step_ssm = torch.mean((pred_ssm - y_test) ** 2).item()
Out[7]:
Console
final train loss  transformer: 0.06859   SSM: 0.05490
one-step test MSE transformer: 0.06768
one-step test MSE SSM:         0.05626

Both models learn a useful one-step predictor in this run; the printed test errors are measured rather than assumed. The observation-noise variance is 0.12=0.010.1^2=0.01, but that is not the full achievable one-step error: process noise and uncertainty about the hidden state also matter. We do not implement a Kalman or fitted linear baseline here, so these two scores do not establish how close either model is to optimal. The environment has coupled, oscillatory dynamics, while our SSM has only real positive decay channels; it is not the exact model class for this system. To examine recursive prediction, we warm-start each model with the start token (0,u0)(0,u_0) and the first genuine observation token (y1,u1)(y_1,u_1), preserving their training positions. The first forecast targets y2y_2. Each subsequent token pairs the predicted observation at time tt with the known control utu_t, and the prediction is scored against the raw held-out observation at time t+1t+1. After the warm start, no new real observation enters the prefix.

In[8]:
Code
def rollout_errors(model, y_true, u_seq, horizon=16, start=1):
    """Forecast y[start+1:start+horizon+1] from the recorded prefix through start.

    y_true: raw (N, T+1, 1) observations, not shifted training targets.
    u_seq: (N, T, 1) controls. Token t is (y_t, u_t).
    """
    if y_true.ndim != 3 or u_seq.ndim != 3:
        raise ValueError("Expected batched observation and control sequences")
    if y_true.shape != (u_seq.shape[0], u_seq.shape[1] + 1, 1):
        raise ValueError("Use raw T+1 observations, not shifted targets")
    if (
        u_seq.shape[2] != 1
        or start < 0
        or horizon < 1
        or start + horizon > u_seq.shape[1]
    ):
        raise ValueError("Forecast must stay inside the control sequence")
    observations = torch.as_tensor(y_true, dtype=torch.float32)
    controls = torch.as_tensor(u_seq, dtype=torch.float32)
    prefix = torch.cat(
        [observations[:, : start + 1], controls[:, : start + 1]], dim=2
    )
    errs = np.zeros(horizon)
    with torch.no_grad():
        for step in range(horizon):
            t = start + step
            out, _ = model(prefix)
            y_next = out[:, -1:, :]
            errs[step] = torch.mean(
                (y_next - observations[:, t + 1 : t + 2]) ** 2
            ).item()
            if step + 1 < horizon:
                token = torch.cat([y_next, controls[:, t + 1 : t + 2]], dim=2)
                prefix = torch.cat([prefix, token], dim=1)
    return errs


horizon = 16
rollout_tr = rollout_errors(transformer, test_ys, test_us, horizon)
rollout_ssm = rollout_errors(ssm, test_ys, test_us, horizon)
print("rollout MSE transformer:", np.round(rollout_tr, 4))
print("rollout MSE SSM:", np.round(rollout_ssm, 4))
Out[8]:
Console
rollout MSE transformer: [0.0295 0.0419 0.0489 0.0524 0.0743 0.0843 0.1025 0.1374 0.1634 0.1697
 0.2057 0.2134 0.1962 0.1799 0.1963 0.198 ]
rollout MSE SSM: [0.0415 0.0664 0.0871 0.1058 0.115  0.1282 0.108  0.126  0.1089 0.1102
 0.1126 0.0997 0.0753 0.0713 0.0564 0.0639]

Two details matter for interpreting this procedure. The warm start consists of the zero start marker and the first genuine observation, each paired with its aligned control; later observation components are predictions. Known controls remain aligned with the timestep of each token. Thus the experiment changes observation feedback after the prefix without shifting the initial token positions or pairing observations with the wrong actions.

We also inspect how the trained transformer distributes attention across temporal lags. To avoid mixing query positions with different available context lengths, every example contributes only its final query, which can access all 32 tokens. We compare the mean weights with uniform attention over that same context. This is a descriptive memory diagnostic, not proof that the model performs Bayesian filtering or uses the most informative observations.

In[9]:
Code
with torch.no_grad():
    _, attn = transformer(x_test)  # (N, L, L), averaged over heads


def matched_query_lag_weights(weights):
    """Mean weights at the final query, ordered from lag 0 to lag L-1."""
    if (
        weights.ndim != 3
        or weights.shape[0] == 0
        or weights.shape[1] != weights.shape[2]
        or weights.shape[1] == 0
    ):
        raise ValueError(
            "Expected nonempty (batch, length, length) attention weights"
        )
    return weights[:, -1, :].mean(dim=0).flip(0).detach().cpu().numpy()


L = attn.size(-1)
lag_weights = matched_query_lag_weights(attn)
uniform_lag_weights = np.full(L, 1.0 / L)

At the final query i=L−1i=L-1, lag kk selects key j=L−1−kj=L-1-k. Reversing that query's key axis places self-attention at lag zero, and averaging across held-out sequences keeps query length fixed. The complete LL-lag profile sums to one up to numerical error. The plot displays only its first eight entries, not a renormalized eight-token distribution. Its dashed reference is 1/L1/L, uniform attention over all available tokens.

Pooling all query positions would introduce a baseline effect: even uniform causal attention has weights 1/(i+1)1/(i+1) at query ii, so averaging its eligible pairs at each lag produces a declining curve. The matched-query construction avoids that particular confound. We can now ask whether this final-query profile favors recent tokens relative to uniform attention, while remembering that learned values, residual paths, and other components also determine predictions.

Out[10]:
Visualization
Eight lag-weight bars for queries with equal context length, compared with a horizontal uniform-attention reference.
Mean attention weight at the final query of each held-out sequence, plotted by lag. Every query has 32 available tokens; the dashed line is uniform attention over that same context. Only the first eight lags are shown. Lag zero is self-attention, and this profile alone does not establish belief tracking.

The environment's coupled oscillatory transition is not exactly representable by this real positive-decay SSM. The model can still fit useful finite-horizon predictions, but an observed advantage would not prove an exact model-class match. Nor does this experiment isolate architecture alone: the models have different parameter counts and parameterizations, and identical optimizer settings need not suit them equally. It is one controlled data-generating task and one training seed, not evidence of a consistent ranking across environments.

The two rollout curves below plot recursive open-loop MSE against horizon, with each model’s all-position teacher-forced MSE as a dashed reference.

Out[11]:
Visualization
Transformer and SSM recursive forecast errors across sixteen horizons, with dashed all-position teacher-forced error references.
Recursive open-loop MSE after the start token (0, u0) and first observation token (y1, u1). Dashed lines show teacher-forced MSE averaged over all test positions; they are descriptive references, not horizon-matched lower bounds.

These rollout curves measure prediction error after genuine observations stop arriving. Growth can reflect both model error and irreducible multi-step uncertainty from process noise. The dashed lines summarize teacher-forced error over all test positions; they are not horizon-matched noise floors, so a crossing is a descriptive comparison rather than a theorem about compounding. Both models condition on their own predicted observations after the warm start. The transformer can still attend to the supplied start token and first observation, but it cannot retrieve later observations that were never supplied. Inspect the printed curves for this run rather than assuming monotonic growth or a stable ordering between architectures.

Separating the quality axes prevents overreading this experiment. Teacher-forced test MSE measures one-step predictive fidelity, and the recursive curves measure a different conditioning regime. Neither deterministic point predictor supplies a calibrated uncertainty distribution, so we have not evaluated uncertainty quality. No controller acts on the forecasts, so we have not measured decision usefulness either. Good performance on either plotted error metric does not settle those remaining questions.

The attention-by-lag plot can suggest hypotheses about retrieval or integration, but it cannot uniquely locate a failure. Query formation, key/value representations, residual paths, and dynamics layers can produce similar visible weight patterns. Test a suspected mechanism by intervening on relevant tokens, probing state information, or ablating components before selecting a remedy. Here the model explicitly returns weights, making inspection convenient; in other implementations requesting them can add memory or prevent use of an optimized attention path.