The fictional Novaforge technical service uses codes and wording that recur in its notices. We want to adapt the model's language distribution while retaining a way to consult revised notices for changing facts.
Compare notice likelihood before and after adaptation, then short answers to questions with or without context supplied by a local lexical index. Show a changed fact in a revised notice after training.
Download the complete project, revision r001
Three different interventions
CPT continues the next-token objective from pretrained weights. Here, every parameter is adapted on unlabelled notices. No question-and-answer demonstration enters this loss. SFT would learn a transformation or response style from explicit targets; that is not this experiment. Retrieval augmentation changes the context supplied at inference and can consult a recent document without changing weights.
The model is Qwen/Qwen2.5-0.5B base, revision 060db6499f32faf8b98477b0a26969ef7d8b9987, with an unchanged tokenizer and strict loading. It has 24 blocks, width 896, 14 query heads, two K/V heads, MLP 4864, and a 151936-token vocabulary. Embedding and output head are tied; unique parameter count is measured. The base model is not an instruction-tuned assistant. A short-answer format failure does not cancel a possible NLL improvement and must not be hidden.
Data and held-out conditions
NovaLex-v1 contains 500 families of eight notices: 350/75/75 families, or 2800/600/600 documents. CommonWorkshop-v1 contains 200 families of five texts: 140/30/30 families, or 700/150/150 documents. References, values, and combinations are generated before use; variants of a family do not cross partitions. Templates and vocabulary remain shared. This artificial corpus does not represent the complexity of an industrial collection.
The 700 control training texts are used for replay. The 150 control test texts never enter it. Their NLL delta measures drift only on this synthetic scope, not forgetting of all general knowledge. The 200 questions concern the first 200 test notices, never seen by CPT. Closed evaluation provides no context: the model may legitimately ignore these held-out facts. Open evaluation searches the 4000 source notices, including test notices; this consultation is declared and distinct from training.
Forty test notices have a separately created version 2 marked inference_only. Their pulse count increases by two. Exact values are in generated files; the 3→5 example explains the mechanism without imposing those same numbers on every case. v2 notices are never mounted in learning.
Targets, boundaries, and exact budget
Each text is tokenised without a chat template or implicit token addition. Identifier 151643 explicitly acts as the anchor at the start and EOS at the end. The same ID is used for padding, but attention_mask and labels are built by position: a real EOS remains a target and PAD is ignored. Windows have at most 512 tokens and a stride of 511. Every next token, including EOS, appears exactly once in a document run, with no concatenation between documents. Context is reset per window.
The initial proposal of 2048 targets per update does not divide exactly into integer 80/20 shares. Before any execution, the protocol fixes 2050 targets × 512 updates = 1049600 targets per branch, 1024 more than proposed. Domain-only: 2050 business targets per step. Replay: exactly 1640 business and 410 control targets, thus 839680 and 209920 in total. Replay receives fewer business tokens despite the same total budget: that is part of the intervention, not a detail to omit.
A sampler traverses window permutations. When the quota cuts a window, only selected target positions contribute to loss; the complete window remains available as causal context. The next update resumes at the first unconsumed target token. Future tokens in the tensor cannot influence earlier logits because of the causal mask. The log retains the window, target start/end, counters, and stream type. The ratio therefore does not depend on document length.
Microbatches each contain a document selection. The network forwards in BF16 autocast; master weights, gradients, and AdamW states are FP32. Selected logits convert to FP32 for CE. BF16 is checked on the actual GPU; no GradScaler is used. Eager attention and deterministic algorithms are fixed, TF32 disabled. Forward cost also depends on context reread when a quota cuts a window: equal supervised tokens do not imply perfectly equal FLOPs.
AdamW uses maximum lr 0.00001, weight_decay 0.01, 51 warm-up updates, then linear decay to 10% of the maximum rate; norm clipping is 1. The final checkpoint at 512 is fixed. Validation every 128 steps is reported without selection. Interruption at 200 and resumption must recover exactly model, optimiser, scheduler, random states, permutations, document cursors, and token offsets. .pt files are locally produced states and must not be replaced with untrusted content. Per-tensor hashes and strictly reread Safetensors check final weights.
NLL and answers do not measure the same thing
Held-out NLL uses the same tokens and denominators for base, domain CPT, and replay CPT: 600 business notices and 150 control texts. Per-token losses and complete reference logits are retained. The test compares the sum of active losses, streaming accumulation, and the Transformers reference on the same batch. A lower NLL indicates these texts become more probable; it proves neither a correct answer nor memorisation of unseen facts.
The three weight states each answer the same 200 questions with empty context and then lexical top-3: six explicit conditions. The prompt is identical except for context and is not adjusted on test results. Greedy generation allows 32 new tokens, without grammar constraints or selecting the first number in a sentence. An answer must be entirely numeric after outer whitespace is removed and end with EOS. UNKNOWN, stray text, and truncation are exact-match failures. Raw outputs and reasons remain available.
Lexical index and provenance
TF-IDF is implemented with tf=1+log(count), idf=log((N+1)/(df+1))+1, L2 normalisation, and cosine. It indexes only the text of the 4000 notices, never answer labels. Queries are tokenised in the same way; score ties are broken by identifier. The three retained notices appear in the prompt with their ID and version. Recall of the expected document in the top-3 is calculated separately from generative accuracy: finding the right text does not guarantee it is used correctly.
The v1 index remains immutable. The update creates a new v2 index containing exactly 40 replacements and their new hashes, then recalculates search. There is no context cache that could hide this revision. Assembly rejects an inconsistent result/version pair; an overly long prompt is rejected without silent truncation. The open prompt does not exceed 1024 tokens.
Revision exercise
After CPT, weights and main results are protected by SHA. The exercise actually runs 40 answers for each of the six conditions with index v2. It compares v1 answers with old and new facts, then v2 answers with new facts. Closed prompts must remain identical; their output stability is measured. The number of new training steps is zero.
Better availability of a recent fact does not guarantee a better answer. Search and generation results, as well as weight hashes, are therefore exposed separately. Replay can limit control NLL drift, without a guarantee and with fewer business tokens. Only one training seed is tested; data would need to be repeated and expanded before concluding that an adaptation strategy is useful in production.
Prepare notices and a separate control
Document partition and versions
Invariant: Separate counts and families; 40 v2 revisions marked as an inference addition.
Frequent error: A notice and its paraphrase cross splits: partition before variants.
File src/preentrainement_continu/data.py
import hashlib
from pathlib import Path
from .config import save_json, sha
MODES = ["azur", "brume", "cuivre", "delta", "ambre", "givre", "orbe", "silex"]
def split_families(family, kind):
a, b = (350, 425) if kind == "domain" else (140, 170)
return "train" if family < a else "validation" if family < b else "test"
def notice_text(family, j, pulses, version=1):
alias = f"N{family:04d}"
mode = MODES[j]
lead = f"Novaforge {alias}, notice {mode}, revision {version}. "
variants = [
f"The {mode} mode waits for {pulses} pulses before its cycle.",
f"Before starting a cycle, {mode} mode expects {pulses} pulses.",
f"The pulse count before a cycle in {mode} mode is {pulses}.",
f"A cycle under {mode} mode begins after {pulses} pulses.",
f"For {mode} mode, wait for {pulses} pulses before the cycle.",
f"The {mode} cycle has a waiting count of {pulses} pulses.",
f"The required wait before the {mode} cycle is {pulses} pulses.",
f"In {mode} mode, {pulses} pulses precede the cycle.",
]
return (
lead
+ variants[j]
+ " This is a fictional language example, not an operating instruction."
)
def generate_notices():
pages = []
facts = []
for family in range(500):
for j in range(8):
value = 1 + (family * 7 + j * 3) % 9
text = notice_text(family, j, value)
identifier = f"N{family:04d}-{j}"
pages.append(
{
"document_id": identifier,
"kind": "domain",
"family": family,
"formulation": j,
"split": split_families(family, "domain"),
"version": 1,
"text": text,
"content_sha256": hashlib.sha256(text.encode()).hexdigest(),
}
)
facts.append({"document_id": identifier, "pulses": value})
for family in range(200):
alias = f"W{family:04d}"
count = 4 + family % 17
place = ["north", "south", "east", "west"][family % 4]
phrases = [
f"Workshop {alias} keeps {count} notebooks on the {place} shelf. The inventory is a written record.",
f"The {place} storage room of workshop {alias} contains {count} empty boxes. A clerk records their presence.",
f"A team at workshop {alias} reads {count} short reports near the {place} desk. The reports describe routine observations.",
f"Workshop {alias} has {count} labelled drawers in its {place} cabinet. A list records each drawer.",
f"The {place} noticeboard at workshop {alias} holds {count} paper cards. The cards remain available for the next shift.",
]
for j, text in enumerate(phrases):
pages.append(
{
"document_id": f"W{family:04d}-{j}",
"kind": "control",
"family": family,
"formulation": j,
"split": split_families(family, "control"),
"version": 1,
"text": text,
"content_sha256": hashlib.sha256(text.encode()).hexdigest(),
}
)
if len({p["text"].casefold() for p in pages}) != 5000:
raise ValueError("Duplicate normalized source")
return pages, facts
def build_revision_set(pages, facts):
lookup = {r["document_id"]: r["pulses"] for r in facts}
reserved = [p for p in pages if p["kind"] == "domain" and p["split"] == "test"][
:200
]
questions = []
gold = []
revisions = []
for index, p in enumerate(reserved):
alias = f"N{p['family']:04d}"
mode = MODES[p["formulation"]]
qid = f"Q{index:03d}"
question = (
f"How many pulses precede the cycle in {mode} mode for Novaforge {alias}?"
)
questions.append({"question_id": qid, "question": question})
value = lookup[p["document_id"]]
gold.append(
{
"question_id": qid,
"document_id": p["document_id"],
"v1_answer": str(value),
"v2_answer": str(value + 2) if index < 40 else str(value),
"revised": index < 40,
}
)
if index < 40:
text = notice_text(p["family"], p["formulation"], value + 2, 2)
revisions.append(
{
**p,
"version": 2,
"text": text,
"content_sha256": hashlib.sha256(text.encode()).hexdigest(),
"inference_only": True,
}
)
return questions, gold, revisions
def prepare(config):
root = Path("data/generated")
root.mkdir(parents=True, exist_ok=True)
if (root / "manifest.json").exists():
raise ValueError("Preserve corpus; use a clean extraction")
pages, facts = generate_notices()
save_json(root / "input/pages.json", pages)
questions, gold, revisions = build_revision_set(pages, facts)
save_json(root / "input/queries.json", questions)
save_json(root / "input/revisions.json", revisions)
save_json(root / "labels/gold.json", gold)
audit = {}
for kind in ["domain", "control"]:
audit[kind] = {}
for split in ["train", "validation", "test"]:
selected = [p for p in pages if p["kind"] == kind and p["split"] == split]
save_json(root / f"{split}/{kind}.json", selected)
audit[kind][split] = {
"documents": len(selected),
"families": len({p["family"] for p in selected}),
}
manifest = {
"datasets": ["NovaLex-v1", "CommonWorkshop-v1"],
"rights": "CC0-1.0",
"seed": config["seed"],
"split_audit": audit,
"questions": len(questions),
"revisions": len(revisions),
"normalized_duplicates": 0,
"files": {str(f.relative_to(root)): sha(f) for f in root.rglob("*.json")},
}
save_json(root / "manifest.json", manifest)
save_json("outputs/data-audit.json", manifest)
return audit
Acquire the model and inspect its NLL
Base model, tokens, and targets
Invariant: Resolved SHA, reproducible download, and exact target counts.
Frequent error: Removing EOS as though it were PAD cuts learning of endings: IDs and attention_mask are distinct.
File src/preentrainement_continu/batch.py
import random, torch
EOS = 151643
def encode_documents(tokenizer, pages, context=512):
if context < 2:
raise ValueError("Context needs an input and a target")
rows = []
for page in pages:
content = tokenizer.encode(page["text"], add_special_tokens=False)
if not content:
raise ValueError("Empty document")
ids = [EOS] + content + [EOS]
for start in range(0, len(ids) - 1, context - 1):
rows.append(
{
"document_id": page["document_id"],
"kind": page["kind"],
"window_start": start,
"ids": ids[start : start + context],
}
)
return rows
def collate_documents(selections, rows, device="cuda", pad_to=None):
if not selections:
raise ValueError("Empty token budget")
width = (
max(len(rows[s["window"]]["ids"]) for s in selections)
if pad_to is None
else pad_to
)
ids = torch.full((len(selections), width), EOS, dtype=torch.long, device=device)
mask = torch.zeros_like(ids, dtype=torch.bool)
labels = torch.full_like(ids, -100)
for i, selection in enumerate(selections):
row = rows[selection["window"]]
tokens = row["ids"]
start = selection["target_start"]
end = selection["target_end"]
if len(tokens) > width or not 1 <= start < end <= len(tokens):
raise ValueError("Invalid token slice; no silent truncation")
ids[i, : len(tokens)] = torch.tensor(tokens, device=device)
mask[i, : len(tokens)] = True
labels[i, start:end] = torch.tensor(tokens[start:end], device=device)
return ids, mask, labels
def sum_token_nll(logits, labels):
active = labels[:, 1:] != -100
if not active.any():
raise ValueError("No active shifted targets")
selected = logits[:, :-1][active].float()
targets = labels[:, 1:][active]
values = torch.nn.functional.cross_entropy(selected, targets, reduction="none")
return values.sum(), int(active.sum()), values
class TokenStream:
def __init__(self, rows, seed):
if not rows or any(len(r["ids"]) < 2 for r in rows):
raise ValueError("A stream needs nonempty next-token targets")
self.rows = rows
self.generator = random.Random(seed)
self.order = []
self.cursor = 0
self.target_offset = 0
self.epoch = 0
self.tokens = 0
def take(self, quota):
if quota < 0:
raise ValueError("Negative quota")
result = []
while quota:
if self.cursor == len(self.order):
self.order = list(range(len(self.rows)))
self.generator.shuffle(self.order)
self.cursor = 0
self.epoch += 1
index = self.order[self.cursor]
row = self.rows[index]
remaining = len(row["ids"]) - 1 - self.target_offset
take = min(quota, remaining)
start = 1 + self.target_offset
result.append(
{
"window": index,
"document_id": row["document_id"],
"kind": row["kind"],
"window_start": row["window_start"],
"target_start": start,
"target_end": start + take,
}
)
self.target_offset += take
self.tokens += take
quota -= take
if self.target_offset == len(row["ids"]) - 1:
self.target_offset = 0
self.cursor += 1
return result
def state_dict(self):
return {
"order": list(self.order),
"cursor": self.cursor,
"target_offset": self.target_offset,
"epoch": self.epoch,
"tokens": self.tokens,
"rng": self.generator.getstate(),
}
def load_state_dict(self, state):
if sorted(state["order"]) != list(range(len(self.rows))) or not 0 <= state[
"cursor"
] <= len(self.rows):
raise ValueError("Invalid stream order/cursor")
if state["cursor"] == len(self.rows):
if state["target_offset"] != 0:
raise ValueError("Invalid completed epoch offset")
elif (
not 0
<= state["target_offset"]
< len(self.rows[state["order"][state["cursor"]]]["ids"]) - 1
):
raise ValueError("Invalid token cursor")
self.order = list(state["order"])
self.cursor = state["cursor"]
self.target_offset = state["target_offset"]
self.epoch = state["epoch"]
self.tokens = state["tokens"]
self.generator.setstate(state["rng"])
class TokenMixer:
def __init__(self, domain, control, seed, branch):
if branch not in ["domain", "replay"]:
raise ValueError("Unknown CPT branch")
self.branch = branch
self.domain = TokenStream(domain, seed + 100)
self.control = TokenStream(control, seed + 200)
self.updates = 0
def take(self):
domain = self.domain.take(1640 if self.branch == "replay" else 2050)
control = self.control.take(410) if self.branch == "replay" else []
self.updates += 1
return domain, control
def state_dict(self):
return {
"branch": self.branch,
"updates": self.updates,
"domain": self.domain.state_dict(),
"control": self.control.state_dict(),
}
def load_state_dict(self, state):
if state["branch"] != self.branch:
raise ValueError("Changed replay policy")
self.domain.load_state_dict(state["domain"])
if self.branch == "replay":
self.control.load_state_dict(state["control"])
elif state["control"] != self.control.state_dict():
raise ValueError("Control stream changed in domain-only branch")
self.updates = state["updates"]
Set the baseline and the open index
Weights versus search context
Invariant: Queries run without reading target; every context cites an existing notice.
Frequent error: The answer field is indexed inadvertently: index only source documents.
File src/preentrainement_continu/retrieval.py
import collections, hashlib, json, math, re, time
from pathlib import Path
from .config import read_json, save_json, sha
def terms(text):
return re.findall(r"\w+", text.casefold())
def vector(text, idf):
counts = collections.Counter(terms(text))
weights = {
word: (1 + math.log(n)) * idf[word]
for word, n in sorted(counts.items())
if word in idf
}
norm = math.sqrt(math.fsum(v * v for v in weights.values()))
return {word: value / norm for word, value in weights.items()} if norm else {}
def fit_tfidf_index(documents):
begun = time.perf_counter()
if not documents or len({d["document_id"] for d in documents}) != len(documents):
raise ValueError("Empty or duplicate source ID")
frequencies = collections.Counter()
for document in documents:
if document["kind"] != "domain":
raise ValueError("The lexical index contains domain source texts only")
if (
hashlib.sha256(document["text"].encode()).hexdigest()
!= document["content_sha256"]
):
raise ValueError("Source digest changed")
frequencies.update(set(terms(document["text"])))
n = len(documents)
idf = {
word: math.log((n + 1) / (count + 1)) + 1
for word, count in sorted(frequencies.items())
}
rows = [
{
"document_id": d["document_id"],
"version": d["version"],
"text": d["text"],
"content_sha256": d["content_sha256"],
"vector": vector(d["text"], idf),
}
for d in documents
]
return {
"documents": rows,
"idf": idf,
"document_frequency": dict(sorted(frequencies.items())),
"document_count": n,
"build_seconds": time.perf_counter() - begun,
"formula": "tf=1+log(count), idf=log((N+1)/(df+1))+1, L2 cosine; source text only",
}
def retrieve_notices(index, question, k=3):
if k < 1:
raise ValueError("Positive retrieval count required")
query = vector(question, index["idf"])
scores = []
for document in index["documents"]:
score = math.fsum(
query[word] * document["vector"].get(word, 0.0) for word in sorted(query)
)
scores.append(
{
"document_id": document["document_id"],
"version": document["version"],
"score": score,
}
)
return sorted(scores, key=lambda row: (-row["score"], row["document_id"]))[:k]
def build_question_context(question, index, hits, condition):
if condition not in ["closed", "open"]:
raise ValueError("Name the evaluation condition")
lookup = {d["document_id"]: d for d in index["documents"]}
documents = [lookup[h["document_id"]] for h in hits] if condition == "open" else []
for d, h in zip(documents, hits):
if d["version"] != h["version"]:
raise ValueError("Stale retrieval version")
context = (
"\n".join(f"[{d['document_id']}@{d['version']}] {d['text']}" for d in documents)
if documents
else "None supplied."
)
text = (
"Answer with digits only. If the answer is unavailable, write UNKNOWN.\nContext:\n"
+ context
+ "\nQuestion: "
+ question
+ "\nAnswer:"
)
return text, [
{
"document_id": d["document_id"],
"version": d["version"],
"content_sha256": d["content_sha256"],
}
for d in documents
]
def build_index(version=1):
if Path("data/generated/labels").exists():
raise ValueError("Indexing service must not mount answer labels")
pages = [
p for p in read_json("data/generated/input/pages.json") if p["kind"] == "domain"
]
changed = []
if version == 2:
revisions = read_json("data/generated/input/revisions.json")
lookup = {p["document_id"]: p for p in pages}
for revision in revisions:
old = lookup[revision["document_id"]]
if (
old["version"] != 1
or revision["version"] != 2
or revision["split"] != "test"
or not revision["inference_only"]
):
raise ValueError("Unexpected revision boundary")
lookup[revision["document_id"]] = revision
changed.append(revision["document_id"])
pages = [lookup[p["document_id"]] for p in pages]
elif version != 1:
raise ValueError("Expected index version1or2")
path = Path(f"outputs/index-v{version}.json")
if path.exists():
raise ValueError("Never overwrite an index snapshot")
index = fit_tfidf_index(pages)
save_json(path, {**index, "snapshot_version": version, "revised_ids": changed})
questions = read_json("data/generated/input/queries.json")
rows = []
for q in questions:
begun = time.perf_counter()
hits = retrieve_notices(index, q["question"], 3)
rows.append({**q, "hits": hits, "seconds": time.perf_counter() - begun})
save_json(
f"outputs/retrieval-v{version}.json",
{
"index_sha256": sha(path),
"snapshot_version": version,
"rows": rows,
"answer_labels_visible": False,
},
)
print(
{
"snapshot_version": version,
"documents": len(pages),
"revised_documents": len(changed),
"build_seconds": index["build_seconds"],
},
flush=True,
)
Adapt with and without replay
Token budget and explicit mixture
Invariant: Weights change and counters match the protocol; base remains immutable.
Frequent error: An announced 80/20 mix becomes 95/5 because of lengths: control the budget by tokens.
File src/preentrainement_continu/train.py
import math, time, shutil
from pathlib import Path
import torch
from .config import read_json, save_json, sha, seed_all, rng_state, restore_rng
from .model import (
load_base,
load_tokenizer,
parameter_manifest,
manifest_digest,
save_learned,
)
from .batch import encode_documents, collate_documents, sum_token_nll, TokenMixer
from .state import equal, save_resume_state
def lr_at(config, step):
warmup = config["warmup_updates"]
peak = config["learning_rate"]
total = config["optimizer_steps"]
return (
peak * (step + 1) / warmup
if step < warmup
else peak
* (
1
- min(1.0, (step - warmup) / (total - warmup - 1))
* (1 - config["minimum_lr_ratio"])
)
)
def training_rows(config, tokenizer):
result = {}
for kind, count in [("domain", 2800), ("control", 700)]:
pages = read_json(f"data/generated/train/{kind}.json")
if len(pages) != count or any(
p["split"] != "train" or p["kind"] != kind or p["version"] != 1
for p in pages
):
raise ValueError("Replay/train/version boundary violated")
result[kind] = encode_documents(tokenizer, pages, config["context"])
return result
@torch.no_grad()
def validation(model, rows):
model.eval()
total = 0.0
count = 0
for index, row in enumerate(rows):
selection = {"window": index, "target_start": 1, "target_end": len(row["ids"])}
ids, mask, labels = collate_documents([selection], rows)
with torch.autocast("cuda", dtype=torch.bfloat16):
logits = model(input_ids=ids, attention_mask=mask, use_cache=False).logits
_, n, values = sum_token_nll(logits, labels)
total += float(values.double().sum())
count += n
model.train()
return {
"windows": len(rows),
"active_tokens": count,
"nll_nats_per_token": total / count,
}
def train_cpt(config, branch, mode):
if (
Path("data/generated/test").exists()
or Path("data/generated/input").exists()
or Path("data/generated/labels").exists()
):
raise ValueError("Learning container must only mount train and validation")
if mode not in ["interrupt", "continuous", "resume"]:
raise ValueError("Unknown mode")
if not torch.cuda.is_bf16_supported():
raise ValueError("This reference profile requires CUDA bfloat16 support")
expected = (2050, 1049600, 1640, 410, 512)
if (
tuple(
config[k]
for k in [
"tokens_per_update",
"total_active_tokens",
"replay_domain_tokens_per_update",
"replay_control_tokens_per_update",
"optimizer_steps",
]
)
!= expected
):
raise ValueError("Frozen exact80/20token budget changed")
seed_all(config["seed"])
tokenizer = load_tokenizer()
rows = training_rows(config, tokenizer)
model = load_base()
initial = parameter_manifest(model)
initial_digest = manifest_digest(initial)
model = model.cuda()
optimizer = torch.optim.AdamW(
model.parameters(),
lr=config["learning_rate"],
weight_decay=config["weight_decay"],
)
mixer = TokenMixer(rows["domain"], rows["control"], config["seed"], branch)
validation_rows = {
k: encode_documents(
tokenizer,
read_json(f"data/generated/validation/{k}.json"),
config["context"],
)
for k in ["domain", "control"]
}
identity = {
"branch": branch,
"config": {k: v for k, v in config.items() if k != "exercise"},
"initial_model_digest": initial_digest,
"base_sha256": sha("assets/model/model.safetensors"),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"model_config_sha256": sha("assets/model/config.json"),
"model_lock_sha256": sha("locks/models.lock.json"),
"training_sha256": {
k: sha(f"data/generated/train/{k}.json") for k in ["domain", "control"]
},
"validation_sha256": {
k: sha(f"data/generated/validation/{k}.json") for k in ["domain", "control"]
},
"implementation_sha256": {
n: sha("src/preentrainement_continu/" + n)
for n in ["model.py", "batch.py", "train.py", "state.py"]
},
}
root = Path("assets/checkpoints")
destination = root / f"{branch}-{mode}.pt"
history = []
validation_history = []
start = 0
if destination.exists():
raise ValueError("Preserve recorded checkpoint")
if mode == "resume":
state = torch.load(
root / f"{branch}-interrupt.pt", map_location="cpu", weights_only=False
)
if state["identity"] != identity or state["scheduler"] != {
"next_step": state["steps"],
"next_lr": lr_at(config, state["steps"]),
}:
raise ValueError("Resume identity or scheduler changed")
model.load_state_dict(state["model"], strict=True)
optimizer.load_state_dict(state["optimizer"])
mixer.load_state_dict(state["mixer"])
restore_rng(state["rng"])
start = state["steps"]
history = state["history"]
validation_history = state["validation_history"]
del state
stop = (
config["interrupt_step"] if mode == "interrupt" else config["optimizer_steps"]
)
begun = time.perf_counter()
torch.cuda.reset_peak_memory_stats()
model.train()
logit_dtype = None
for step in range(start, stop):
domain, control = mixer.take()
denominator = sum(r["target_end"] - r["target_start"] for r in domain + control)
if denominator != 2050:
raise ValueError("Token budget drift")
lr = lr_at(config, step)
for group in optimizer.param_groups:
group["lr"] = lr
optimizer.zero_grad(set_to_none=True)
loss_total = 0.0
for kind, selections in [("domain", domain), ("control", control)]:
for selection in selections:
ids, mask, labels = collate_documents([selection], rows[kind])
with torch.autocast("cuda", dtype=torch.bfloat16):
logits = model(
input_ids=ids, attention_mask=mask, use_cache=False
).logits
logit_dtype = str(logits.dtype)
summed, n, values = sum_token_nll(logits, labels)
if (
not torch.isfinite(summed)
or n != selection["target_end"] - selection["target_start"]
):
raise ValueError("Nonfinite or miscounted loss")
(summed / denominator).backward()
loss_total += float(values.detach().double().sum())
norm = torch.nn.utils.clip_grad_norm_(
model.parameters(), config["gradient_clip"], error_if_nonfinite=True
)
if any(
p.dtype != torch.float32 or p.grad is None or p.grad.dtype != torch.float32
for p in model.parameters()
):
raise ValueError("Expected finite FP32 master gradients")
optimizer.step()
history.append(
{
"step": step + 1,
"selections": domain + control,
"active_tokens": denominator,
"domain_tokens": sum(
r["target_end"] - r["target_start"] for r in domain
),
"control_tokens": sum(
r["target_end"] - r["target_start"] for r in control
),
"total_tokens_seen": mixer.domain.tokens + mixer.control.tokens,
"nll_nats_per_token": loss_total / denominator,
"learning_rate": lr,
"gradient_norm_before_clip": float(norm),
}
)
if (step + 1) % config["validation_every"] == 0:
observed = {k: validation(model, v) for k, v in validation_rows.items()}
validation_history.append({"step": step + 1, "results": observed})
print(
{
"branch": branch,
"mode": mode,
"step": step + 1,
"total_tokens_seen": history[-1]["total_tokens_seen"],
"training_nll": history[-1]["nll_nats_per_token"],
"validation": observed,
},
flush=True,
)
elif (step + 1) % 32 == 0:
print(
{
"branch": branch,
"mode": mode,
"step": step + 1,
"total_tokens_seen": history[-1]["total_tokens_seen"],
"training_nll": history[-1]["nll_nats_per_token"],
},
flush=True,
)
torch.cuda.synchronize()
final = parameter_manifest(model)
final_digest = manifest_digest(final)
if final_digest == initial_digest or logit_dtype != "torch.bfloat16":
raise ValueError("Expected real updates and BF16 forward logits")
state = {
"identity": identity,
"steps": stop,
"model": model.state_dict(),
"optimizer": optimizer.state_dict(),
"scheduler": {"next_step": stop, "next_lr": lr_at(config, stop)},
"mixer": mixer.state_dict(),
"rng": rng_state(),
"history": history,
"validation_history": validation_history,
}
save_resume_state(destination, state)
if mode == "resume":
save_learned(model, root / f"{branch}-resume.safetensors", identity)
save_json(
f"outputs/train-{branch}-{mode}.json",
{
"branch": branch,
"mode": mode,
"identity": identity,
"steps": stop,
"history": history,
"validation_history": validation_history,
"tokenized_train": rows,
"initial_parameters": initial,
"final_parameters": final,
"initial_model_digest": initial_digest,
"final_model_digest": final_digest,
"parameters": sum(p.numel() for p in model.parameters()),
"active_tokens": mixer.domain.tokens + mixer.control.tokens,
"domain_tokens": mixer.domain.tokens,
"control_tokens": mixer.control.tokens,
"checkpoint_sha256": sha(destination),
"checkpoint_bytes": destination.stat().st_size,
"precision": {
"master": "torch.float32",
"gradients": "torch.float32",
"logits": logit_dtype,
"ce": "torch.float32",
"scaler": False,
},
"seconds": time.perf_counter() - begun,
"max_torch_allocated_bytes": torch.cuda.max_memory_allocated(),
"test_visible": False,
"v2_visible": False,
},
)
def checkpoint_mixer(branch):
root = Path("assets/checkpoints")
a = torch.load(
root / f"{branch}-continuous.pt", map_location="cpu", weights_only=False
)
b = torch.load(root / f"{branch}-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("CPT resume differs: " + str(checks))
save_json(
f"outputs/resume-{branch}.json",
{
"status": "PASS",
"all_state_exact": True,
"checks": checks,
"boundary": "complete2050targettoken optimizer update; includes stream target offset",
},
)
Measure NLL and answers under named conditions
Closed and open evaluation
Invariant: The same 200 questions in every condition, with every format failure counted.
Frequent error: Comparing CPT without context with a base model with context without saying so mixes two interventions.
File src/preentrainement_continu/evaluate.py
import hashlib, json, math, re, shutil, time
from pathlib import Path
import torch
from transformers import GenerationConfig
from .config import read_json, save_json, sha, seed_all
from .model import (
load_base,
load_tokenizer,
reload_learned,
parameter_manifest,
manifest_digest,
)
from .batch import encode_documents, collate_documents, sum_token_nll, EOS
from .retrieval import build_question_context
def weight_state(variant):
if variant == "base":
model = load_base()
path = Path("assets/model/model.safetensors")
identity = manifest_digest(parameter_manifest(model))
elif variant in ["domain", "replay"]:
path = Path(f"assets/checkpoints/{variant}-resume.safetensors")
model, metadata = reload_learned(path)
identity = metadata["model_digest"]
else:
raise ValueError("Unknown weight state")
return model, path, identity
def audit_base():
model = load_base()
rows = parameter_manifest(model)
save_json(
"outputs/base-parameters.json",
{
"parameters": rows,
"unique_parameters": sum(r["numel"] for r in rows),
"model_digest": manifest_digest(rows),
"model_sha256": sha("assets/model/model.safetensors"),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"all_master_fp32": all(
p.dtype == torch.float32 for p in model.parameters()
),
"tied_embedding_output": model.lm_head.weight
is model.model.embed_tokens.weight,
},
)
shutil.copyfile("assets/model/tokenizer.json", "outputs/tokenizer-reference.json")
def parse_short_answer(text, finish):
stripped = text.strip()
valid = re.fullmatch(r"[0-9]{1,3}", stripped) is not None
return {
"numeric_format": valid,
"completion_valid": valid and finish == "eos",
"answer": stripped if valid else None,
"reason": (
"TRUNCATED"
if finish != "eos"
else "VALID_INTEGER" if valid else "UNKNOWN_OR_INVALID_FORMAT"
),
}
@torch.no_grad()
def predict_questions(config, variant, version=1):
if Path("data/generated/labels").exists():
raise ValueError("Question generation must not mount answer labels")
if version not in [1, 2]:
raise ValueError("Index version must be explicit")
seed_all(config["seed"])
model, path, identity = weight_state(variant)
model = model.cuda().eval()
tokenizer = load_tokenizer()
index_path = Path(f"outputs/index-v{version}.json")
index = read_json(index_path)
retrieval = read_json(f"outputs/retrieval-v{version}.json")
if retrieval["index_sha256"] != sha(index_path):
raise ValueError("Stale retrieval snapshot")
selected = retrieval["rows"] if version == 1 else retrieval["rows"][:40]
generation = GenerationConfig(
do_sample=False,
num_beams=1,
use_cache=True,
max_new_tokens=config["max_new_tokens"],
eos_token_id=EOS,
pad_token_id=EOS,
bos_token_id=EOS,
)
result = []
begun = time.perf_counter()
torch.cuda.reset_peak_memory_stats()
for condition in ["closed", "open"]:
for q in selected:
text, sources = build_question_context(
q["question"], index, q["hits"], condition
)
ids = tokenizer.encode(text, add_special_tokens=False)
if not ids or len(ids) > config["generation_max_prompt_tokens"]:
raise ValueError(
"No prompt truncation: declared context budget exceeded"
)
input_ids = torch.tensor([ids], device="cuda")
mask = torch.ones_like(input_ids)
started = time.perf_counter()
with torch.autocast("cuda", dtype=torch.bfloat16):
generated = (
model.generate(
input_ids=input_ids,
attention_mask=mask,
generation_config=generation,
)[0, len(ids) :]
.cpu()
.tolist()
)
torch.cuda.synchronize()
finish = "eos" if generated and generated[-1] == EOS else "length"
content = generated[:-1] if finish == "eos" else generated
decoded = tokenizer.decode(
content, skip_special_tokens=False, clean_up_tokenization_spaces=False
)
result.append(
{
"question_id": q["question_id"],
"question": q["question"],
"condition": condition,
"input_text": text,
"input_ids": ids,
"source_documents": sources,
"prompt_sha256": hashlib.sha256(text.encode()).hexdigest(),
"generated_ids": generated,
"text": decoded,
"finish_reason": finish,
"parsed": parse_short_answer(decoded, finish),
"seconds": time.perf_counter() - started,
}
)
print(
{
"variant": variant,
"version": version,
"condition": condition,
"questions": len(selected),
},
flush=True,
)
save_json(
f"outputs/answers-{variant}-v{version}.json",
{
"variant": variant,
"snapshot_version": version,
"model_digest": identity,
"model_file_sha256": sha(path),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"index_sha256": sha(index_path),
"generation": generation.to_dict(),
"rows": result,
"seconds": time.perf_counter() - begun,
"max_torch_allocated_bytes": torch.cuda.max_memory_allocated(),
"answer_labels_visible": False,
"weights_updated": False,
},
)
@torch.no_grad()
def evaluate_nll(config, variant):
seed_all(config["seed"])
model, path, identity = weight_state(variant)
model = model.cuda().eval()
tokenizer = load_tokenizer()
pages = read_json("data/generated/input/pages.json")
results = {}
for kind, count in [("domain", 600), ("control", 150)]:
selected = [p for p in pages if p["kind"] == kind and p["split"] == "test"]
if len(selected) != count:
raise ValueError("Expected frozen heldout documents")
rows = encode_documents(tokenizer, selected, config["context"])
records = []
probes = []
for index, row in enumerate(rows):
selection = {
"window": index,
"target_start": 1,
"target_end": len(row["ids"]),
}
ids, mask, labels = collate_documents([selection], rows)
with torch.autocast("cuda", dtype=torch.bfloat16):
logits = model(
input_ids=ids, attention_mask=mask, use_cache=False
).logits
_, n, values = sum_token_nll(logits, labels)
records.append(
{
**row,
"active_tokens": n,
"targets": row["ids"][1:],
"token_nll": values.cpu().tolist(),
}
)
if index == 0:
for pos in [0, n - 1]:
probes.append(
{
"document_id": row["document_id"],
"target_position": pos + 1,
"target": row["ids"][pos + 1],
"logits": logits[0, pos].float().cpu().tolist(),
"nll": float(values[pos]),
}
)
total = math.fsum(v for row in records for v in row["token_nll"])
tokens = sum(r["active_tokens"] for r in records)
results[kind] = {
"documents": count,
"active_tokens": tokens,
"nll_nats_per_token": total / tokens,
"rows": records,
"probes": probes,
}
save_json(
f"outputs/nll-{variant}.json",
{
"variant": variant,
"model_digest": identity,
"model_file_sha256": sha(path),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"precision": "FP32 master weights, BF16 autocast, FP32 CE",
"results": results,
},
)
def score_short_answers(rows, gold, version):
details = []
correct = numeric = valid = truncated = 0
for row in rows:
parsed = parse_short_answer(row["text"], row["finish_reason"])
if parsed != row["parsed"]:
raise ValueError("Stored parse changed")
expected = gold[row["question_id"]][f"v{version}_answer"]
exact = parsed["completion_valid"] and parsed["answer"] == expected
correct += exact
numeric += parsed["numeric_format"]
valid += parsed["completion_valid"]
truncated += row["finish_reason"] == "length"
details.append(
{
"question_id": row["question_id"],
"expected": expected,
"exact_match": exact,
"revised": gold[row["question_id"]]["revised"],
}
)
n = len(rows)
return {
"questions": n,
"numeric_format": numeric,
"valid_completions": valid,
"exact_match": correct,
"exact_match_rate": correct / n,
"truncated": truncated,
"rows": details,
}
def evaluate_retrieval(version):
gold = {r["question_id"]: r for r in read_json("data/generated/labels/gold.json")}
retrieval = read_json(f"outputs/retrieval-v{version}.json")
details = []
for q in retrieval["rows"]:
label = gold[q["question_id"]]
expected_version = 2 if version == 2 and label["revised"] else 1
hit = any(
h["document_id"] == label["document_id"]
and h["version"] == expected_version
for h in q["hits"]
)
details.append(
{
"question_id": q["question_id"],
"expected_document_id": label["document_id"],
"expected_version": expected_version,
"hit_top3": hit,
}
)
return {
"questions": len(details),
"hit_top3": sum(r["hit_top3"] for r in details),
"recall_top3": sum(r["hit_top3"] for r in details) / len(details),
"rows": details,
}
def compare_results(config):
gold = {r["question_id"]: r for r in read_json("data/generated/labels/gold.json")}
result = {}
base = read_json("outputs/nll-base.json")
for variant in ["base", "domain", "replay"]:
answers = read_json(f"outputs/answers-{variant}-v1.json")
nll = read_json(f"outputs/nll-{variant}.json")
result[variant] = {
"nll": {
kind: {
"documents": r["documents"],
"active_tokens": r["active_tokens"],
"nll_nats_per_token": r["nll_nats_per_token"],
"delta_from_base": r["nll_nats_per_token"]
- base["results"][kind]["nll_nats_per_token"],
}
for kind, r in nll["results"].items()
},
"answers": {
condition: score_short_answers(
[r for r in answers["rows"] if r["condition"] == condition], gold, 1
)
for condition in ["closed", "open"]
},
}
save_json(
"outputs/comparison.json",
{
"variants": result,
"retrieval": evaluate_retrieval(1),
"quality_gain_required": False,
"control_scope": "150synthetic heldout workshop texts, not a benchmark of all general capabilities",
"open_book_test": True,
},
)
Update a fact without updating weights
Counterfactual version test
Invariant: The index refers to v2 and weights retain their hash.
Frequent error: The old context cache survives the revision: include document version in its key.
File src/preentrainement_continu/exercise.py
from pathlib import Path
from .config import read_json, save_json, sha
from .evaluate import score_short_answers, evaluate_retrieval
def preserve_main():
paths = [
"assets/model/model.safetensors",
"assets/checkpoints/domain-resume.safetensors",
"assets/checkpoints/replay-resume.safetensors",
"outputs/index-v1.json",
"outputs/retrieval-v1.json",
"outputs/comparison.json",
]
for variant in ["base", "domain", "replay"]:
paths.extend(
[f"outputs/nll-{variant}.json", f"outputs/answers-{variant}-v1.json"]
)
save_json("outputs/exercise-before.json", {name: sha(name) for name in paths})
def compare_revision_answers():
before = read_json("outputs/exercise-before.json")
if any(sha(name) != h for name, h in before.items()):
raise ValueError("Revision exercise changed a weight or main artifact")
gold = {r["question_id"]: r for r in read_json("data/generated/labels/gold.json")}
selected = {r["question_id"] for r in gold.values() if r["revised"]}
result = {}
if len(selected) != 40:
raise ValueError("Expected40revised notices")
for variant in ["base", "domain", "replay"]:
old = read_json(f"outputs/answers-{variant}-v1.json")
new = read_json(f"outputs/answers-{variant}-v2.json")
if (
old["model_digest"] != new["model_digest"]
or old["model_file_sha256"] != new["model_file_sha256"]
):
raise ValueError("Weights changed between document revisions")
result[variant] = {}
for condition in ["closed", "open"]:
a = [
r
for r in old["rows"]
if r["condition"] == condition and r["question_id"] in selected
]
b = [r for r in new["rows"] if r["condition"] == condition]
if (
len(a) != 40
or len(b) != 40
or [r["question_id"] for r in a] != [r["question_id"] for r in b]
):
raise ValueError("Revised question population changed")
same_inputs = [x["input_ids"] == y["input_ids"] for x, y in zip(a, b)]
same_outputs = [
x["generated_ids"] == y["generated_ids"] for x, y in zip(a, b)
]
if condition == "closed" and not all(same_inputs):
raise ValueError("Closed condition gained context")
result[variant][condition] = {
"v1_answers_against_old_facts": score_short_answers(a, gold, 1),
"v1_answers_against_new_facts": score_short_answers(a, gold, 2),
"v2_answers_against_new_facts": score_short_answers(b, gold, 2),
"identical_input_count": sum(same_inputs),
"identical_output_count": sum(same_outputs),
}
save_json(
"outputs/exercise/revised-notices.json",
{
"variants": result,
"retrieval_v1": evaluate_retrieval(1),
"retrieval_v2": evaluate_retrieval(2),
"protected_sha256": before,
"weights_and_main_artifacts_unchanged": True,
"new_training_steps": 0,
"quality_gain_required": False,
},
)
Sources
Don’t Stop Pretraining provides context for domain adaptation; its results are not those of this causal practical exercise. Qwen2.5-0.5B model card and Transformers causal objective were consulted. Loading and executed versions are pinned locally. Docker GPU and NVIDIA Container Toolkit document prerequisites.
Results observed during GPU development
The complete run succeeded with 14 tests in 13,673.51 seconds. Both branches perform 512 updates and 1,049,600 supervised targets. Replay replaces 20% of business targets with 209,920 control targets, retaining 839,680 business targets. It therefore does not see as much business text as the dedicated branch. Resumptions after 200 steps exactly recover weights, optimiser, token streams, RNGs, and histories of continuous paths. Reproduction from the public ZIP remains a separate verification.
Across 600 test business notices, or 25,275 targets, NLL falls from 5.197560 to 0.429035 after domain CPT and to 0.417514 with replay. Across 150 control test texts, or 3,840 targets, it rises from 5.193446 to 6.220325 after domain CPT, versus 1.255818 with replay. This shows drift on this limited fictional control and its mitigation by mixing. This control does not measure all general abilities of the model. Validation texts are distinct and their token counts must not replace test counts.
Loss and free-answer quality do not tell the same story. Without context, the base model answers exactly 13/200 questions, compared with 21/200 for each trained branch. With the three retrieved documents, it reaches 200/200, compared with 168/200 after domain CPT and 178/200 with replay. Top-3 document recall is 200/200: errors from trained branches remain even when the correct notice is supplied. No CPT advantage over the base model's document reading is observed. The 41 truncated outputs of the base model without context remain in the denominator; no partial answer is repaired to improve the score.
The exercise revises 40 facts in documents that can be consulted, without training the models again. On new values, consulting index v2 gives 39/40 exact answers with the base model, 34/40 after domain CPT, and 36/40 with replay. Without documents, inputs and outputs remain identical in all 40 cases; no answer is correct for revised values. Top-3 recall of the revised index is 199/200 over all questions. Updating a document can therefore expose a new fact without guaranteeing every case is read correctly.
The retained thread is notice N0425-0: the value “6 pulses” becomes “8 pulses”. All three variants answer 6 with the old document and 8 with its revision, ending with EOS. This example is one successful case among complete populations, not a substitute for their evaluation. Consulted test notices never served CPT training: this is explicitly a test with supplied documents.
Continuous paths take 3029.10 s (business) and 3412.02 s (replay), including validation and saving. Their corresponding PyTorch allocation peaks are 9990552576 and 9984970752 bytes. They do not represent all VRAM visible to Linux. Interrupted and resumed runs, acquisition, tests, inference, and audits explain the higher cost of the complete workflow. CPU preparation of other tutorials took place during this run: no measurement on a fully idle host is claimed.
One seed, a closed synthetic corpus, and a bounded numeric response format are insufficient to establish a general business benefit. Replay has a measured benefit here on the language control; CPT is neither an updatable document base nor a guarantee of factual accuracy.
Development observations
Values recalculated from the raw outputs of this run. They are not improvement thresholds; the workflow from the ZIP recalculates them over the same populations.
{
"status": "PASS",
"domain_documents": 4000,
"control_documents": 1000,
"questions": 200,
"revised_questions": 40,
"cpt_tokens_per_branch": 1049600,
"replay_domain_tokens": 839680,
"replay_control_tokens": 209920,
"metrics": {
"base": {
"nll": {
"domain": {
"documents": 600,
"active_tokens": 25275,
"nll_nats_per_token": 5.197560175851901,
"delta_from_base": 0.0
},
"control": {
"documents": 150,
"active_tokens": 3840,
"nll_nats_per_token": 5.19344591589955,
"delta_from_base": 0.0
}
},
"answers": {
"closed": {
"questions": 200,
"numeric_format": 159,
"valid_completions": 159,
"exact_match": 13,
"exact_match_rate": 0.065,
"truncated": 41
},
"open": {
"questions": 200,
"numeric_format": 200,
"valid_completions": 200,
"exact_match": 200,
"exact_match_rate": 1.0,
"truncated": 0
}
}
},
"domain": {
"nll": {
"domain": {
"documents": 600,
"active_tokens": 25275,
"nll_nats_per_token": 0.42903490273272693,
"delta_from_base": -4.768525273119174
},
"control": {
"documents": 150,
"active_tokens": 3840,
"nll_nats_per_token": 6.220325457319268,
"delta_from_base": 1.026879541419718
}
},
"answers": {
"closed": {
"questions": 200,
"numeric_format": 200,
"valid_completions": 200,
"exact_match": 21,
"exact_match_rate": 0.105,
"truncated": 0
},
"open": {
"questions": 200,
"numeric_format": 197,
"valid_completions": 197,
"exact_match": 168,
"exact_match_rate": 0.84,
"truncated": 0
}
}
},
"replay": {
"nll": {
"domain": {
"documents": 600,
"active_tokens": 25275,
"nll_nats_per_token": 0.4175135541870015,
"delta_from_base": -4.7800466216649
},
"control": {
"documents": 150,
"active_tokens": 3840,
"nll_nats_per_token": 1.255818447340746,
"delta_from_base": -3.9376274685588033
}
},
"answers": {
"closed": {
"questions": 200,
"numeric_format": 200,
"valid_completions": 200,
"exact_match": 21,
"exact_match_rate": 0.105,
"truncated": 0
},
"open": {
"questions": 200,
"numeric_format": 200,
"valid_completions": 200,
"exact_match": 178,
"exact_match_rate": 0.89,
"truncated": 0
}
}
}
},
"quality_gain_required": false,
"scope": "Data-only per-token sampler replay, tokenizer decoding, parameter manifests, resume evidence, all TF-IDF/cosines, CE probes, raw answers and revision metrics; no full BPE re-encoding or scalar Transformer reconstruction"
}
