#!/usr/bin/env python3 """Stateful truncated BPTT training loop for the RNN stack. - layer_0.seq_windows(): B contiguous tapes walked in order - RNN state carried across windows but detached at every boundary: gradients are truncated to one window, while information flows forward through the carried state (the k1 == k2 == T form of truncated BPTT) - state reset to zeros only at sweep boundaries (the one real discontinuity) Usage: python3 src/train_seq.py # train (defaults: 1 epoch, cpu/cuda auto) python3 src/train_seq.py --check # prove truncation + statefulness python3 src/train_seq.py --epochs 3 --lr 3e-4 --save out.pt """ import argparse import math import sys import time from itertools import islice from pathlib import Path import numpy as np import torch import torch.nn.functional as F PROJ = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJ)) sys.path.insert(0, str(PROJ / "scripts")) import tokenizer as tk from model import Notio, NotioConfig tk.load_vocab() # also asserts vocab.txt == vocab.bin at train start def detach_states(states): """Detach every hidden/cell tensor: no gradient flows across windows. Handles LSTM (h, c) tuples and GRU bare tensors.""" out = [] for st in states: out.append(tuple(s.detach() for s in st) if isinstance(st, tuple) else st.detach()) return out def ids_to_display(ids): """id tensor -> human text via the runtime decoder (tags dropped/mapped).""" out = bytes(ids.cpu().numpy().astype(np.uint8)).translate(tk.TABLE_DEC) for sent, tok in tk.SENT_DEC: out = out.replace(sent, tok) return tk.display(out).decode() @torch.no_grad() def generate(m, device, max_tokens=256, temperature=0.9, top_k=0): """Sample one story: prompt, carry RNN state, stop at .""" m.eval() ids = torch.tensor([[1]], dtype=torch.long, device=device) # states = None pos = 0 for _ in range(max_tokens): logits, states = m(ids[:, -1:], states, pos_offset=pos) states = detach_states(states) pos += 1 logits = logits[:, -1, :] / max(temperature, 1e-6) if top_k > 0: v, _ = torch.topk(logits, top_k) logits = torch.where(logits < v[:, -1:], torch.full_like(logits, -float("inf")), logits) probs = F.softmax(logits, dim=-1) nxt = torch.multinomial(probs, 1) ids = torch.cat([ids, nxt], dim=1) if nxt.item() == 2: # break m.train() return ids def sample_and_print(m, device, n=2): for i in range(n): ids = generate(m, device) print(f"--- sample {i} ({ids.numel()} ids) ---") print(ids_to_display(ids)) @torch.no_grad() def val_loss(m, device, n_blocks=20): """Mean next-token loss over the first n_blocks contiguous val blocks, carrying state across them (mirrors training).""" m.eval() states = None total = 0.0 n = min(n_blocks, m.layer0.n_val_blocks) for i in range(n): x, y = m.layer0.val_block(i) x, y = x.unsqueeze(0).to(device), y.unsqueeze(0).to(device) logits, states = m(x, states) total += F.cross_entropy(logits.view(-1, m.cfg.head.vocab_size), y.view(-1)).item() states = detach_states(states) m.train() return total / n def lr_at(step, lr, warmup, total, min_lr): """Linear warmup, then cosine decay to min_lr over [warmup, total).""" if step < warmup: return lr * step / max(warmup, 1) if step >= total: return min_lr prog = (step - warmup) / max(total - warmup, 1) return min_lr + 0.5 * (lr - min_lr) * (1 + math.cos(math.pi * prog)) def check(): torch.manual_seed(0) m = Notio(NotioConfig()) m.train() it = m.layer0.seq_windows() x1, y1 = next(it) x2, y2 = next(it) # 1) statefulness: carried state changes the second window's output _, st = m(x1) logits_carry, _ = m(x2, st) logits_zero, _ = m(x2) assert not torch.equal(logits_carry, logits_zero), "state is not used" # 2) truncation: detach cuts the graph back to window 1. # (values still flow - that is the point of carrying state) V = m.cfg.head.vocab_size assert st[0][0].grad_fn is not None, "state should carry history to window 1" st_d = detach_states(st) assert st_d[0][0].grad_fn is None, "detached state should be a leaf" logits_d, _ = m(x2, st_d) assert torch.equal(logits_carry, logits_d), "detach changed the forward pass" def pgrad(states_): m.zero_grad() logits, _ = m(x2, states_) F.cross_entropy(logits.view(-1, V), y2.view(-1)).backward() return sum(p.grad.detach().abs().sum() for p in m.parameters() if p.grad is not None) gd = pgrad(st_d) # detached: graph stops at the carried state gu = pgrad(st) # undetached: gradient also flows back through window 1 assert torch.equal(gd, gu) is False, "detach did not truncate the gradient path" # 3) tape continuity: window w+1 continues window w inside a tape assert torch.equal(x2[:, 0], y1[:, -1]), "tapes are not contiguous" # 4) a few real optimizer steps with carried+detached state opt = torch.optim.AdamW(m.parameters(), lr=1e-3) states = None it2 = m.layer0.seq_windows() t0 = time.time() for step in range(3): x, y = next(it2) logits, states = m(x, states) loss = F.cross_entropy(logits.view(-1, m.cfg.head.vocab_size), y.view(-1)) opt.zero_grad() loss.backward() opt.step() states = detach_states(states) print(f"step {step}: loss {loss.item():.3f}") toks = 3 * x.numel() print(f"truncation: True | statefulness: True | tape continuity: True") print(f"{toks / (time.time() - t0):,.0f} tokens/s (CPU, {m.n_params:,} params, " f"T={m.cfg.layer0.block_size}, B={m.cfg.layer0.batch_size}, d={m.cfg.layer1.d_model}, " f"n_blocks={m.cfg.n_blocks})") def main(device, epochs, lr, log_every, save, max_steps, save_every, warmup, min_lr, resume, val_blocks): torch.backends.cudnn.benchmark = True if resume: ck = torch.load(resume, map_location=device, weights_only=False) # our own file, contains NotioConfig m = Notio(ck["cfg"]) m.load_state_dict(ck["state_dict"]) m.to(device) start = ck.get("step", 0) print(f"resumed {resume} @ step {start} " f"(loss {ck.get('loss', float('nan')):.3f}, " f"val loss {ck.get('val_loss', float('nan')):.3f})") else: m = Notio(NotioConfig()) m.to(device) start = 0 m.train() opt = torch.optim.AdamW(m.parameters(), lr=lr) total = max_steps if max_steps else epochs * m.layer0.n_seq_windows states = None step = start t0 = time.time() it = islice(m.layer0.seq_windows(), start, None) # continue the sweep position for epoch in range(epochs if max_steps is None else 1): src = islice(it, max_steps) if max_steps is not None else it for x, y in src: new_lr = lr_at(step, lr, warmup, total, min_lr) for g in opt.param_groups: g["lr"] = new_lr x, y = x.to(device), y.to(device) logits, states = m(x, states) loss = F.cross_entropy(logits.view(-1, m.cfg.head.vocab_size), y.view(-1)) opt.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(m.parameters(), 1.0) opt.step() states = detach_states(states) # truncate gradients at window length T step += 1 if step % log_every == 0: tok = step * x.numel() print(f"epoch {epoch} step {step}: loss {loss.item():.3f} | " f"{tok / (time.time() - t0):,.0f} tok/s | lr {new_lr:.2e}") if step % save_every == 0: vl = val_loss(m, device, val_blocks) print(f"val loss @ step {step}: {vl:.3f}") sample_and_print(m, device) if save: torch.save({"step": step, "loss": loss.item(), "val_loss": vl, "lr": new_lr, "window": step, "cfg": m.cfg, "state_dict": m.state_dict()}, save) print(f"checkpoint saved: {save} @ step {step}") states = None # sweep ended: reset at the real discontinuity if max_steps is not None: break it = m.layer0.seq_windows() total_tok = (step - start) * m.cfg.layer0.batch_size * m.cfg.layer0.block_size vl = val_loss(m, device, val_blocks) print(f"done: {step} steps | {total_tok / (time.time() - t0):,.0f} tok/s on {device} " f"| final loss {loss.item():.3f} | val loss {vl:.3f}") if save: torch.save({"step": step, "loss": loss.item(), "val_loss": vl, "lr": new_lr, "window": step, "cfg": m.cfg, "state_dict": m.state_dict()}, save) print(f"saved {save}") if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--check", action="store_true", help="prove truncation + statefulness, then exit") ap.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu") ap.add_argument("--epochs", type=int, default=1) ap.add_argument("--lr", type=float, default=4e-4) ap.add_argument("--log-every", type=int, default=20) ap.add_argument("--max-steps", type=int, default=None, help="benchmark: train this many steps then stop") ap.add_argument("--save", default=None, help="checkpoint path (saved every --save-every steps + at end)") ap.add_argument("--save-every", type=int, default=500) ap.add_argument("--warmup", type=int, default=500) ap.add_argument("--min-lr", type=float, default=4e-5) ap.add_argument("--resume", default=None, help="checkpoint to resume from (continues sweep position)") ap.add_argument("--val-blocks", type=int, default=20) a = ap.parse_args() if a.check: check() else: main(a.device, a.epochs, a.lr, a.log_every, a.save, a.max_steps, a.save_every, a.warmup, a.min_lr, a.resume, a.val_blocks)