A fictional workshop wants to produce trace continuations to test its log parser without using customer documents. Commands in the traces are symbols, never system actions. We build a statistical generator whose errors remain visible, not a machine controller.
A log prefix produces several continuations, compared with an n-gram model. A parser distinguishes a readable trace, an allowed state transition and a copy of a training trace.
Download the complete project, revision r001
Corpus and splits
AtelierTrace contains 500 families of ten traces: 350 training families (3,500 traces), 75 validation families and 75 test families (750 traces each). A family fixes an alias, a length range and a transition preference. Routes contain 6 to 24 events. The generator checks exact duplicates and normalized routes without aliases; it regenerates the whole route on collision, then freezes counts and hashes. A mere identifier change is therefore not considered a new trace. States, commands, short patterns and parameter combinations remain shared: this split does not prove generalization to an unseen language.
The automaton has three states. From off, start leads to ready; from ready, load leads to busy and stop to off; from busy, unload leads to ready and stop to off. Events trigger no system action. The parser rereads the lines and independently checks syntax and transitions; it does not use the event list recorded by the generator as a predicted result.
Tokens, windows and loss
A token is a UTF-8 byte, an integer from 0 to 255. PAD=256, BOS=257 and EOS=258 give a vocabulary of 259 values. No tokenizer is fitted on the data. An accented character can occupy several tokens; NLL is measured in nats per byte/control token, not per character. Control tokens never become text. An invalid byte sequence remains visible and fails the parser.
BOS and EOS are added to every trace after splitting. Windows of 256 tokens overlap by one token, with a stride of 255: every next target appears exactly once, including EOS. A window never crosses two documents. Context restarts at the beginning of every window; this effect is shared by neural NLL and the n-gram. There is no claim of full context beyond 256 tokens.
At position t, logits predict token t+1. The loss selects active targets before cross-entropy, sums their losses and divides by their exact count. Padding adds neither a target nor a denominator. During teacher forcing, preceding inputs are the real tokens. During free generation, the next prefixes include the model's errors.
Attention and learned weights
The original decoder has six pre-LayerNorm blocks, width 256, eight heads and an MLP of 1024 with GELU. Position embeddings are learned up to 256. Q/K/V projections form tensors [batch, heads, time, width_per_head]; softmax(QKᵀ/√d) weights V. The causal mask forbids future keys and the padding mask forbids dummy keys. Fully padded queries receive a fallback key at their own position, then their outputs are ignored; this avoids an all −inf softmax during left-padded generation.
Residuals, normalizations and dropout 0.1 are visible. Dropout is disabled during evaluation. The output projection reuses exactly the input embedding Parameter; the program computes the number of unique parameters. All weights are FP32 CUDA, with TF32 disabled and deterministic algorithms. This profile is neither a general-purpose model nor a reproduction of a large GPT.
Comparisons and training
The order-three n-gram learns next-token counts given the two previous tokens, on the same training windows. Missing context is completed by BOS. Additive smoothing alpha=0.1 covers all 259 values; an unseen context gives a uniform distribution. Its NLL uses exactly the same test targets as the random decoder and the trained decoder.
AdamW uses lr=0.0003, weight_decay=0.01, norm clipping at 1, 800 fixed updates, 80 warm-up steps and then linear decay to 10% of the initial rate. A sampler walks through window permutations and retains its cursor. Each update accumulates complete windows until it reaches at least 4,096 targets, in microbatches of eight windows; the actual count can exceed 4,096 and is the exact denominator. This is not an average of microbatch averages.
Validation is computed every 80 steps, with no checkpoint selection or hyperparameter adaptation. The final checkpoint is fixed at 800. Interrupting at 320 and resuming must reproduce exactly the model, AdamW, scheduler, history, permutation, cursor and Python/NumPy/PyTorch CPU/CUDA random states. The .pt states are created locally and must not be replaced by untrusted files. FP32 weights encoded as JSON with dimensions, sizes and SHA permit an audit without importing an external pickle.
Measuring free outputs
The 750 test traces provide NLL and perplexity exp(NLL) in the same unit. The first 200 test prefixes are frozen by identifier, with three continuation seeds each: 600 outputs per variant. Every prefix contains two real transitions then CMD . Generation uses at most 128 new tokens, temperature 1, with no top-k or top-p. PAD and BOS are removed from support for both models; EOS ends generation. The decoder reads the last 256 tokens with positions reset to zero. Categorical draws use a separate Python generator per case and actually computed probabilities, to keep the same seeds despite outputs of different lengths.
The report distinguishes valid syntax, correct transitions among parsable outputs, overall validity and truncations. Overall validity requires EOS, complete syntax and only allowed transitions. An output that reaches the limit remains an overall failure even if its last byte closes a valid line. UTF-8 errors and unparsable outputs remain in the denominator of 600. Exact copies and copies after removing the alias are counted separately. A test prefix partly shared with training is not a complete copy.
Exercise: look into the future
The unmasked variant starts again from the same initial weights, with the same sampler and the same 800 steps, but allows attention to future positions. The next target can therefore be visible during teacher forcing. Its NLL and 600 generations are compared without requiring an improvement or a degradation. The test changes a suffix and measures earlier logits: the faulty version must reveal an influence, while the causal version must block it. Restoring the mask on the same faulty weights must restore this mechanical property; that is not equivalent to training those weights correctly again. The initial results are protected by SHA.
Interpreting
A better NLL guarantees neither syntax nor correct behaviour in free generation. High validity on this simple automaton demonstrates no ability in general French, workshop control or physical safety. The deterministic simulator is a known alternative that guarantees its own rules; the network studies a distribution, not a machine command. One training seed is run; the three generation seeds are not three independent training runs.
Build a corpus where leaks can be detected
Split by family before windowing
Common error: different aliases hide the same trace; also compare a normalized form without aliases.
Invariant: no family or identical trace crosses splits; actual counts are written to outputs/data-audit.json.
File src/preentrainement_causal/corpus.py
import hashlib, random, re
from pathlib import Path
from .config import save_json, sha
AUTOMATON = {
"off": {"start": "ready"},
"ready": {"load": "busy", "stop": "off"},
"busy": {"unload": "ready", "stop": "off"},
}
def split_families(family):
if not 0 <= family < 500:
raise ValueError("Family outside frozen population")
return "train" if family < 350 else "validation" if family < 425 else "test"
def normalized(text):
return (
"\n".join(line for line in text.splitlines() if not line.startswith("atelier="))
+ "\n"
)
def generate_trace(family, index, attempt, seed):
rng = random.Random(seed * 100000000 + family * 100000 + index * 1000 + attempt)
minimum = 6 + family % 16
maximum = min(24, minimum + 8)
# Retrying changes the route, not just an identifier or an alias.
length = rng.randint(minimum, maximum)
state = "off"
events = []
probability = 0.25 + 0.1 * ((family // 16) % 6)
for position in range(length):
if state == "off":
command = "start"
else:
favored = "load" if state == "ready" else "unload"
command = favored if rng.random() < probability else "stop"
next_state = AUTOMATON[state][command]
events.append({"command": command, "before": state, "after": next_state})
state = next_state
alias = f'{["AX","BY","CZ","DU","EV"][family//100]}{family:03d}'
text = f"atelier={alias}\nINIT off\n" + "".join(
f'CMD {e["command"]}\nSTATE {e["after"]}\n' for e in events
)
return {
"document_id": f"T{family:03d}-{index:02d}",
"family": family,
"split": split_families(family),
"alias": alias,
"text": text,
"events": events,
"generation_attempt": attempt,
"family_parameters": {
"minimum": minimum,
"maximum": maximum,
"favored_transition_probability": probability,
},
"content_sha256": hashlib.sha256(text.encode()).hexdigest(),
"normalized_sha256": hashlib.sha256(normalized(text).encode()).hexdigest(),
}
def audit_duplicates(documents):
for field in ["document_id", "content_sha256", "normalized_sha256"]:
if len({d[field] for d in documents}) != len(documents):
raise ValueError("Duplicate " + field)
splits = {
s: {d["family"] for d in documents if d["split"] == s}
for s in ["train", "validation", "test"]
}
if (
splits["train"] & splits["validation"]
or splits["train"] & splits["test"]
or splits["validation"] & splits["test"]
):
raise ValueError("Family leakage")
return {
"documents": len(documents),
"split_documents": {s: sum(d["split"] == s for d in documents) for s in splits},
"split_families": {s: len(v) for s, v in splits.items()},
"normalized_duplicates": 0,
"exact_duplicates": 0,
"regenerated_candidates": sum(d["generation_attempt"] for d in documents),
"minimum_events": min(len(d["events"]) for d in documents),
"maximum_events": max(len(d["events"]) for d in documents),
}
def make_corpus(config):
documents = []
seen = set()
for family in range(500):
for index in range(10):
for attempt in range(1000):
d = generate_trace(family, index, attempt, config["seed"])
if d["normalized_sha256"] not in seen:
break
else:
raise ValueError(
"Unable to generate a distinct normalized route; revise generator explicitly"
)
seen.add(d["normalized_sha256"])
documents.append(d)
return documents, audit_duplicates(documents)
def parse_trace(text):
# Syntax is checked independently from the generator's event list.
lines = text.splitlines()
result = {
"syntax_valid": False,
"transitions_valid": False,
"valid_transitions": 0,
"transitions": 0,
"reason": "SYNTAX",
}
if (
not text.endswith("\n")
or len(lines) < 4
or not re.fullmatch(r"atelier=[A-Z]{2}[0-9]{3}", lines[0])
or lines[1] != "INIT off"
or (len(lines) - 2) % 2
):
return result
state = "off"
valid = 0
for i in range(2, len(lines), 2):
command = re.fullmatch(r"CMD (start|load|unload|stop)", lines[i])
observed = re.fullmatch(r"STATE (off|ready|busy)", lines[i + 1])
if command is None or observed is None:
return result
correct = AUTOMATON[state].get(command[1]) == observed[1]
valid += int(correct)
state = observed[1]
count = (len(lines) - 2) // 2
return {
"syntax_valid": True,
"transitions_valid": valid == count,
"valid_transitions": valid,
"transitions": count,
"reason": "VALID" if valid == count else "INVALID_TRANSITION",
}
def prepare(config):
documents, audit = make_corpus(config)
root = Path("data/generated")
root.mkdir(parents=True, exist_ok=True)
if (root / "manifest.json").exists():
raise ValueError("Corpus already frozen; use a clean directory")
save_json(root / "input/pages.json", documents)
for split in ["train", "validation", "test"]:
save_json(
root / f"{split}/traces.json", [d for d in documents if d["split"] == split]
)
selected = [d for d in documents if d["split"] == "test"][
: config["generation_prefixes"]
]
prefixes = []
for d in selected:
lines = d["text"].splitlines(keepends=True)
prefix = "".join(lines[:6]) + "CMD "
for repeat in range(config["generations_per_prefix"]):
prefixes.append(
{
"case_id": d["document_id"] + f"-{repeat}",
"document_id": d["document_id"],
"prefix": prefix,
"seed": 410000 + len(prefixes),
}
)
save_json(root / "input/prefixes.json", prefixes)
files = {str(f.relative_to(root)): sha(f) for f in root.rglob("*.json")}
save_json(
root / "manifest.json",
{
"dataset": "AtelierTrace-v1",
"seed": config["seed"],
"rights": "CC0-1.0",
"files": files,
"audit": audit,
},
)
save_json("outputs/data-audit.json", audit)
return audit
Create tokens, targets and the causal mask
Causal shift and padding
Common error: shifting labels twice teaches token t+2; test this on a known sequence of five tokens.
Invariant: decoding reconstructs the text; changing PAD values does not change the loss.
File src/preentrainement_causal/tokens.py
import torch
import torch.nn.functional as F
PAD = 256
BOS = 257
EOS = 258
VOCAB = 259
def encode_bytes(text, controls=True):
data = list(text.encode("utf-8"))
return [BOS] + data + [EOS] if controls else data
def decode_bytes(ids):
if any(type(t) != int or not 0 <= t < VOCAB for t in ids):
raise ValueError("Invalid token")
return bytes(t for t in ids if t < 256).decode("utf-8", errors="strict")
def windows(text, context=256):
if context < 2:
raise ValueError("At least one source and one target required")
ids = encode_bytes(text)
return [
{"start": start, "ids": ids[start : start + context]}
for start in range(0, len(ids) - 1, context - 1)
]
def make_causal_batch(rows, device="cpu", pad_to=None):
if not rows or any(len(r) < 2 for r in rows):
raise ValueError("Empty sequence or missing target")
length = max(map(len, rows)) if pad_to is None else pad_to
if length < max(map(len, rows)):
raise ValueError("No silent truncation")
ids = torch.full((len(rows), length), PAD, dtype=torch.long, device=device)
mask = torch.zeros_like(ids, dtype=torch.bool)
for i, r in enumerate(rows):
ids[i, : len(r)] = torch.tensor(r, device=device)
mask[i, : len(r)] = True
return ids, mask
def token_losses(logits, tokens, mask):
if (
logits.shape[:2] != tokens.shape
or mask.shape != tokens.shape
or mask.dtype != torch.bool
):
raise ValueError("Incompatible shapes")
active = mask[:, 1:] & mask[:, :-1]
if not active.any():
raise ValueError("No active targets")
return (
F.cross_entropy(
logits[:, :-1][active], tokens[:, 1:][active], reduction="none"
),
active,
)
def masked_token_nll(logits, tokens, mask):
values, active = token_losses(logits, tokens, mask)
return values.sum() / active.sum()
Write the decoder and its baseline
Q/K/V, residuals and likelihood
Common error: reversing the mask direction allows the future; inspect a 4×4 matrix, then the logits.
Invariant: causality test passes at the fixed tolerance; parameter count and initial NLL are captured.
File src/preentrainement_causal/model.py
import math, torch
from torch import nn
from .tokens import PAD, VOCAB
class CausalSelfAttention(nn.Module):
def __init__(self, width, heads, dropout):
super().__init__()
if width % heads:
raise ValueError("Width must divide into heads")
self.heads = heads
self.width = width
self.qkv = nn.Linear(width, 3 * width)
self.projection = nn.Linear(width, width)
self.dropout = nn.Dropout(dropout)
def forward(self, x, mask, causal=True):
batch, time, width = x.shape
head_width = width // self.heads
q, k, v = (
self.qkv(x)
.reshape(batch, time, 3, self.heads, head_width)
.permute(2, 0, 3, 1, 4)
.unbind(0)
)
score = q @ k.transpose(-1, -2) / math.sqrt(head_width)
allowed = mask[:, None, None, :].expand(batch, 1, time, time)
if causal:
allowed = (
allowed
& torch.ones(time, time, dtype=torch.bool, device=x.device).tril()[
None, None
]
)
# A left-padding query has no valid causal key; allow its own key only.
# Real queries never attend padding, and padded outputs are discarded.
empty = ~allowed.any(-1, keepdim=True)
allowed = allowed | (
empty & torch.eye(time, dtype=torch.bool, device=x.device)[None, None]
)
weights = score.masked_fill(~allowed, float("-inf")).softmax(-1)
attended = (
(self.dropout(weights) @ v)
.transpose(1, 2)
.contiguous()
.reshape(batch, time, width)
)
return self.dropout(self.projection(attended))
class DecoderBlock(nn.Module):
def __init__(self, width, heads, mlp_width, dropout):
super().__init__()
self.norm1 = nn.LayerNorm(width)
self.attention = CausalSelfAttention(width, heads, dropout)
self.norm2 = nn.LayerNorm(width)
self.mlp = nn.Sequential(
nn.Linear(width, mlp_width),
nn.GELU(),
nn.Linear(mlp_width, width),
nn.Dropout(dropout),
)
def forward(self, x, mask, causal=True):
x = x + self.attention(self.norm1(x), mask, causal)
return x + self.mlp(self.norm2(x))
class TraceLanguageModel(nn.Module):
def __init__(self, config):
super().__init__()
self.context = config["context"]
self.causal = config.get("causal", True)
width = config["width"]
self.embedding = nn.Embedding(VOCAB, width)
self.position = nn.Embedding(self.context, width)
self.dropout = nn.Dropout(config["dropout"])
self.blocks = nn.ModuleList(
[
DecoderBlock(
width, config["heads"], config["mlp_width"], config["dropout"]
)
for _ in range(config["blocks"])
]
)
self.final_norm = nn.LayerNorm(width)
self.apply(self._initialize)
@staticmethod
def _initialize(module):
if isinstance(module, (nn.Linear, nn.Embedding)):
nn.init.normal_(module.weight, mean=0.0, std=0.02)
if isinstance(module, nn.Linear) and module.bias is not None:
nn.init.zeros_(module.bias)
def forward(self, ids, mask=None):
if mask is None:
mask = ids != PAD
if (
ids.ndim != 2
or ids.shape[1] > self.context
or mask.shape != ids.shape
or not mask.any(1).all()
):
raise ValueError("Invalid context or empty sequence")
positions = (mask.long().cumsum(1) - 1).clamp_min(0)
x = self.dropout(self.embedding(ids) + self.position(positions))
for block in self.blocks:
x = block(x, mask, self.causal)
# Tied projection: one Parameter is used for input and output.
return self.final_norm(x) @ self.embedding.weight.T
def parameter_count(self):
return sum(p.numel() for p in self.parameters())
Pretrain, interrupt and resume
Optimization and complete state
Common error: a resume restarts at the beginning of the epoch and overweights traces; save the cursor or the sampler.
Invariant: finite gradients, parameters actually changed; checkpoint and resume tested without changing splits.
File src/preentrainement_causal/train.py
import base64, copy, hashlib, math, os, time
from pathlib import Path
import numpy as np
import torch
from .config import read_json, save_json, sha, seed_all, rng_state, restore_rng
from .model import TraceLanguageModel
from .tokens import windows, make_causal_batch, token_losses
def digest(state):
h = hashlib.sha256()
for key, value in sorted(state.items()):
h.update(key.encode())
h.update(value.detach().cpu().contiguous().numpy().tobytes())
return h.hexdigest()
def equal(a, b):
if isinstance(a, torch.Tensor):
return isinstance(b, torch.Tensor) and torch.equal(a, b)
if isinstance(a, np.ndarray):
return isinstance(b, np.ndarray) and np.array_equal(a, b)
if isinstance(a, dict):
return (
isinstance(b, dict)
and a.keys() == b.keys()
and all(equal(a[k], b[k]) for k in a)
)
if isinstance(a, (list, tuple)):
return (
type(a) == type(b)
and len(a) == len(b)
and all(equal(x, y) for x, y in zip(a, b))
)
return a == b
def save_training_state(path, state):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
if path.exists():
raise ValueError("Never overwrite a recorded checkpoint")
temporary = path.with_suffix(".pending")
torch.save(state, temporary)
os.replace(temporary, path)
def export_weights(path, model):
weights = {}
for key, tensor in model.state_dict().items():
array = tensor.detach().cpu().contiguous().numpy().astype("<f4", copy=False)
raw = array.tobytes()
weights[key] = {
"shape": list(array.shape),
"dtype": "float32-little-endian",
"bytes": len(raw),
"sha256": hashlib.sha256(raw).hexdigest(),
"base64": base64.b64encode(raw).decode(),
}
save_json(
path,
{
"model_sha256": digest(model.state_dict()),
"parameters": model.parameter_count(),
"weights": weights,
},
)
def load_weights(path, config):
payload = read_json(path)
model = TraceLanguageModel(config)
state = model.state_dict()
if state.keys() != payload["weights"].keys():
raise ValueError("Unexpected weight keys")
for key, entry in payload["weights"].items():
raw = base64.b64decode(entry["base64"], validate=True)
if (
entry["dtype"] != "float32-little-endian"
or len(raw) != entry["bytes"]
or len(raw) != state[key].numel() * 4
or list(state[key].shape) != entry["shape"]
or hashlib.sha256(raw).hexdigest() != entry["sha256"]
):
raise ValueError("Invalid weight metadata")
array = np.frombuffer(raw, dtype="<f4").reshape(entry["shape"]).copy()
if not np.isfinite(array).all():
raise ValueError("Nonfinite weights")
state[key] = torch.from_numpy(array)
model.load_state_dict(state, strict=True)
if digest(model.state_dict()) != payload["model_sha256"]:
raise ValueError("Reload mismatch")
return model
class WindowSampler:
def __init__(self, rows, seed):
self.rows = rows
self.generator = torch.Generator().manual_seed(seed)
self.permutation = []
self.cursor = 0
self.epoch = 0
def take(self, target):
selected = []
active = 0
while active < target:
if self.cursor == len(self.permutation):
self.permutation = torch.randperm(
len(self.rows), generator=self.generator
).tolist()
self.cursor = 0
self.epoch += 1
i = self.permutation[self.cursor]
self.cursor += 1
selected.append(i)
active += len(self.rows[i]["ids"]) - 1
return selected, active
def state_dict(self):
return {
"generator": self.generator.get_state(),
"permutation": self.permutation,
"cursor": self.cursor,
"epoch": self.epoch,
}
def load_state_dict(self, state):
if sorted(state["permutation"]) != list(
range(len(self.rows))
) or not 0 <= state["cursor"] <= len(self.rows):
raise ValueError("Invalid sampler checkpoint")
self.generator.set_state(state["generator"])
self.permutation = list(state["permutation"])
self.cursor = state["cursor"]
self.epoch = state["epoch"]
def rows_from_documents(documents, context):
return [
{"document_id": d["document_id"], **r}
for d in documents
for r in windows(d["text"], context)
]
def lr_at(config, step):
if step < config["warmup_steps"]:
return config["learning_rate"] * (step + 1) / config["warmup_steps"]
progress = min(
1.0,
(step - config["warmup_steps"])
/ max(1, config["optimizer_steps"] - config["warmup_steps"] - 1),
)
return config["learning_rate"] * (1 - progress * (1 - config["minimum_lr_ratio"]))
@torch.no_grad()
def validation_nll(model, rows, batch_size):
model.eval()
total = 0.0
count = 0
for start in range(0, len(rows), batch_size):
ids, mask = make_causal_batch(
[r["ids"] for r in rows[start : start + batch_size]], "cuda"
)
values, active = token_losses(model(ids, mask), ids, mask)
total += float(values.double().sum())
count += int(active.sum())
model.train()
return {"nll_nats_per_token": total / count, "active_tokens": count}
def initialize(config):
seed_all(config["seed"])
model = TraceLanguageModel(config)
destination = Path("assets/checkpoints/initial.pt")
save_training_state(
destination,
{
"config": config,
"model": model.state_dict(),
"model_sha256": digest(model.state_dict()),
},
)
export_weights("outputs/model-initial.json", model)
def resume_training_state(path, identity, model, optimizer, sampler):
# Only this project's locally created checkpoint; never an external artifact.
state = torch.load(path, map_location="cpu", weights_only=False)
if state["identity"] != identity:
raise ValueError("Resume inputs or configuration changed")
model.load_state_dict(state["model"])
optimizer.load_state_dict(state["optimizer"])
sampler.load_state_dict(state["sampler"])
restore_rng(state["rng"])
return state
def train(config, variant="causal", mode="continuous"):
if variant not in ["causal", "unmasked"] or mode not in [
"interrupt",
"continuous",
"resume",
]:
raise ValueError("Unknown run")
if Path("data/generated/test").exists():
raise ValueError("Training service must not mount test traces")
actual = dict(config, causal=variant == "causal")
train_path = Path("data/generated/train/traces.json")
val_path = Path("data/generated/validation/traces.json")
documents = read_json(train_path)
validation = read_json(val_path)
if any(d["split"] != "train" for d in documents) or any(
d["split"] != "validation" for d in validation
):
raise ValueError("Wrong partition")
seed_all(config["seed"])
model = load_weights("outputs/model-initial.json", actual).cuda()
initial = digest(model.state_dict())
optimizer = torch.optim.AdamW(
model.parameters(),
lr=config["learning_rate"],
weight_decay=config["weight_decay"],
)
rows = rows_from_documents(documents, config["context"])
valrows = rows_from_documents(validation, config["context"])
sampler = WindowSampler(rows, config["seed"] + 100)
identity = {
"config": actual,
"train_sha256": sha(train_path),
"validation_sha256": sha(val_path),
"initial_sha256": initial,
"train_windows": len(rows),
"validation_windows": len(valrows),
}
root = Path("assets/checkpoints")
destination = root / f"{variant}-{mode}.pt"
if destination.exists():
raise ValueError("Preserve existing checkpoint")
start = 0
history = []
valhistory = []
if mode == "resume":
state = resume_training_state(
root / f"{variant}-interrupt.pt", identity, model, optimizer, sampler
)
start = state["step"]
history = state["history"]
valhistory = state["validation_history"]
if state["scheduler"] != {"next_step": start, "next_lr": lr_at(config, start)}:
raise ValueError("Scheduler mismatch")
stop = (
config["interrupt_step"] if mode == "interrupt" else config["optimizer_steps"]
)
begun = time.perf_counter()
torch.cuda.reset_peak_memory_stats()
model.train()
for step in range(start, stop):
selected, denominator = sampler.take(config["target_active_tokens_per_update"])
learning_rate = lr_at(config, step)
for group in optimizer.param_groups:
group["lr"] = learning_rate
optimizer.zero_grad(set_to_none=True)
total = 0.0
for offset in range(0, len(selected), config["microbatch_windows"]):
chosen = selected[offset : offset + config["microbatch_windows"]]
ids, mask = make_causal_batch([rows[i]["ids"] for i in chosen], "cuda")
values, _ = token_losses(model(ids, mask), ids, mask)
loss = values.sum() / denominator
if not torch.isfinite(loss):
raise ValueError("Nonfinite loss")
loss.backward()
total += float(values.detach().double().sum())
norm = torch.nn.utils.clip_grad_norm_(
model.parameters(), config["gradient_clip"], error_if_nonfinite=True
)
optimizer.step()
history.append(
{
"step": step + 1,
"active_tokens": denominator,
"windows": selected,
"learning_rate": learning_rate,
"nll_nats_per_token": total / denominator,
"gradient_norm_before_clip": float(norm),
"sampler_epoch": sampler.epoch,
"sampler_cursor": sampler.cursor,
}
)
if (step + 1) % 80 == 0:
valhistory.append(
{
"step": step + 1,
**validation_nll(model, valrows, config["microbatch_windows"]),
}
)
print(
{
"train": {k: v for k, v in history[-1].items() if k != "windows"},
"validation": valhistory[-1],
},
flush=True,
)
torch.cuda.synchronize()
state = {
"identity": identity,
"step": stop,
"model": copy.deepcopy(model.state_dict()),
"optimizer": optimizer.state_dict(),
"sampler": sampler.state_dict(),
"scheduler": {"next_step": stop, "next_lr": lr_at(config, stop)},
"rng": rng_state(),
"history": history,
"validation_history": valhistory,
}
save_training_state(destination, state)
final = digest(model.state_dict())
if initial == final:
raise ValueError("Weights did not change")
if mode != "interrupt":
export_weights(f"outputs/model-{variant}-{mode}.json", model)
save_json(
f"outputs/train-{variant}-{mode}.json",
{
"identity": identity,
"steps": stop,
"history": history,
"validation_history": valhistory,
"parameters": model.parameter_count(),
"initial_sha256": initial,
"model_sha256": final,
"checkpoint_sha256": sha(destination),
"seconds": time.perf_counter() - begun,
"max_torch_allocated_bytes": torch.cuda.max_memory_allocated(),
"scheduler": state["scheduler"],
},
)
def verify_resume(variant="causal"):
root = Path("assets/checkpoints")
a = torch.load(
root / f"{variant}-continuous.pt", map_location="cpu", weights_only=False
)
b = torch.load(
root / f"{variant}-resume.pt", map_location="cpu", weights_only=False
)
checks = {k: equal(a[k], b[k]) for k in a}
if not all(checks.values()):
raise ValueError("Resume differs: " + str(checks))
save_json(
f"outputs/resume-{variant}.json",
{"status": "PASS", "all_state_exact": True, "checks": checks},
)
Generate, then separate syntax from behaviour
Exposure to its own generated tokens
Common error: discarding unparsable outputs inflates validity; include them as failures in the overall score.
Invariant: 200 frozen prefixes, 600 continuations per variant; truncated and invalid outputs counted.
File src/preentrainement_causal/evaluate.py
import math, random, time
from pathlib import Path
import torch
from .config import read_json, save_json, sha
from .tokens import (
PAD,
BOS,
EOS,
VOCAB,
encode_bytes,
windows,
make_causal_batch,
token_losses,
)
from .train import load_weights, rows_from_documents
from .baseline import NGram
from .corpus import parse_trace, normalized
def categorical(probabilities, uniform):
if (
not 0 <= uniform < 1
or len(probabilities) != VOCAB
or any(not math.isfinite(p) or p < 0 for p in probabilities)
):
raise ValueError("Invalid categorical distribution")
total = sum(probabilities)
if total <= 0:
raise ValueError("Empty categorical support")
threshold = uniform * total
accumulator = 0.0
for token, p in enumerate(probabilities):
accumulator += p
if threshold < accumulator:
return token
return max(i for i, p in enumerate(probabilities) if p > 0)
@torch.no_grad()
def next_probabilities(model, sequences, context):
chosen = [s[-context:] for s in sequences]
length = max(map(len, chosen))
ids = torch.full((len(chosen), length), PAD, dtype=torch.long, device="cuda")
mask = torch.zeros_like(ids, dtype=torch.bool)
for i, row in enumerate(chosen):
ids[i, -len(row) :] = torch.tensor(row, device="cuda")
mask[i, -len(row) :] = True
logits = model(ids, mask)[:, -1].float()
logits[:, PAD] = float("-inf")
logits[:, BOS] = float("-inf")
return logits.softmax(-1).cpu().tolist()
def score_generated(prefix, generated, finish, train_exact, train_normalized):
byte_ids = encode_bytes(prefix, False) + [x for x in generated if x < 256]
raw = bytes(byte_ids)
try:
text = raw.decode("utf-8", errors="strict")
utf8 = True
parsed = parse_trace(text)
except UnicodeDecodeError:
text = None
utf8 = False
parsed = {
"syntax_valid": False,
"transitions_valid": False,
"transitions": 0,
"valid_transitions": 0,
"reason": "INVALID_UTF8",
}
return {
"full_bytes_hex": raw.hex(),
"text": text,
"utf8_valid": utf8,
"parser": parsed,
"global_valid": finish == "eos"
and parsed["syntax_valid"]
and parsed["transitions_valid"],
"exact_train_copy": text in train_exact if utf8 else False,
"normalized_train_copy": (
normalized(text) in train_normalized if utf8 else False
),
}
def sample_trace(model, prefixes, config, train_documents, ngram=False):
exact = {d["text"] for d in train_documents}
canonical = {normalized(t) for t in exact}
results = []
for offset in range(0, len(prefixes), config["generation_batch_size"]):
selected = prefixes[offset : offset + config["generation_batch_size"]]
sequences = [[BOS] + encode_bytes(row["prefix"], False) for row in selected]
generators = [random.Random(row["seed"]) for row in selected]
outputs = [[] for row in selected]
draws = [[] for row in selected]
finishes = ["length"] * len(selected)
active = list(range(len(selected)))
for step in range(config["max_new_tokens"]):
if not active:
break
if ngram:
probabilities = [model.probabilities(sequences[i]) for i in active]
for values in probabilities:
values[PAD] = values[BOS] = 0.0
else:
probabilities = next_probabilities(
model, [sequences[i] for i in active], config["context"]
)
remaining = []
for local, index in enumerate(active):
values = probabilities[local]
uniform = generators[index].random()
token = categorical(values, uniform)
outputs[index].append(token)
sequences[index].append(token)
record = {
"uniform": uniform,
"token": token,
"chosen_probability": values[token] / sum(values),
}
if offset == 0 and index == 0 and step < 3:
record["distribution"] = values
draws[index].append(record)
if token == EOS:
finishes[index] = "eos"
else:
remaining.append(index)
active = remaining
for i, row in enumerate(selected):
results.append(
{
**row,
"generated_ids": outputs[i],
"finish_reason": finishes[i],
"sampling_draws": draws[i],
**score_generated(
row["prefix"], outputs[i], finishes[i], exact, canonical
),
}
)
return results
@torch.no_grad()
def evaluate_nll(model, documents, config, ngram=False):
rows = rows_from_documents(documents, config["context"])
records = []
probes = []
for offset in range(0, len(rows), config["microbatch_windows"]):
selected = rows[offset : offset + config["microbatch_windows"]]
if ngram:
losses = [model.nll(row["ids"]) for row in selected]
else:
ids, mask = make_causal_batch([r["ids"] for r in selected], "cuda")
logits = model(ids, mask)
values, active = token_losses(logits, ids, mask)
counts = active.sum(1).tolist()
losses = [v.cpu().tolist() for v in values.split(counts)]
if offset == 0:
for i, row in enumerate(selected[:2]):
for position in [0, len(row["ids"]) - 2]:
probes.append(
{
"document_id": row["document_id"],
"window_start": row["start"],
"position": position,
"target": row["ids"][position + 1],
"logits": logits[i, position].cpu().tolist(),
"nll": losses[i][position],
}
)
for row, values in zip(selected, losses):
records.append(
{
"document_id": row["document_id"],
"start": row["start"],
"targets": row["ids"][1:],
"nll": values,
}
)
count = sum(len(r["nll"]) for r in records)
total = math.fsum(x for r in records for x in r["nll"])
return {
"windows": records,
"probes": probes,
"active_tokens": count,
"nll_nats_per_token": total / count,
"perplexity": math.exp(total / count),
}
def generation_metrics(rows):
n = len(rows)
syntax = [r for r in rows if r["parser"]["syntax_valid"]]
transitions = sum(r["parser"]["transitions"] for r in syntax)
correct = sum(r["parser"]["valid_transitions"] for r in syntax)
return {
"cases": n,
"syntax_valid": len(syntax),
"syntax_valid_rate": len(syntax) / n,
"global_valid": sum(r["global_valid"] for r in rows),
"global_valid_rate": sum(r["global_valid"] for r in rows) / n,
"parsed_transitions": transitions,
"correct_parsed_transitions": correct,
"conditional_transition_accuracy": (
correct / transitions if transitions else None
),
"truncated": sum(r["finish_reason"] == "length" for r in rows),
"invalid_utf8": sum(not r["utf8_valid"] for r in rows),
"exact_train_copies": sum(r["exact_train_copy"] for r in rows),
"normalized_train_copies": sum(r["normalized_train_copy"] for r in rows),
}
def evaluate(config, variant):
paths = {
"initial": "outputs/model-initial.json",
"causal": "outputs/model-causal-resume.json",
"unmasked": "outputs/model-unmasked-continuous.json",
}
ngram = variant == "ngram"
model = (
NGram.from_json(read_json("outputs/ngram.json"))
if ngram
else load_weights(paths[variant], dict(config, causal=variant != "unmasked"))
.cuda()
.eval()
)
test = read_json("data/generated/test/traces.json")
train = read_json("data/generated/train/traces.json")
prefixes = read_json("data/generated/input/prefixes.json")
if (
len(test) != 750
or len(prefixes) != 600
or any(d["split"] != "test" for d in test)
):
raise ValueError("Expected frozen test population")
begun = time.perf_counter()
nll = evaluate_nll(model, test, config, ngram)
generations = sample_trace(model, prefixes, config, train, ngram)
save_json(f"outputs/nll-{variant}.json", nll)
save_json(f"outputs/generations-{variant}.json", generations)
result = {
"variant": variant,
"model_sha256": sha("outputs/ngram.json" if ngram else paths[variant]),
"test_documents": len(test),
"test_sha256": sha("data/generated/test/traces.json"),
"prefixes_sha256": sha("data/generated/input/prefixes.json"),
"active_tokens": nll["active_tokens"],
"nll_nats_per_token": nll["nll_nats_per_token"],
"perplexity": nll["perplexity"],
"generation": generation_metrics(generations),
"seconds": time.perf_counter() - begun,
}
save_json(f"outputs/evaluation-{variant}.json", result)
print(result, flush=True)
Detect a false success through a future leak
Controlled ablation
Common error: do not deliver the faulty version by default; the exercise and its results have a distinct name.
Invariant: the unmasked version fails the planned test; the restored final version passes it.
File src/preentrainement_causal/exercise.py
from .config import read_json, save_json, sha
from .train import load_weights
from .mechanism import compare_prefix_logits
def exercise(config):
causal = load_weights(
"outputs/model-causal-resume.json", dict(config, causal=True)
).cuda()
correct = compare_prefix_logits(causal)
del causal
faulty = load_weights(
"outputs/model-unmasked-continuous.json", dict(config, causal=False)
).cuda()
incorrect = compare_prefix_logits(faulty)
faulty.causal = True
restored = compare_prefix_logits(faulty)
if not correct["passes"] or incorrect["passes"] or not restored["passes"]:
raise ValueError(
"Ablation must expose future leakage and restored mask must pass"
)
before = read_json("outputs/exercise-before.json")
after = {path: sha(path) for path in before}
if before != after:
raise ValueError("Exercise changed main artifacts")
main = read_json("outputs/evaluation-causal.json")
ablation = read_json("outputs/evaluation-unmasked.json")
save_json(
"outputs/exercise/causality.json",
{
"causal": correct,
"unmasked": incorrect,
"mask_restored_same_ablation_weights": restored,
"main_artifacts_unchanged": True,
"protected_sha256": after,
"causal_evaluation": main,
"unmasked_evaluation": ablation,
"quality_gain_required": False,
},
)
print({"correct": correct, "faulty": incorrect, "restored": restored})
def preserve_main():
paths = [
"outputs/model-causal-continuous.json",
"outputs/model-causal-resume.json",
"outputs/resume-causal.json",
"outputs/evaluation-causal.json",
"outputs/generations-causal.json",
"outputs/nll-causal.json",
]
save_json("outputs/exercise-before.json", {path: sha(path) for path in paths})
Consulted sources
Causal modelling, Hugging Face presents the next-token objective; this TP codes its own PyTorch decoder and imports no Transformers model. Docker Compose GPU reservation and NVIDIA Container Toolkit installation document the host prerequisites. These pages were opened during preparation; the versions actually executed come from the locks and the environment report.
Read the results that were actually obtained
The 750 test logs represent 301,127 target tokens. The n-gram obtains an NLL of 0.2344045 nat/token; the initial model 5.5835855; the trained causal model 0.0878211, or perplexity 1.09179. This low teacher-forced perplexity does not predict free generation by itself.
On the same 200 prefixes, with three draws each, the n-gram produces 4 globally valid traces out of 600, the initial model 0 and the causal model 12 (2%). The causal model reaches the limit of 128 new tokens in 577 cases; 59 outputs contain invalid UTF-8. These figures describe this run and this generator, not a general log-production capacity.
Here the unmasked ablation gives more globally valid traces: 235/600, with one truncation, 249 syntactically valid traces and 52 invalid UTF-8 outputs. It generates 19.818 tokens on average, versus 126.550 for the causal model, including EOS. This apparent success must be read with the following causality test.
Its teacher-forced score of 0.0817819 nat/token is calculated with access to future tokens. Its exponential is therefore not an autoregressive perplexity comparable with the other three values: this score does not correspond to the conditionals used at generation time.
The suffix test decides the causality property: changing future tokens changes the ablation's earlier logits by up to 0.906817, above 10⁻⁶, versus zero with the causal mask. Restoring the mask on the same ablation weights restores the expected result. The exercise shows why a lower loss alone does not prove correct causal generation.
The experiment concerns a fictional automaton, one seed, shared patterns and windows of 256 tokens. The 600 draws correspond to 200 prefixes, not 600 independent logs. A different generation limit, another corpus or another criterion could change these results.
Development observations
Values recalculated from this run's raw outputs. They are not performance thresholds; the path from the ZIP recalculates them on the same populations.
{
"status": "PASS",
"training": true,
"documents": 5000,
"test_documents": 750,
"generation_cases_per_variant": 600,
"parameters": 4870912,
"metrics": {
"ngram": {
"nll_nats_per_token": 0.23440450702708443,
"perplexity": 1.2641557486810593,
"generation": {
"cases": 600,
"syntax_valid": 60,
"syntax_valid_rate": 0.1,
"global_valid": 4,
"global_valid_rate": 0.006666666666666667,
"parsed_transitions": 271,
"correct_parsed_transitions": 143,
"conditional_transition_accuracy": 0.5276752767527675,
"truncated": 426,
"invalid_utf8": 48,
"exact_train_copies": 0,
"normalized_train_copies": 0
}
},
"initial": {
"nll_nats_per_token": 5.583585544218517,
"perplexity": 266.0237376852991,
"generation": {
"cases": 600,
"syntax_valid": 0,
"syntax_valid_rate": 0.0,
"global_valid": 0,
"global_valid_rate": 0.0,
"parsed_transitions": 0,
"correct_parsed_transitions": 0,
"conditional_transition_accuracy": null,
"truncated": 399,
"invalid_utf8": 595,
"exact_train_copies": 0,
"normalized_train_copies": 0
}
},
"causal": {
"nll_nats_per_token": 0.08782112727113893,
"perplexity": 1.0917928126012122,
"generation": {
"cases": 600,
"syntax_valid": 12,
"syntax_valid_rate": 0.02,
"global_valid": 12,
"global_valid_rate": 0.02,
"parsed_transitions": 77,
"correct_parsed_transitions": 77,
"conditional_transition_accuracy": 1.0,
"truncated": 577,
"invalid_utf8": 59,
"exact_train_copies": 0,
"normalized_train_copies": 9
}
},
"unmasked": {
"nll_nats_per_token": 0.08178190243999106,
"perplexity": 1.0852191003797427,
"generation": {
"cases": 600,
"syntax_valid": 249,
"syntax_valid_rate": 0.415,
"global_valid": 235,
"global_valid_rate": 0.39166666666666666,
"parsed_transitions": 763,
"correct_parsed_transitions": 748,
"conditional_transition_accuracy": 0.9803407601572739,
"truncated": 1,
"invalid_utf8": 52,
"exact_train_copies": 0,
"normalized_train_copies": 0
}
}
},
"quality_gain_required": false,
"scope": "Data-only counts, hashes, active-token budgets, resume, NLL probes, ngram, sampling streams, complete parse and metric recomputation; no claim of scalar reimplementation of every Transformer layer"
}
