Un’officina fittizia vuole produrre sequenze di tracce per testare il proprio analizzatore di log, senza usare documenti dei clienti. I comandi nelle tracce sono simboli, mai azioni di sistema. Costruiamo un generatore statistico i cui errori restano visibili, non un controllore di macchine.
Un prefisso di log fornisce più continuazioni, confrontate con un modello n-gramma. Un parser distingue una traccia leggibile, una transizione di stato autorizzata e una copia di una traccia train.
Scaricare il progetto completo, revisione r001
Corpus e suddivisioni
AtelierTrace contiene 500 famiglie di dieci tracce: 350 famiglie train (3500 tracce), 75 validation e 75 test (750 ciascuna). Una famiglia fissa un alias, un intervallo di lunghezza e una preferenza di transizione. Le rotte comportano da 6 a 24 eventi. Il generatore verifica i duplicati esatti e le rotte normalizzate senza alias; rigenera l’intera rotta in caso di collisione, poi fissa conteggi e hash. Una semplice differenza di identificatore non è quindi considerata una nuova traccia. Stati, comandi, brevi motivi e combinazioni di parametri restano condivisi: questa suddivisione non dimostra una generalizzazione a un linguaggio inedito.
L’automa possiede tre stati. Da off, start porta a ready; da ready, load porta a busy e stop a off; da busy, unload porta a ready e stop a off. Gli eventi non avviano alcuna azione di sistema. Il parser rilegge le righe e verifica indipendentemente sintassi e transizioni; non usa come risultato predetto l’elenco degli eventi registrato dal generatore.
Token, finestre e perdita
Un token è un byte UTF-8, intero da 0 a 255. PAD=256, BOS=257, EOS=258 danno un vocabolario di 259 valori. Nessun tokenizer è adattato ai dati. Un carattere accentato può occupare più token; la NLL è in nat per token byte/controllo, non per carattere. I token di controllo non diventano testo. Una sequenza di byte non valida resta visibile e fallisce il parser.
BOS ed EOS vengono aggiunti a ogni traccia dopo la suddivisione. Le finestre di 256 token si sovrappongono di un token, con passo 255: ogni destinazione successiva appare esattamente una volta, incluso EOS. Una finestra non attraversa mai due documenti. Il contesto riparte all’inizio di ogni finestra; questo effetto è comune alla NLL neurale e all’n-gramma. Non si pretende un contesto completo oltre 256 token.
Alla posizione t, i logit predicono il token t+1. La loss seleziona i target attivi prima della CE, somma le loro perdite e divide per il loro numero esatto. Il padding non aggiunge né target né denominatore. Durante il teacher forcing, gli input precedenti sono i token reali. Nella generazione libera, i prefissi successivi includono gli errori del modello.
Attenzione e pesi appresi
Il decoder originale contiene sei blocchi pre-LayerNorm, larghezza 256, otto teste e MLP 1024 con GELU. Gli embedding di posizione sono appresi fino a 256. Le proiezioni Q/K/V formano tensori [batch, teste, tempo, larghezza_per_testa]; softmax(QKᵀ/√d) pondera V. La maschera causale vieta le chiavi future e quella di padding vieta le chiavi fittizie. Le query interamente riempite ricevono una chiave di soccorso sulla propria posizione, poi le loro uscite vengono ignorate; ciò evita una softmax interamente −inf nella generazione con padding a sinistra.
Residui, normalizzazioni e dropout 0,1 sono visibili. Il dropout è disattivato in valutazione. La proiezione di uscita riutilizza esattamente il Parameter dell’embedding di ingresso; il programma calcola il numero di parametri unici. Tutti i pesi sono FP32 CUDA, con TF32 disattivato e algoritmi deterministici. Questo profilo non è un modello generalista né una riproduzione di un grande GPT.
Confronti e apprendimento
L’n-gramma di ordine tre apprende i conteggi del token successivo dati i due precedenti, sulle stesse finestre train. Il contesto mancante è completato da BOS. Lo smoothing additivo alpha=0,1 riguarda i 259 valori; un contesto assente produce una distribuzione uniforme. La sua NLL usa esattamente gli stessi target test del decoder casuale e del decoder appreso.
AdamW usa lr=0,0003, weight_decay=0,01, clipping della norma a 1, 800 aggiornamenti fissi, 80 passi di warm-up poi decadimento lineare fino al 10% del tasso iniziale. Un sampler percorre permutazioni di finestre e conserva il proprio cursore. Ogni aggiornamento accumula finestre intere fino a raggiungere almeno 4096 target, tramite microbatch di otto finestre; il numero reale può superare 4096 e costituisce il denominatore esatto. Non è una media delle medie dei microbatch.
La validation viene calcolata ogni 80 passi, senza selezione di checkpoint né adattamento di iperparametri. Il checkpoint finale è fissato a 800. L’interruzione a 320 e la ripresa devono ritrovare esattamente modello, AdamW, scheduler, storico, permutazione, cursore e stati casuali Python/NumPy/PyTorch CPU/CUDA. Gli stati .pt sono creati localmente e non devono essere sostituiti da file non affidabili. Pesi FP32 codificati in JSON con dimensioni, taglie e SHA consentono un audit senza importare pickle esterni.
Misurare le uscite libere
Le 750 tracce test forniscono NLL e perplessità exp(NLL) nella stessa unità. Si fissano per identificatore i primi 200 prefissi test, con tre semi di continuazione ciascuno: 600 uscite per variante. Ogni prefisso contiene due transizioni reali poi CMD . La generazione usa al massimo 128 nuovi token, temperatura 1, senza top-k né top-p. PAD e BOS sono rimossi dal supporto per entrambi i modelli; EOS arresta la generazione. Il decoder consulta gli ultimi 256 token con posizioni riallineate a zero. I sorteggi categoriali usano un generatore Python distinto per caso e probabilità effettivamente calcolate, per conservare gli stessi semi nonostante uscite di lunghezza diversa.
Il rapporto distingue sintassi valida, transizioni corrette tra le uscite parsabili, validità globale e troncamenti. La validità globale richiede EOS, sintassi completa e tutte le transizioni autorizzate. Un’uscita che raggiunge il limite resta un fallimento globale anche se il suo ultimo byte chiude una riga valida. Errori UTF-8 e uscite non parsabili restano nel denominatore di 600. Le copie esatte e le copie dopo la rimozione dell’alias sono conteggiate separatamente. Un prefisso test condiviso solo in parte con train non è una copia completa.
Esercizio: guardare il futuro
La variante unmasked riparte dagli stessi pesi iniziali, con lo stesso sampler e gli stessi 800 passi, ma autorizza l’attenzione alle posizioni future. La destinazione successiva può così essere visibile durante il teacher forcing. Si confrontano la sua NLL e le sue 600 generazioni, senza imporre un guadagno o un peggioramento. Il test cambia un suffisso e misura i logit anteriori: la versione difettosa deve rivelare un’influenza, quella causale deve bloccarla. Ripristinare la maschera sugli stessi pesi difettosi deve ripristinare questa proprietà meccanica; ciò non equivale a riaddestrare correttamente tali pesi. I risultati iniziali sono protetti da SHA.
Interpretare
Una NLL migliore non garantisce né sintassi né comportamento corretto nella generazione libera. Una validità elevata su questo semplice automa non dimostra alcuna capacità per il francese generale, la guida di un’officina o la sicurezza fisica. Il simulatore deterministico è un’alternativa nota che garantisce le proprie regole; la rete studia una distribuzione, non un comando di macchina. Viene eseguito un solo seme di apprendimento; i tre semi di generazione non sono tre addestramenti indipendenti.
Costruire un corpus in cui le fughe siano rilevabili
Partizione per famiglia prima del fenêtrage
Errore frequente: alias diversi mascherano una stessa traccia; confrontare anche una forma normalizzata senza alias.
Invariante: nessuna famiglia né traccia identica attraversa le suddivisioni; i conteggi effettivi sono scritti in 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
Formare token, destinazioni e maschera causale
Scorrimento causale e padding
Errore frequente: un doppio scorrimento delle label apprende il token t+2; testare su una sequenza nota di cinque token.
Invariante: la decodifica ricostruisce il testo; modificare i PAD non cambia la 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()
Scrivere il decoder e la sua baseline
Q/K/V, residui e verosimiglianza
Errore frequente: invertire il verso della maschera autorizza il futuro; ispezionare una matrice 4×4, poi i logit.
Invariante: test di causalità superato alla tolleranza fissata; numero di parametri e NLL iniziale acquisiti.
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())
Preaddestrare, interrompere e riprendere
Ottimizzazione e stato completo
Errore frequente: una ripresa ricomincia dall’inizio dell’epoca e sovrappesa le tracce; salvare il cursore o il sampler.
Invariante: gradienti finiti, parametri effettivamente modificati; checkpoint e ripresa testati senza cambiare le suddivisioni.
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},
)
Generare, poi separare sintassi e comportamento
Esposizione ai propri token generati
Errore frequente: escludere le uscite non parsabili gonfia la validità; includerle come fallimenti nel punteggio globale.
Invariante: 200 prefissi fissati, 600 continuazioni per variante; troncate e non valide conteggiate.
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)
Rilevare un falso successo causato da una fuga dal futuro
Ablazione controllata
Errore frequente: non distribuire per impostazione predefinita la versione difettosa; l’esercizio e i suoi risultati portano un nome distinto.
Invariante: la versione senza maschera fallisce il test previsto; la versione finale ripristinata lo supera.
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})
Fonti consultate
Modellazione causale, Hugging Face presenta l’obiettivo del token successivo; questo laboratorio programma il proprio decoder PyTorch e non importa alcun modello Transformers. Prenotazione GPU Docker Compose e installazione NVIDIA Container Toolkit documentano i prerequisiti host. Le versioni realmente eseguite provengono dai lock e dal rapporto dell’ambiente.
Leggere i risultati effettivamente ottenuti
I 750 log test rappresentano 301 127 token destinazione. L’n-gramma ottiene una NLL di 0,2344045 nat/token; il modello iniziale 5,5835855; il modello causale addestrato 0,0878211, cioè una perplessità di 1,09179. Questa bassa perplessità teacher-forced non garantisce log completi in generazione libera.
Sugli stessi 200 prefissi, con tre estrazioni ciascuno, l’n-gramma produce 4 tracce globalmente valide su 600, il modello iniziale 0 e il causale 12 (2%). Il causale raggiunge il limite di 128 nuovi token in 577 casi; 59 uscite contengono UTF-8 non valido. Le categorie possono sovrapporsi. Nessuna copia train esatta è osservata.
L’ablazione senza maschera produce qui più tracce globalmente valide: 235/600, 249 tracce sintatticamente valide e 52 UTF-8 non validi. Genera in media 19,818 token, contro 126,550 per il causale, EOS incluso. Il codice non impone alcuna quota di successo.
Il suo punteggio teacher-forced di 0,0817819 nat/token è calcolato con accesso ai token futuri. La sua esponenziale non è quindi una perplessità autoregressiva comparabile: non corrisponde alle condizionali usate nella generazione senza futuro.
Il test di suffisso decide la proprietà di causalità: modificare i token futuri cambia i logit passati dell’ablazione fino a 0,906817, oltre 10⁻⁶, contro zero con la maschera causale. Ripristinare la maschera sugli stessi pesi ripristina l’invarianza; non è una nuova misura di qualità generativa.
L’esperimento riguarda un automa fittizio, un seme, motivi condivisi e finestre di 256 token. Le 600 estrazioni corrispondono a 200 prefissi, non a 600 log indipendenti. Un altro corpus o criterio di lunghezza utile richiederebbe un protocollo distinto.
Osservazioni sullo sviluppo
Valori ricalcolati dalle uscite grezze di questo percorso. Non sono soglie di guadagno; il percorso dallo ZIP li ricalcola sulle stesse popolazioni.
{
"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"
}
