Un taller ficticio quiere producir secuencias de trazas para probar su analizador de registros, sin utilizar documentos de clientes. Los comandos de las trazas son símbolos, nunca acciones del sistema. Construimos un generador estadístico cuyos errores siguen siendo visibles, no un controlador de máquinas.
Un prefijo de registro da varias continuaciones, comparadas con un modelo n-grama. Un analizador distingue una traza legible, una transición de estado autorizada y una copia de una traza de train.
Descargar el proyecto completo, revisión r001
Corpus y particiones
AtelierTrace contiene 500 familias de diez trazas: 350 familias de train (3500 trazas), 75 de validation y 75 de test (750 cada una). Una familia fija un alias, un intervalo de longitud y una preferencia de transición. Las rutas tienen de 6 a 24 eventos. El generador comprueba los duplicados exactos y las rutas normalizadas sin alias; regenera la ruta completa en caso de colisión y después fija los recuentos y hashes. Por tanto, una simple diferencia de identificador no se considera una nueva traza. Los estados, comandos, motivos cortos y combinaciones de parámetros siguen compartidos: esta partición no prueba una generalización a un lenguaje inédito.
El autómata tiene tres estados. Desde off, start conduce a ready; desde ready, load conduce a busy y stop a off; desde busy, unload conduce a ready y stop a off. Los eventos no ejecutan ninguna acción del sistema. El analizador vuelve a leer las líneas y comprueba de forma independiente la sintaxis y las transiciones; no utiliza como resultado predicho la lista de eventos guardada por el generador.
Tokens, ventanas y pérdida
Un token es un byte UTF-8, un entero de 0 a 255. PAD=256, BOS=257 y EOS=258 dan un vocabulario de 259 valores. No se ajusta ningún tokenizer a los datos. Un carácter acentuado puede ocupar varios tokens; la NLL se expresa en nats por token byte/control, no por carácter. Los tokens de control no se convierten en texto. Una secuencia de bytes no válida permanece visible y falla en el analizador.
Se añaden BOS y EOS a cada traza después de la partición. Las ventanas de 256 tokens se solapan en un token, con un paso de 255: cada objetivo siguiente aparece exactamente una vez, incluido EOS. Una ventana nunca atraviesa dos documentos. El contexto se reinicia al comienzo de cada ventana; este efecto es común a la NLL neuronal y al n-grama. No se pretende disponer de un contexto completo más allá de 256 tokens.
En la posición t, los logits predicen el token t+1. La loss selecciona los objetivos activos antes de la CE, suma sus pérdidas y divide por su número exacto. El padding no añade ni objetivo ni denominador. Durante el teacher forcing, las entradas anteriores son los tokens reales. En generación libre, los prefijos siguientes incluyen los errores del modelo.
Atención y pesos aprendidos
El decodificador original contiene seis bloques pre-LayerNorm, anchura 256, ocho cabezas y MLP 1024 con GELU. Los embeddings de posición se aprenden hasta 256. Las proyecciones Q/K/V forman tensors [batch, cabezas, tiempo, anchura_por_cabeza]; softmax(QKᵀ/√d) pondera V. La máscara causal prohíbe las claves futuras y la máscara de padding prohíbe las claves ficticias. Las consultas completamente paddadas reciben una clave de respaldo en su propia posición y después se ignoran sus salidas; esto evita un softmax completamente −inf en generación con padding a la izquierda.
Los residuos, las normalizaciones y el dropout 0,1 son visibles. El dropout se desactiva en evaluación. La proyección de salida reutiliza exactamente el Parameter del embedding de entrada; el programa calcula el número de parámetros únicos. Todos los pesos son FP32 CUDA, con TF32 desactivado y algoritmos deterministas. Este perfil no es un modelo generalista ni una reproducción de un GPT grande.
Comparaciones y aprendizaje
El n-grama de orden tres aprende los recuentos del token siguiente dados los dos anteriores, sobre las mismas ventanas de train. El contexto ausente se completa con BOS. El suavizado aditivo alpha=0,1 se aplica a los 259 valores; un contexto ausente da una distribución uniforme. Su NLL utiliza exactamente los mismos objetivos de test que el decodificador aleatorio y el decodificador aprendido.
AdamW utiliza lr=0,0003, weight_decay=0,01, clipping de norma a 1, 800 actualizaciones fijas, 80 pasos de warm-up y después un descenso lineal hasta el 10 % de la tasa inicial. Un sampler recorre permutaciones de ventanas y conserva su cursor. Cada actualización acumula ventanas completas hasta alcanzar al menos 4096 objetivos, mediante microbatches de ocho ventanas; el número real puede superar 4096 y constituye el denominador exacto. No es una media de las medias de los microbatches.
La validación se calcula cada 80 pasos, sin selección de checkpoint ni adaptación de hiperparámetros. El checkpoint final se fija en 800. La interrupción en 320 y la reanudación deben recuperar exactamente modelo, AdamW, scheduler, historial, permutación, cursor y estados aleatorios de Python/NumPy/PyTorch CPU/CUDA. Los estados .pt se crean localmente y no deben sustituirse por archivos no fiables. Los pesos FP32 codificados en JSON con dimensiones, tamaños y SHA permiten una auditoría sin importar un pickle externo.
Medir las salidas libres
Las 750 trazas de test dan NLL y perplejidad exp(NLL) en la misma unidad. Se fijan los primeros 200 prefijos de test por identificador, con tres semillas de continuación para cada uno: 600 salidas por variante. Cada prefijo contiene dos transiciones verdaderas y después CMD . La generación usa como máximo 128 tokens nuevos, temperatura 1, sin top-k ni top-p. PAD y BOS se eliminan del soporte para ambos modelos; EOS detiene la generación. El decodificador consulta los 256 últimos tokens con posiciones recalculadas a cero. Los sorteos categóricos usan un generador Python distinto por caso y probabilidades realmente calculadas, para conservar las mismas semillas pese a salidas de longitud diferente.
El informe distingue sintaxis válida, transiciones correctas entre las salidas analizables, validez global y truncamientos. La validez global exige EOS, sintaxis completa y todas las transiciones autorizadas. Una salida que alcanza el límite sigue siendo un fallo global aunque su último byte cierre una línea válida. Los errores UTF-8 y las salidas no analizables permanecen en el denominador de 600. Las copias exactas y las copias tras eliminar el alias se cuentan por separado. Un prefijo de test compartido parcialmente con train no es una copia completa.
Ejercicio: mirar el futuro
La variante unmasked parte de los mismos pesos iniciales, con el mismo sampler y los mismos 800 pasos, pero permite la atención a posiciones futuras. Así, el objetivo siguiente puede ser visible durante el teacher forcing. Se comparan su NLL y sus 600 generaciones, sin imponer una mejora ni un empeoramiento. La prueba cambia un sufijo y mide los logits anteriores: la versión defectuosa debe revelar una influencia, la versión causal debe bloquearla. Restaurar la máscara sobre los mismos pesos defectuosos debe restablecer esta propiedad mecánica; no equivale a reentrenar correctamente esos pesos. Los resultados iniciales están protegidos mediante SHA.
Interpretar
Una NLL mejor no garantiza ni sintaxis ni comportamiento correcto en generación libre. Una validez alta en este autómata simple no demuestra ninguna aptitud para el francés general, la conducción de un taller o la seguridad física. El simulador determinista es una alternativa conocida que garantiza sus propias reglas; la red estudia una distribución, no un comando de máquina. Se ejecuta una sola semilla de aprendizaje; las tres semillas de generación no son tres entrenamientos independientes.
Construir un corpus donde las fugas sean detectables
Partición por familia antes del ventaneo
Error frecuente: Alias diferentes ocultan una misma traza: compare también una forma normalizada sin alias.
Invariante: Ninguna familia ni traza idéntica atraviesa las particiones; recuentos efectivos escritos en outputs/data-audit.json.
Archivo 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
Formar tokens, objetivos y máscara causal
Desplazamiento causal y padding
Error frecuente: Un doble desplazamiento de las etiquetas aprende el token t+2: probar sobre una secuencia conocida de cinco tokens.
Invariante: La decodificación reconstruye el texto; modificar los PAD no cambia la loss.
Archivo 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()
Escribir el decodificador y su baseline
Q/K/V, residuos y verosimilitud
Error frecuente: Invertir el sentido de la máscara permite el futuro: inspeccionar una matriz 4×4 y después los logits.
Invariante: Prueba de causalidad superada con la tolerancia fijada; número de parámetros y NLL inicial capturados.
Archivo 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())
Preentrenar, interrumpir y reanudar
Optimización y estado completo
Error frecuente: Una reanudación vuelve al inicio de la epoch y da demasiado peso a ciertas trazas: guardar el cursor o el sampler.
Invariante: Gradientes finitos, parámetros efectivamente modificados; checkpoint y reanudación probados sin cambiar las particiones.
Archivo 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},
)
Generar y separar sintaxis y comportamiento
Exposición a los propios tokens generados
Error frecuente: Excluir las salidas no analizables infla la validez: incluirlas como fallos en la puntuación global.
Invariante: 200 prefijos fijados, 600 continuaciones por variante; truncadas e inválidas contabilizadas.
Archivo 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)
Detectar un falso éxito por fuga del futuro
Ablación controlada
Error frecuente: No entregar la versión defectuosa por defecto: el ejercicio y sus resultados llevan un nombre distinto.
Invariante: La versión sin máscara falla la prueba prevista; la versión final restaurada la supera.
Archivo 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})
Fuentes consultadas
Modelización causal, Hugging Face presenta el objetivo del token siguiente; este TP programa su propio decodificador PyTorch y no importa ningún modelo Transformers. Reserva de GPU Docker Compose e instalación de NVIDIA Container Toolkit documentan los requisitos del host. Estas páginas se abrieron durante la preparación; las versiones realmente ejecutadas proceden de los locks y del informe de entorno.
Leer los resultados efectivamente obtenidos
Los 750 registros de test representan 301 127 tokens objetivo. El n-grama obtiene una NLL de 0,2344045 nat/token; el modelo inicial, 5,5835855; el modelo causal entrenado, 0,0878211, es decir, una perplejidad de 1,09179. Esta baja perplejidad con teacher forcing no garantiza registros completos en generación libre.
Sobre los mismos 200 prefijos, con tres sorteos cada uno, el n-grama produce 4 trazas globalmente válidas de 600, el modelo inicial 0 y el causal 12 (2 %). El causal alcanza el límite de 128 tokens nuevos en 577 casos; 59 salidas contienen UTF-8 no válido. Estas categorías pueden solaparse. Su exactitud de transición de 77/77 se refiere solo a las transiciones de las 12 trazas sintácticamente válidas: no significa un 100 % de éxito en los 600 casos. No se observa ninguna copia exacta de train, pero 9 continuaciones causales reproducen una traza de train tras eliminar el identificador del taller.
La ablación sin máscara da aquí más trazas globalmente válidas: 235/600, con un único truncamiento, 249 trazas sintácticamente válidas y 52 UTF-8 no válidos. Genera una media de 19.818 tokens, frente a 126.550 para el causal, EOS incluido. El presupuesto y la distribución de longitudes cuentan en esta comparación. El código no fija ninguna cuota de éxito ni obliga a que la ablación sea peor.
Su puntuación con teacher forcing de 0,0817819 nat/token se calcula con acceso a los tokens futuros. Por tanto, su exponencial no es una perplejidad autorregresiva comparable a los otros tres valores: esta puntuación no corresponde a las condicionales utilizadas al generar sin futuro. La generación de la ablación recalcula cada prefijo disponible; no se le proporciona ningún objetivo futuro de test.
La prueba de sufijo decide la propiedad de causalidad: modificar los tokens futuros cambia los logits pasados de la ablación hasta 0,906817, por encima de 10⁻⁶, frente a cero con máscara causal. Restaurar la máscara sobre los mismos pesos de ablación restablece la invariancia; este control no constituye una nueva medida de calidad generativa. El resultado observado no permite interpretar su loss con fuga como una verosimilitud causal correcta.
El experimento trata sobre un autómata ficticio, una semilla, patrones compartidos y ventanas de 256 tokens. Los 600 sorteos corresponden a 200 prefijos, no a 600 registros independientes. Un límite de generación diferente, otro corpus o un criterio de longitud útil requerirían un protocolo distinto. El recorrido conserva los malos resultados y el umbral de causalidad inicial.
Observaciones de desarrollo
Valores recalculados a partir de las salidas brutas de este recorrido. No son umbrales de ganancia; el recorrido desde el ZIP los recalcula sobre las mismas poblaciones.
{
"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"
}
