Un atelier fictif veut produire des suites de traces pour tester son analyseur de journaux, sans utiliser de documents clients. Les commandes dans les traces sont des symboles, jamais des actions système. Nous construisons un générateur statistique dont les erreurs restent visibles, pas un contrôleur de machine.
Un préfixe de journal donne plusieurs continuations, comparées à un modèle n-gramme. Un parseur distingue trace lisible, transition d'état autorisée et copie d'une trace train.
Télécharger le projet complet, révision r001
Corpus et partitions
AtelierTrace contient 500 familles de dix traces : 350 familles train (3500 traces), 75 validation et 75 test (750 chacune). Une famille fixe un alias, une plage de longueur et une préférence de transition. Les routes comportent 6 à 24 événements. Le générateur vérifie les doublons exacts et les routes normalisées sans alias ; il régénère la route entière en cas de collision, puis fige les comptes et hashes. Une simple différence d’identifiant n’est donc pas considérée comme une nouvelle trace. Les états, commandes, courts motifs et combinaisons de paramètres restent partagés : cette partition ne prouve pas une généralisation à un langage inédit.
L’automate possède trois états. Depuis off, start mène à ready ; depuis ready, load mène à busy et stop à off ; depuis busy, unload mène à ready et stop à off. Les événements ne lancent aucune action système. Le parseur relit les lignes et vérifie indépendamment syntaxe et transitions ; il n’utilise pas la liste d’événements enregistrée par le générateur comme résultat prédit.
Tokens, fenêtres et perte
Un token est un octet UTF-8, entier de 0 à 255. PAD=256, BOS=257, EOS=258 donnent un vocabulaire de 259 valeurs. Aucun tokenizer n’est ajusté sur les données. Un caractère accentué peut occuper plusieurs tokens ; la NLL est en nats par token octet/contrôle, pas par caractère. Les tokens de contrôle ne deviennent pas du texte. Une suite d’octets invalide reste visible et échoue au parseur.
On ajoute BOS et EOS à chaque trace après la partition. Les fenêtres de 256 tokens se recouvrent d’un token, avec un pas de 255 : chaque cible suivante apparaît exactement une fois, y compris EOS. Une fenêtre ne traverse jamais deux documents. Le contexte repart au début de chaque fenêtre ; cet effet est commun à la NLL neuronale et au n-gramme. Aucune prétention à un contexte complet au-delà de 256 tokens.
À la position t, les logits prédisent le token t+1. La loss sélectionne les cibles actives avant la CE, somme leurs pertes et divise par leur nombre exact. Le padding n’ajoute ni cible ni dénominateur. Pendant le teacher forcing, les entrées précédentes sont les vrais tokens. En génération libre, les prochains préfixes incluent les erreurs du modèle.
Attention et poids appris
Le décodeur original contient six blocs pré-LayerNorm, largeur 256, huit têtes et MLP 1024 avec GELU. Les embeddings de position sont appris jusqu’à 256. Les projections Q/K/V forment des tensors [batch, têtes, temps, largeur_par_tête] ; softmax(QKᵀ/√d) pondère V. Le masque causal interdit les clés futures et le masque de padding interdit les clés fictives. Les requêtes entièrement paddées reçoivent une clé de secours sur leur propre position, puis leurs sorties sont ignorées ; cela évite un softmax entièrement −inf en génération avec padding gauche.
Résidus, normalisations et dropout 0,1 sont visibles. Le dropout est désactivé en évaluation. La projection de sortie réutilise exactement le Parameter de l’embedding d’entrée ; le nombre de paramètres uniques est calculé par le programme. Tous les poids sont FP32 CUDA, avec TF32 désactivé et algorithmes déterministes. Ce profil n’est pas un modèle généraliste ni une reproduction d’un grand GPT.
Comparaisons et apprentissage
Le n-gramme d’ordre trois apprend les comptes du prochain token étant donnés les deux précédents, sur les mêmes fenêtres train. Le contexte manquant est complété par BOS. Le lissage additif alpha=0,1 porte sur les 259 valeurs ; un contexte absent donne une distribution uniforme. Sa NLL utilise exactement les mêmes cibles test que le décodeur aléatoire et le décodeur appris.
AdamW utilise lr=0,0003, weight_decay=0,01, clipping de norme à 1, 800 mises à jour fixes, 80 pas de warm-up puis décroissance linéaire jusqu’à 10 % du taux initial. Un sampler parcourt des permutations de fenêtres et conserve son curseur. Chaque mise à jour accumule des fenêtres entières jusqu’à atteindre au moins 4096 cibles, par microbatches de huit fenêtres ; le nombre réel peut dépasser 4096 et constitue le dénominateur exact. Ce n’est pas une moyenne des moyennes des microbatches.
La validation est calculée tous les 80 pas, sans sélection de checkpoint ni adaptation d’hyperparamètres. Le checkpoint final est fixé à 800. L’interruption à 320 et la reprise doivent retrouver exactement modèle, AdamW, scheduler, historique, permutation, curseur et états aléatoires Python/NumPy/PyTorch CPU/CUDA. Les états .pt sont créés localement et ne doivent pas être remplacés par des fichiers non fiables. Des poids FP32 encodés en JSON avec dimensions, tailles et SHA permettent un audit sans import pickle externe.
Mesurer les sorties libres
Les 750 traces test donnent NLL et perplexité exp(NLL) dans la même unité. On fige les 200 premiers préfixes test par identifiant, avec trois graines de continuation chacun : 600 sorties par variante. Chaque préfixe contient deux transitions vraies puis CMD . La génération utilise au plus 128 nouveaux tokens, température 1, sans top-k ni top-p. PAD et BOS sont supprimés du support pour les deux modèles ; EOS arrête la génération. Le décodeur consulte les 256 derniers tokens avec positions recalées à zéro. Les tirages catégoriels utilisent un générateur Python distinct par cas et des probabilités effectivement calculées, pour conserver les mêmes graines malgré les sorties de longueur différente.
Le rapport distingue syntaxe valide, transitions correctes parmi les sorties parsables, validité globale et troncatures. La validité globale exige EOS, syntaxe complète et toutes les transitions autorisées. Une sortie qui atteint le plafond reste un échec global même si son dernier octet ferme une ligne valide. Les erreurs UTF-8 et les sorties non parsables restent au dénominateur de 600. Les copies exactes et les copies après suppression de l’alias sont comptées séparément. Un préfixe test partagé partiellement avec train n’est pas une copie complète.
Exercice : regarder le futur
La variante unmasked repart des mêmes poids initiaux, avec le même sampler et les mêmes 800 pas, mais autorise l’attention aux positions futures. La cible suivante peut ainsi être visible pendant le teacher forcing. On compare sa NLL et ses 600 générations, sans imposer de gain ou de dégradation. Le test change un suffixe et mesure les logits antérieurs : la version fautive doit révéler une influence, la version causale doit la bloquer. Restaurer le masque sur les mêmes poids fautifs doit restaurer cette propriété mécanique ; cela n’équivaut pas à réentraîner correctement ces poids. Les résultats initiaux sont protégés par SHA.
Interpréter
Une meilleure NLL ne garantit ni syntaxe ni comportement correct en génération libre. Une validité élevée sur cet automate simple ne démontre aucune aptitude au français général, au pilotage d’un atelier ou à la sécurité physique. Le simulateur déterministe est une alternative connue qui garantit ses propres règles ; le réseau étudie une distribution, pas une commande de machine. Une seule graine d’apprentissage est exécutée ; les trois graines de génération ne sont pas trois entraînements indépendants.
Construire un corpus où les fuites sont détectables
Partition par famille avant fenêtrage
Erreur fréquente : Des alias différents masquent une même trace : comparer aussi une forme normalisée sans alias.
Invariant : Aucune famille ni trace identique ne traverse les splits ; nombres effectifs écrits dans outputs/data-audit.json.
Fichier 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
Former tokens, cibles et masque causal
Décalage causal et padding
Erreur fréquente : Un double décalage des labels apprend le token t+2 : tester sur une suite de cinq tokens connue.
Invariant : Le décodage reconstitue le texte ; modifier les PAD ne change pas la loss.
Fichier 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()
Écrire le décodeur et sa baseline
Q/K/V, résidus et vraisemblance
Erreur fréquente : Inverser le sens du masque autorise le futur : inspecter une matrice 4×4, puis les logits.
Invariant : Test de causalité réussi à la tolérance fixée ; nombre de paramètres et NLL initiale capturés.
Fichier 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())
Préentraîner, interrompre et reprendre
Optimisation et état complet
Erreur fréquente : Une reprise repart au début de l'epoch et surpondère des traces : sauvegarder le curseur ou le sampler.
Invariant : Gradients finis, paramètres effectivement modifiés ; checkpoint et reprise testés sans changer les splits.
Fichier 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},
)
Générer puis séparer syntaxe et comportement
Exposition aux propres tokens générés
Erreur fréquente : Écarter les sorties non parsables gonfle la validité : les inclure comme échecs dans le score global.
Invariant : 200 préfixes gelés, 600 continuations par variante ; tronquées et invalides comptabilisées.
Fichier 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)
Détecter une fausse réussite par fuite du futur
Ablation contrôlée
Erreur fréquente : Ne pas livrer la version fautive par défaut : l'exercice et ses résultats portent un nom distinct.
Invariant : La version sans masque échoue au test prévu ; la version finale restaurée le passe.
Fichier 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})
Sources consultées
Modélisation causale, Hugging Face présente l’objectif du prochain token ; le présent TP code son propre décodeur PyTorch et n’importe aucun modèle Transformers. Réservation GPU Docker Compose et installation NVIDIA Container Toolkit documentent les prérequis hôte. Ces pages ont été ouvertes lors de la préparation ; les versions réellement exécutées viennent des locks et du rapport d’environnement.
Lire les résultats effectivement obtenus
Les 750 journaux test représentent 301 127 tokens cibles. Le n-gramme obtient une NLL de 0,2344045 nat/token ; le modèle initial 5,5835855 ; le modèle causal entraîné 0,0878211, soit une perplexité de 1,09179. Cette faible perplexité teacher-forced ne garantit pas des journaux complets en génération libre.
Sur les mêmes 200 préfixes, avec trois tirages chacun, le n-gramme produit 4 traces globalement valides sur 600, le modèle initial 0 et le causal 12 (2 %). Le causal atteint la limite de 128 nouveaux tokens dans 577 cas ; 59 sorties contiennent un UTF-8 invalide. Ces catégories peuvent se recouvrir. Son exactitude de transition de 77/77 concerne seulement les transitions des 12 traces syntaxiquement valides : elle ne signifie pas 100 % de réussite sur les 600 cas. Aucune copie train exacte n’est observée, mais 9 continuations causales reproduisent une trace train après suppression de l’identifiant d’atelier.
L’ablation sans masque donne ici davantage de traces globalement valides : 235/600, avec une seule coupure, 249 traces syntaxiquement valides et 52 UTF-8 invalides. Elle génère en moyenne 19.818 tokens, contre 126.550 pour le causal, EOS compris. Le budget et la distribution des longueurs comptent dans cette comparaison. Le code ne fixe aucun quota de réussite et ne force pas l’ablation à être moins bonne.
Son score teacher-forced de 0,0817819 nat/token est calculé avec accès aux tokens futurs. Son exponentielle n’est donc pas une perplexité autorégressive comparable aux trois autres valeurs : ce score ne correspond pas aux conditionnelles utilisées lorsque l’on génère sans futur. La génération de l’ablation recalcule chaque préfixe disponible ; aucune cible future test ne lui est fournie.
Le test de suffixe tranche la propriété de causalité : modifier les tokens futurs change les logits passés de l’ablation jusqu’à 0,906817, au-delà de 10⁻⁶, contre zéro avec masque causal. Restaurer le masque sur les mêmes poids d’ablation rétablit l’invariance ; ce contrôle ne constitue pas une nouvelle mesure de qualité générative. Le résultat observé n’autorise pas à interpréter sa loss avec fuite comme une vraisemblance causale correcte.
L’expérience porte sur un automate fictif, une graine, des motifs partagés et des fenêtres de 256 tokens. Les 600 tirages correspondent à 200 préfixes, pas à 600 journaux indépendants. Une limite de génération différente, un autre corpus ou un critère de longueur utile nécessiteraient un protocole distinct. Le parcours conserve les mauvais résultats et le seuil de causalité initial.
Observations de développement
Valeurs recalculées depuis les sorties brutes de ce parcours. Elles ne sont pas des seuils de gain ; le parcours depuis le ZIP les recalcule sur les mêmes 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"
}
