The fictional AtelierDesk support service must answer with case-file facts and request missing information instead of promising an invented deadline. For one case file, two answers are labelled preferred/rejected by public simulation rules.
Compare the policy after local SFT, after additional SFT on chosen answers, and after DPO. Show policy and reference log-probabilities for every pair, then free answers on new case files.
Download the complete project, revision r001
From case file to preference
A preference compares two answers to the same case file. It is neither an absolute grade nor proof of human satisfaction. AtelierDeskPrefs-v1 rules are public and simulated: respect recorded replacement status and shipping date; ask for the order number only if it is missing. An unknown date remains unknown. Polite wording does not make an invented promise true.
300 families provide four pairs each: 840 train, 180 validation, and 180 test, from disjoint families 0–209, 210–254, and 255–299. One additional demonstration per training family provides 210 examples for SFT warm-up. Case files and identifiers are distinct, but elementary facts and language templates remain shared. This does not measure generalisation to real customer services or unknown intents.
Two variants change status; one changes the date; one reverses whether the order number is requested. Courtesy forms alternate: chosen is shorter, equal in word length, or longer. An audit publishes presence of every word in both labels, a binary classifier per word, and selection by length with half credit for ties. This audit does not prove every learned shortcut is absent. Evaluation strata use actual token lengths, which may differ from word counts. chosen/rejected metadata and preference reason are never included in generation case files.
The policy produces conditional probabilities
Qwen2.5-0.5B-Instruct, revision 7ae557604adf67be50417f59c2c2f167def9a775, is loaded strictly: 24 blocks, width 896, 14 Q heads, two K/V heads, MLP 4864, vocabulary 151936, and tied embeddings/output. All 494032768 unique parameters are trainable in FP32. The profile uses eager attention, TF32 disabled, and a one-answer microbatch; no quantisation, LoRA, or automatic multi-GPU distribution.
The tokenizer applies its template to the system message and case file, then adds the assistant boundary. The answer is encoded separately and receives exactly one final EOS=151645. The contract is explicit: it does not claim to reproduce retokenisation of the concatenated string. Maximums are 256 prompt tokens, 128 answer tokens with EOS, and a total below 512; any excess fails without truncation.
To predict the token at position t, the logit at t−1 is used. Code aligns logits[:, :-1] with labels[:, 1:]. Prompt and padding labels are −100; answer targets, including EOS, remain active. PAD=151643 is masked by position. The prompt has no CE of its own, but its tokens condition answer probabilities: ignoring its labels does not remove its influence.
For an answer y, s(x,y) is the SUM of log-probabilities of active tokens. A long answer accumulates more generally negative terms. The report also publishes per-token means to diagnose length bias; these means never replace sums in DPO.
Local SFT then frozen reference
Warm-up learns the 210 training demonstrations for two epochs, accumulating eight examples: 27 updates per epoch, the last carrying two examples, hence 54 total. CE is normalised by the total number of answer tokens in the update; an average of averages would weight short answers differently. AdamW uses lr=0.00002 and weight_decay=0.01, clipping at 1, five-step warm-up, then linear decay to 10% of the maximum rate.
The final SFT checkpoint is saved in Safetensors and reloaded strictly. The reference is a frozen eval, no_grad instance. It never enters the optimiser. Its hash and parameter digest remain unchanged. Each preference policy starts from an independent reload of the same SFT; it shares no mutable object with the reference.
For every pair, cache retains the two actually tokenised answers, masks, per-token scores, and sums. Its key binds checkpoint, tokenizer and tokenizer configuration, template and system text, scoring code, model lock, complete IDs, and labels. An existing cache whose identity differs is refused before use. Only train and validation files are mounted in learning; test data and generation outputs are inaccessible to it.
The costs of SFT, reference cache, and each optimisation are measured separately. The cache avoids loading a second reference during every DPO backward pass; its initial cost is not free.
The coded loss and its gradient
For a pair, four numbers are involved: sθ+ and sθ− for policy, sref+ and sref− for reference. The margin is m = β[(sθ+ − sθ−) − (sref+ − sref−)]. Loss is −logsigmoid(m), averaged over pairs in the update. Code uses logsigmoid for numerical stability. The reference supplies constants without gradients.
If policy and reference are equal, margin is zero and loss is log(2). Yet for two different answers, policy-score derivatives are opposite and non-zero: −β/2 for sθ+ and +β/2 for sθ− on one pair. The model can therefore learn. If both answers have exactly the same tokens and masks, the terms cancel and give no signal; these pairs are rejected during construction.
FP64 values and gradients are compared with the DPO authors' preference_loss function, extracted from commit f8b8c0f49dc92a430bae41585f9d467d3618fe2f, with label_smoothing=0, ipo=False, and reference_free=False. Apache-2.0 licence and provenance are provided. Only this function and necessary imports are used; the original trainer and its global settings are not run.
A comparison defined before results
After SFT, four policies each receive 840 prompts for two epochs in the same deterministic order: 210 updates and 1680 presentations. Chosen-only learns only the chosen answer with token-normalised CE. DPO receives chosen and rejected, with β=0.05, 0.1, or 0.2. All four branches use AdamW lr=0.000001, weight_decay=0.01, clipping at 1, 21-step warm-up, then decay to 10%.
Prompt presentations are equal; scored tokens, forwards, and computation are not. DPO averages loss by pair and chosen-only by answer token. Every beta and final checkpoint is reported: no selection of the best beta or checkpoint uses test data. Validation after every epoch provides chosen NLL and, for preference branches, margins and preference accuracy. It does not stop training.
Every phase, including SFT, runs with interruption after one epoch, a continuous two-epoch run, and resumption for the second epoch. Complete model, AdamW, sampler, scheduler, Python/NumPy/PyTorch CPU/CUDA RNG, and final histories must be exactly equal. This contract targets an epoch boundary, not a stop in the middle of backward.
The fifteen local states are written losslessly in gzip level 1, to a temporary file then renamed. Uncompressed serialised bytes and actual written size are measured. This choice promises neither compression rate nor VRAM reduction. Torch states are trusted objects produced locally; do not replace them with an untrusted download. The five final policies are also retained in Safetensors. Existing artefacts are not overwritten.
Offline preference and free answers
The five SFT/chosen-only/DPO×3 policies score the same 180 test pairs. Accuracy counts s+ > s−, zero for reverse order, and half a point for equality. Reference-corrected margins and per-token means are separate. Length-stratified reports prevent a failure on longer chosen answers from being hidden behind a general mean.
Each policy also generates 180 answers without chosen/rejected in context: case file alone with the same instruction, greedy decoding, one beam, at most 128 new tokens, EOS stop. All IDs, raw texts, stop reasons, and durations remain available. A response stopped at the limit fails even if early sentences seem correct. No judge model, customer send, or business action is used.
A parser independent from generation recognises the controlled dataset clauses: status, known/unknown date, and request or acknowledgement of the number. It requires coverage of ALL text, three distinct fields, and EOS. An unknown sentence, duplicate, Markdown tag, or refusal makes the answer incomplete. A correct natural sentence outside the grammar may therefore fail: this narrow score is not a universal measure of semantic faithfulness.
Correctness requires all three correct facts/actions across 180 cases, without excluding invalid outputs. Useful requests are reported where the number is missing. Invented dates count recognised dates when the date is unknown; unrecognised sentences are reported separately and never certified as non-invented. Counted refusals follow explicit expressions; this detector does not identify every possible refusal. A higher DPO margin therefore does not prove higher free-answer quality, still less customer satisfaction.
Exercise: an old file guarantees nothing
Across the first twelve test case files, support-v2 adds a concision instruction. Conditioning IDs must genuinely change. The old cache is rejected before its scores are read. A frozen reference recalculates a new cache and, separately, the same scores without cache; they must be exactly equal. The dpo-010 policy scores new pairs and both reference paths give the same DPO loss. No additional training occurs. Weight, cache, and main-result hashes remain unchanged.
One seed, simulated preferences, and controlled language limit this practical exercise's scope. Beta does not correct contradictory annotations or a poorly designed corpus. No gain is required: a regression or invalid generation must be retained and explained.
Build informative pairs without shortcuts
Preferences and length bias
Frequent error: A generation marker reveals chosen: remove metadata from texts and test the shortcut.
Invariant: 840/180/180 pairs and 210 training demonstrations, with reasons and lengths audited.
File src/dpo_preferences/data.py
import hashlib, re
from pathlib import Path
from .config import save_json, sha
STATUSES = ["approved", "pending", "declined"]
COURTESY = ["Thank you for your message.", "I hope this explanation is clear."]
def split_pair_families(family):
return "train" if family < 210 else "validation" if family < 255 else "test"
def facts_for(family):
return {
"replacement": STATUSES[family % 3],
"shipping_date": None if family % 2 == 0 else f"2026-11-{(family//2)%27+1:02d}",
"order_reference": None if family % 4 < 2 else f"OD{family+1000}",
}
def dossier(family, variant, facts):
identifier = f"D{family:03d}-{variant}"
date = facts["shipping_date"] or "unknown"
order = facts["order_reference"] or "missing"
return f"Ticket {identifier}. Replacement status: {facts['replacement']}. Shipping date: {date}. Order reference: {order}."
def reply(record, j, request_order=None, courtesy=False):
status = record["replacement"]
date = record["shipping_date"]
ask = record["order_reference"] is None if request_order is None else request_order
status_clause = [
f"Your replacement is {status}.",
f"The replacement is {status}.",
f"Replacement status: {status}.",
f"I can confirm that the replacement is {status}.",
][j % 4]
date_clause = (
f"The shipping date is {date}."
if date
else (
"I cannot confirm a shipping date."
if j % 2
else "The shipping date is unknown."
)
)
next_clause = (
"Please provide your order number."
if ask
else "Your order number is already recorded."
)
sentences = [status_clause, date_clause, next_clause]
if courtesy:
sentences = COURTESY[:1] + sentences + COURTESY[1:]
return " ".join(sentences)
def audit_preference_shortcuts(labels):
n = len(labels)
vocabulary = sorted(
{
word
for row in labels
for side in ["chosen", "rejected"]
for word in re.findall(r"[a-z]+", row[side].lower())
}
)
features = []
lengths = {"chosen_shorter": 0, "same_words": 0, "chosen_longer": 0}
for word in vocabulary:
a = sum(word in set(re.findall(r"[a-z]+", r["chosen"].lower())) for r in labels)
b = sum(
word in set(re.findall(r"[a-z]+", r["rejected"].lower())) for r in labels
)
accuracy = (a + n - b) / (2 * n)
features.append(
{
"word": word,
"chosen_presence": a,
"rejected_presence": b,
"best_binary_label_accuracy": max(accuracy, 1 - accuracy),
}
)
for row in labels:
a = len(row["chosen"].split())
b = len(row["rejected"].split())
lengths[
"chosen_shorter" if a < b else "chosen_longer" if a > b else "same_words"
] += 1
long = (lengths["chosen_longer"] + 0.5 * lengths["same_words"]) / n
return {
"pairs": n,
"word_length_regimes": lengths,
"best_longer_or_shorter_preference_accuracy_with_half_ties": max(
long, 1 - long
),
"word_presence_features": features,
"maximum_single_word_binary_label_accuracy": max(
r["best_binary_label_accuracy"] for r in features
),
"scope": "Original words and word counts, not every learned lexical/semantic shortcut or BPE length",
}
def build_support_pairs():
pages = []
labels = []
warm_pages = []
warm_labels = []
for family in range(300):
facts = facts_for(family)
split = split_pair_families(family)
for j in range(4):
identifier = f"D{family:03d}-{j}"
text = dossier(family, j, facts)
chosen = reply(facts, j, courtesy=j == 1)
wrong = dict(facts)
reason = [
"replacement_status",
"shipping_date",
"missing_order_action",
"replacement_status",
][j]
if reason == "replacement_status":
wrong["replacement"] = STATUSES[(family + 1) % 3]
if reason == "shipping_date":
wrong["shipping_date"] = (
f"2026-11-{(family//2)%27+1:02d}"
if facts["shipping_date"] is None
else None
)
rejected = reply(
wrong,
j,
request_order=(
(facts["order_reference"] is not None)
if reason == "missing_order_action"
else None
),
courtesy=j == 2,
)
if chosen == rejected:
raise ValueError("Identical preference pair has no information")
pages.append(
{
"document_id": identifier,
"family": family,
"formulation": j,
"split": split,
"dossier": text,
"record": facts,
"content_sha256": hashlib.sha256(text.encode()).hexdigest(),
}
)
labels.append(
{
"document_id": identifier,
"chosen": chosen,
"rejected": rejected,
"preference_reason": reason,
"expected": {
"replacement": facts["replacement"],
"shipping_date": facts["shipping_date"],
"request_order": facts["order_reference"] is None,
},
"simulated_preference": True,
}
)
if split == "train":
identifier = f"D{family:03d}-warm"
text = dossier(family, "warm", facts)
warm_pages.append(
{
"document_id": identifier,
"family": family,
"split": "train",
"dossier": text,
"record": facts,
"content_sha256": hashlib.sha256(text.encode()).hexdigest(),
}
)
warm_labels.append(
{
"document_id": identifier,
"response": reply(facts, family % 4, courtesy=family % 2 == 0),
}
)
if (
len({p["document_id"] for p in pages + warm_pages}) != 1410
or len({p["dossier"] for p in pages + warm_pages}) != 1410
):
raise ValueError("Duplicate ticket or ID")
return pages, labels, warm_pages, warm_labels
def prepare(config):
root = Path("data/generated")
if (root / "manifest.json").exists():
raise ValueError("Preserve existing corpus; use a fresh extraction")
pages, labels, warm_pages, warm_labels = build_support_pairs()
lookup = {r["document_id"]: r for r in labels}
save_json(root / "input/pages.json", pages)
save_json(root / "labels/gold.json", labels)
save_json(root / "warmstart/pages.json", warm_pages)
save_json(root / "labels/warmstart.json", warm_labels)
counts = {}
audits = {}
for split in ["train", "validation", "test"]:
selected = [p for p in pages if p["split"] == split]
gold = [lookup[p["document_id"]] for p in selected]
counts[split] = {
"pairs": len(selected),
"families": len({p["family"] for p in selected}),
}
audits[split] = audit_preference_shortcuts(gold)
if split != "test":
save_json(root / f"{split}/pages.json", selected)
save_json(root / f"labels/{split}.json", gold)
save_json(
"outputs/data-audit.json",
{
"dataset": "AtelierDeskPrefs-v1",
"rights": "CC0-1.0",
"counts": counts,
"warmstart_demonstrations": 210,
"warmstart_train_families_only": True,
"identical_pairs": 0,
"shortcuts": audits,
"human_preferences": False,
"shared_semantic_and_language_primitives": True,
},
)
manifest = {
"dataset": "AtelierDeskPrefs-v1",
"seed": config["seed"],
"counts": counts,
"warmstart": 210,
"files": {
str(f.relative_to(root)): sha(f) for f in root.rglob("*") if f.is_file()
},
}
save_json(root / "manifest.json", manifest)
return counts
Score answers exactly
Sum of log-probabilities and masks
Frequent error: A duplicated EOS artificially favours an answer: inspect the decoded sequence and its IDs.
Invariant: Scores identical to token-by-token calculation and PAD-invariant.
File src/dpo_preferences/scores.py
import hashlib, json
import torch
from .config import sha, save_json
SYSTEM = "Reply to the customer using only the recorded replacement status and shipping date. If the order reference is missing, ask for the order number; otherwise acknowledge that it is already recorded. State each fact clearly. Do not invent a date or change a decision."
def system_text(version):
if version == "support-v1":
return SYSTEM
if version == "support-v2":
return SYSTEM + " Keep your answer concise."
raise ValueError("Unknown explicit template version")
def format_prompt(tokenizer, dossier, version):
messages = [
{"role": "system", "content": system_text(version)},
{"role": "user", "content": dossier},
]
return messages, tokenizer.apply_chat_template(
messages, tokenize=True, add_generation_prompt=True
)
def completion_row(tokenizer, dossier, response, config, identifier, version=None):
version = version or config["template_version"]
messages, prompt = format_prompt(tokenizer, dossier, version)
target = tokenizer.encode(response, add_special_tokens=False) + [151645]
if (
not prompt
or len(prompt) > config["max_prompt_tokens"]
or not 1 < len(target) <= config["max_response_tokens_including_eos"]
or len(prompt) + len(target) > config["context"]
):
raise ValueError(
identifier + ": declared token budget exceeded; do not truncate"
)
return {
"document_id": identifier,
"messages": messages,
"input_ids": prompt + target,
"labels": [-100] * len(prompt) + target,
"prompt_tokens": len(prompt),
"response_tokens": len(target),
"template_version": version,
}
def paired_completion_batch(tokenizer, page, label, config, version=None):
chosen = completion_row(
tokenizer,
page["dossier"],
label["chosen"],
config,
page["document_id"],
version,
)
rejected = completion_row(
tokenizer,
page["dossier"],
label["rejected"],
config,
page["document_id"],
version,
)
if (
chosen["input_ids"] == rejected["input_ids"]
and chosen["labels"] == rejected["labels"]
):
raise ValueError("Identical completions are not an informative preference")
return {
"document_id": page["document_id"],
"chosen": chosen,
"rejected": rejected,
"response_length_delta": chosen["response_tokens"]
- rejected["response_tokens"],
}
def collate(rows, device="cuda", pad_to=None):
if not rows:
raise ValueError("Empty completion batch")
width = max(len(r["input_ids"]) for r in rows) if pad_to is None else pad_to
ids = torch.full((len(rows), width), 151643, dtype=torch.long, device=device)
mask = torch.zeros_like(ids, dtype=torch.bool)
labels = torch.full_like(ids, -100)
for i, row in enumerate(rows):
n = len(row["input_ids"])
if n > width:
raise ValueError("Padding width would truncate a response")
ids[i, :n] = torch.tensor(row["input_ids"], device=device)
mask[i, :n] = True
labels[i, :n] = torch.tensor(row["labels"], device=device)
return ids, mask, labels
def sequence_logprob(logits, labels):
if logits.shape[:2] != labels.shape:
raise ValueError("Logits and labels shape mismatch")
active = labels[:, 1:] != -100
counts = active.sum(dim=1)
if (counts == 0).any():
raise ValueError("Every sequence needs an active response target")
selected = logits[:, :-1][active]
targets = labels[:, 1:][active]
values = (
torch.log_softmax(selected, dim=-1)
.gather(-1, targets.unsqueeze(-1))
.squeeze(-1)
)
sums = torch.stack([v.sum() for v in values.split(counts.tolist())])
return sums, counts, values, selected, targets
def model_scores(model, rows):
ids, mask, labels = collate(rows)
logits = model(
input_ids=ids, attention_mask=mask, use_cache=False, return_dict=True
).logits
return sequence_logprob(logits.float(), labels)
def reference_identity(template_version="support-v1"):
return {
"checkpoint_sha256": sha("assets/reference/sft.safetensors"),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"tokenizer_config_sha256": sha("assets/model/tokenizer_config.json"),
"template_version": template_version,
"system_text": system_text(template_version),
"scores_implementation_sha256": sha("src/dpo_preferences/scores.py"),
"model_lock_sha256": sha("locks/models.lock.json"),
"eos": 151645,
"pad": 151643,
"reduction": "sum-response-logprob-including-EOS",
}
def reference_cache_key(identity, pair):
payload = {"identity": identity, "pair": pair}
return hashlib.sha256(
json.dumps(
payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")
).encode()
).hexdigest()
def read_cached_pair(cache, identity, pair):
if cache["identity"] != identity:
raise ValueError(
"STALE_REFERENCE_CACHE: model/tokenizer/template/code identity differs"
)
entry = cache["rows"].get(pair["document_id"])
if (
entry is None
or entry["key"] != reference_cache_key(identity, pair)
or entry["pair"] != pair
):
raise ValueError("STALE_REFERENCE_CACHE: prompt/completion IDs or masks differ")
return entry
def dpo_sigmoid_loss(
policy_chosen, policy_rejected, reference_chosen, reference_rejected, beta
):
if beta <= 0 or any(
t.shape != policy_chosen.shape
for t in [policy_rejected, reference_chosen, reference_rejected]
):
raise ValueError("Positive beta and matching score shapes required")
if reference_chosen.requires_grad or reference_rejected.requires_grad:
raise ValueError("Reference scores must be fixed")
margin = beta * (
(policy_chosen - policy_rejected) - (reference_chosen - reference_rejected)
)
return -torch.nn.functional.logsigmoid(margin), margin
def inspect_batch(config):
from .model import load_tokenizer
from .reference_loss import preference_loss
from .data import build_support_pairs
tokenizer = load_tokenizer()
pages, labels, _, _ = build_support_pairs()
pairs = [
paired_completion_batch(tokenizer, pages[i], labels[i], config)
for i in [0, 1, 2, 3]
]
a = torch.tensor([-4.0, -12.0, -8.0], dtype=torch.float64, requires_grad=True)
b = torch.tensor([-5.0, -3.0, -6.0], dtype=torch.float64, requires_grad=True)
ra = torch.tensor([-4.0, -10.0, -9.0], dtype=torch.float64)
rb = torch.tensor([-5.0, -5.0, -7.0], dtype=torch.float64)
ours, margin = dpo_sigmoid_loss(a, b, ra, rb, 0.1)
grad = torch.autograd.grad(ours.mean(), (a, b), retain_graph=True)
ref = preference_loss(
a, b, ra, rb, 0.1, label_smoothing=0.0, ipo=False, reference_free=False
)[0]
other_grad = torch.autograd.grad(ref.mean(), (a, b))
if not torch.allclose(ours, ref, atol=1e-12, rtol=0) or not all(
torch.allclose(x, y, atol=1e-12, rtol=0) for x, y in zip(grad, other_grad)
):
raise ValueError("Original DPO value/gradient comparison failed")
save_json(
"outputs/mechanism.json",
{
"pairs": pairs,
"illustrative_scores": {
"policy_chosen": a.tolist(),
"policy_rejected": b.tolist(),
"reference_chosen": ra.tolist(),
"reference_rejected": rb.tolist(),
"beta": 0.1,
"margins": margin.tolist(),
"losses": ours.tolist(),
"reference_losses": ref.tolist(),
"chosen_gradient": grad[0].tolist(),
"rejected_gradient": grad[1].tolist(),
"reference_chosen_gradient": other_grad[0].tolist(),
"reference_rejected_gradient": other_grad[1].tolist(),
"atol": 1e-12,
"dtype": "float64",
},
"sums_not_length_means": True,
"reference_no_grad": True,
},
)
Warm up the policy and freeze the reference
Local SFT and common initialisation
Frequent error: Reference points to the same mutable object as policy: really copy weights and test independence.
Invariant: policy/ref hashes are equal at the start and ref has no gradient.
File src/dpo_preferences/warmstart.py
import time
from pathlib import Path
import torch
from .config import read_json, save_json, sha, seed_all
from .model import load_tokenizer, reload_learned, parameter_manifest, manifest_digest
from .scores import (
paired_completion_batch,
model_scores,
reference_identity,
reference_cache_key,
)
def freeze_reference():
model, metadata = reload_learned("assets/reference/sft.safetensors")
for parameter in model.parameters():
parameter.requires_grad_(False)
model = model.cuda().eval()
if any(p.requires_grad or p.grad is not None for p in model.parameters()):
raise ValueError("Reference must not have trainable parameters or gradients")
if manifest_digest(parameter_manifest(model)) != metadata["model_digest"]:
raise ValueError("Reference differs from local SFT")
return model, metadata
@torch.no_grad()
def score_pair(model, pair):
result = {}
for side in ["chosen", "rejected"]:
sums, counts, values, _, targets = model_scores(model, [pair[side]])
result[side] = {
"sum_logprob": float(sums[0]),
"mean_logprob": float(sums[0] / counts[0]),
"response_tokens": int(counts[0]),
"token_logprobs": values.cpu().tolist(),
"targets": targets.cpu().tolist(),
}
return result
@torch.no_grad()
def cache_reference_scores(config):
seed_all(config["seed"])
model, metadata = freeze_reference()
tokenizer = load_tokenizer()
pages = read_json("data/generated/input/pages.json")
gold = {r["document_id"]: r for r in read_json("data/generated/labels/gold.json")}
identity = reference_identity()
before = sha("assets/reference/sft.safetensors")
reports = {}
torch.cuda.reset_peak_memory_stats()
for split in ["train", "validation", "test"]:
path = Path(f"outputs/reference-cache/{split}.json")
if path.exists():
raise ValueError("Never overwrite a reference cache")
selected = [p for p in pages if p["split"] == split]
rows = {}
torch.cuda.reset_peak_memory_stats()
torch.cuda.synchronize()
begun = time.perf_counter()
for i, page in enumerate(selected):
pair = paired_completion_batch(
tokenizer, page, gold[page["document_id"]], config
)
rows[page["document_id"]] = {
"key": reference_cache_key(identity, pair),
"pair": pair,
"scores": score_pair(model, pair),
}
if (i + 1) % 120 == 0:
print({"reference_cache": split, "pairs": i + 1}, flush=True)
torch.cuda.synchronize()
seconds = time.perf_counter() - begun
payload = {
"identity": identity,
"reference_model_digest": metadata["model_digest"],
"split": split,
"rows": rows,
"pairs": len(selected),
"seconds": seconds,
"response_tokens": sum(
r["scores"][side]["response_tokens"]
for r in rows.values()
for side in ["chosen", "rejected"]
),
"max_torch_allocated_bytes": torch.cuda.max_memory_allocated(),
"reference_eval": not model.training,
"reference_parameters_require_grad": sum(
p.requires_grad for p in model.parameters()
),
"reference_gradients_present": sum(
p.grad is not None for p in model.parameters()
),
}
save_json(path, payload)
reports[split] = {
"sha256": sha(path),
"pairs": len(selected),
"seconds": seconds,
"response_tokens": payload["response_tokens"],
}
after = sha("assets/reference/sft.safetensors")
if (
before != after
or manifest_digest(parameter_manifest(model)) != metadata["model_digest"]
):
raise ValueError("Reference mutated during cache creation")
save_json(
"outputs/reference-cache/audit.json",
{
"identity": identity,
"reference_model_digest": metadata["model_digest"],
"reference_sha256_before": before,
"reference_sha256_after": after,
"immutable": True,
"splits": reports,
},
)
Code DPO and the chosen-only baseline
Relative margin, sigmoid, and optimisation
Frequent error: Replacing sums with means changes the objective: numerical test on unequal lengths.
Invariant: Gradients are finite, ref immutable, and exposure budgets captured.
File src/dpo_preferences/train.py
import math, time
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,
reload_learned,
parameter_manifest,
manifest_digest,
save_learned,
)
from .scores import (
completion_row,
paired_completion_batch,
model_scores,
dpo_sigmoid_loss,
reference_identity,
read_cached_pair,
)
from .state import equal, save_resume_state, load_resume_state
BRANCH_BETA = {"dpo-005": 0.05, "dpo-010": 0.1, "dpo-020": 0.2}
def learning_rate(config, branch, step, total):
peak = (
config["warmstart_learning_rate"]
if branch == "warmstart"
else config["learning_rate"]
)
warmup = max(1, int(total * config["warmup_ratio"]))
if step < warmup:
return peak * (step + 1) / warmup
return peak * (
1
- min(1.0, (step - warmup) / max(1, total - warmup - 1))
* (1 - config["minimum_lr_ratio"])
)
def encoded_rows(config, tokenizer, branch):
split = "warmstart" if branch == "warmstart" else "train"
pages = read_json(f"data/generated/{split}/pages.json")
labels = {
r["document_id"]: r for r in read_json(f"data/generated/labels/{split}.json")
}
if any(p["split"] != "train" for p in pages):
raise ValueError("Training examples must belong to train families")
if branch == "warmstart":
return [
completion_row(
tokenizer,
p["dossier"],
labels[p["document_id"]]["response"],
config,
p["document_id"],
)
for p in pages
]
return [
paired_completion_batch(tokenizer, p, labels[p["document_id"]], config)
for p in pages
]
def validation_rows(config, tokenizer):
pages = read_json("data/generated/validation/pages.json")
labels = {
r["document_id"]: r for r in read_json("data/generated/labels/validation.json")
}
if any(p["split"] != "validation" for p in pages):
raise ValueError("Validation partition mismatch")
return [
paired_completion_batch(tokenizer, p, labels[p["document_id"]], config)
for p in pages
]
@torch.no_grad()
def validate_policy(model, pairs, branch, cache=None, identity=None):
model.eval()
chosen_losses = 0.0
chosen_tokens = 0
raw_wins = relative_wins = 0.0
details = []
for pair in pairs:
sums, counts, _, _, _ = model_scores(model, [pair["chosen"]])
a = float(sums[0])
chosen_losses -= a
chosen_tokens += int(counts[0])
if branch != "warmstart":
rejected = model_scores(model, [pair["rejected"]])[0]
b = float(rejected[0])
ref = read_cached_pair(cache, identity, pair)["scores"]
raw = a - b
relative = raw - (
ref["chosen"]["sum_logprob"] - ref["rejected"]["sum_logprob"]
)
raw_wins += 1 if raw > 0 else 0.5 if raw == 0 else 0
relative_wins += 1 if relative > 0 else 0.5 if relative == 0 else 0
details.append(
{
"document_id": pair["document_id"],
"policy_chosen_sum": a,
"policy_rejected_sum": b,
"reference_chosen_sum": ref["chosen"]["sum_logprob"],
"reference_rejected_sum": ref["rejected"]["sum_logprob"],
"raw_margin": raw,
"reference_adjusted_margin": relative,
}
)
model.train()
return {
"pairs": len(pairs),
"chosen_response_tokens": chosen_tokens,
"chosen_nll_nats_per_token": chosen_losses / chosen_tokens,
"raw_preference_accuracy_half_ties": raw_wins / len(pairs) if details else None,
"reference_adjusted_accuracy_half_ties": (
relative_wins / len(pairs) if details else None
),
"rows": details,
"used_for_checkpoint_selection": False,
}
def train_policy(config, branch="chosen", mode="continuous"):
if branch not in ["warmstart", "chosen", *BRANCH_BETA] or mode not in [
"interrupt",
"continuous",
"resume",
]:
raise ValueError("Unknown branch or mode")
if (
Path("data/generated/input").exists()
or Path("data/generated/labels/gold.json").exists()
or Path("outputs/reference-cache/test.json").exists()
):
raise ValueError(
"Learning service must not mount heldout inputs, gold or test cache"
)
seed_all(config["seed"])
tokenizer = load_tokenizer()
rows = encoded_rows(config, tokenizer, branch)
validation = validation_rows(config, tokenizer)
if len(rows) != (210 if branch == "warmstart" else 840) or len(validation) != 180:
raise ValueError("Frozen training population differs")
model = (
load_base()
if branch == "warmstart"
else reload_learned("assets/reference/sft.safetensors")[0]
)
for parameter in model.parameters():
parameter.requires_grad_(True)
model = model.cuda()
initial = parameter_manifest(model)
initial_digest = manifest_digest(initial)
parameters = list(model.parameters())
if sum(p.numel() for p in parameters) != 494032768 or len(
{id(p) for p in parameters}
) != len(parameters):
raise ValueError("Full unique trainable base expected")
ref_identity = reference_identity() if branch != "warmstart" else None
cache = (
read_json("outputs/reference-cache/train.json")
if branch != "warmstart"
else None
)
val_cache = (
read_json("outputs/reference-cache/validation.json")
if branch != "warmstart"
else None
)
if cache:
for row in rows:
read_cached_pair(cache, ref_identity, row)
for row in validation:
read_cached_pair(val_cache, ref_identity, row)
if cache["reference_model_digest"] != initial_digest:
raise ValueError(
"Policy and fixed reference did not start from the same SFT"
)
training_split = "warmstart" if branch == "warmstart" else "train"
identity = {
"branch": branch,
"beta": BRANCH_BETA.get(branch),
"config": config,
"initial_model_digest": initial_digest,
"base_sha256": sha("assets/model/model.safetensors"),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"initial_sft_sha256": (
sha("assets/reference/sft.safetensors") if branch != "warmstart" else None
),
"reference_identity": ref_identity,
"reference_train_cache_sha256": (
sha("outputs/reference-cache/train.json") if cache else None
),
"reference_validation_cache_sha256": (
sha("outputs/reference-cache/validation.json") if cache else None
),
"implementation_sha256": {
name: sha("src/dpo_preferences/" + name)
for name in ["model.py", "scores.py", "state.py", "train.py"]
},
"data_sha256": {
name: sha("data/generated/" + name)
for name in [
f"{training_split}/pages.json",
f"labels/{training_split}.json",
"validation/pages.json",
"labels/validation.json",
]
},
}
optimizer = torch.optim.AdamW(
parameters, lr=config["learning_rate"], weight_decay=config["weight_decay"]
)
sampler = torch.Generator().manual_seed(config["seed"] + 100)
epochs = config["warmstart_epochs"] if branch == "warmstart" else config["epochs"]
per_epoch = math.ceil(len(rows) / config["accumulation_examples"])
total = epochs * per_epoch
root = Path("assets/checkpoints")
destination = root / f"{branch}-{mode}.pt.gz"
history = []
validation_history = []
start = updates = 0
if destination.exists():
raise ValueError("Preserve previous training checkpoint")
if mode == "resume":
state = load_resume_state(root / f"{branch}-interrupt.pt.gz")
if state["identity"] != identity:
raise ValueError("Resume model/tokenizer/reference/data identity changed")
model.load_state_dict(state["model"], strict=True)
optimizer.load_state_dict(state["optimizer"])
sampler.set_state(state["sampler"])
restore_rng(state["rng"])
start = state["epoch"]
updates = state["updates"]
history = state["history"]
validation_history = state["validation_history"]
if state["scheduler"] != {
"next_update": updates,
"next_lr": learning_rate(config, branch, updates, total),
}:
raise ValueError("Resume scheduler changed")
del state
stop = config["interrupt_epoch"] if mode == "interrupt" else epochs
model.train()
torch.cuda.reset_peak_memory_stats()
torch.cuda.synchronize()
begun = time.perf_counter()
optimization_seconds = 0.0
for epoch in range(start, stop):
permutation = torch.randperm(len(rows), generator=sampler).tolist()
torch.cuda.synchronize()
epoch_start = time.perf_counter()
for offset in range(0, len(rows), config["accumulation_examples"]):
selected = permutation[offset : offset + config["accumulation_examples"]]
lr = learning_rate(config, branch, updates, total)
denominator = (
len(selected)
if branch in BRANCH_BETA
else sum(
(rows[i] if branch == "warmstart" else rows[i]["chosen"])[
"response_tokens"
]
for i in selected
)
)
objective_sum = 0.0
pair_details = []
tokens = 0
for group in optimizer.param_groups:
group["lr"] = lr
optimizer.zero_grad(set_to_none=True)
for i in selected:
row = rows[i]
if branch in BRANCH_BETA:
a, na, _, _, _ = model_scores(model, [row["chosen"]])
b, nb, _, _, _ = model_scores(model, [row["rejected"]])
reference = read_cached_pair(cache, ref_identity, row)["scores"]
ra = torch.tensor(
[reference["chosen"]["sum_logprob"]],
device="cuda",
dtype=torch.float32,
)
rb = torch.tensor(
[reference["rejected"]["sum_logprob"]],
device="cuda",
dtype=torch.float32,
)
losses, margins = dpo_sigmoid_loss(
a, b, ra, rb, BRANCH_BETA[branch]
)
loss = losses.sum()
tokens += int(na[0] + nb[0])
pair_details.append(
{
"document_id": row["document_id"],
"policy_chosen_sum": float(a.detach()[0]),
"policy_rejected_sum": float(b.detach()[0]),
"reference_chosen_sum": float(ra[0]),
"reference_rejected_sum": float(rb[0]),
"chosen_tokens": int(na[0]),
"rejected_tokens": int(nb[0]),
"margin": float(margins.detach()[0]),
"loss": float(loss.detach()),
}
)
else:
completion = row if branch == "warmstart" else row["chosen"]
sums, counts, _, _, _ = model_scores(model, [completion])
loss = -sums.sum()
tokens += int(counts[0])
if not torch.isfinite(loss):
raise ValueError("Nonfinite learning objective")
(loss / denominator).backward()
objective_sum += float(loss.detach())
norm = torch.nn.utils.clip_grad_norm_(
parameters, config["gradient_clip"], error_if_nonfinite=True
)
if any(
p.grad is None
or p.dtype != torch.float32
or p.grad.dtype != torch.float32
for p in parameters
):
raise ValueError("Full FP32 policy gradients expected")
optimizer.step()
updates += 1
history.append(
{
"update": updates,
"epoch": epoch + 1,
"document_ids": [rows[i]["document_id"] for i in selected],
"objective_denominator": denominator,
"objective_reduction": (
"mean_pairs"
if branch in BRANCH_BETA
else "mean_response_tokens"
),
"policy_response_tokens": tokens,
"objective": objective_sum / denominator,
"learning_rate": lr,
"gradient_norm_before_clip": float(norm),
"pairs": pair_details,
}
)
if updates % 21 == 0:
print(
{k: v for k, v in history[-1].items() if k != "pairs"}, flush=True
)
torch.cuda.synchronize()
optimization_seconds += time.perf_counter() - epoch_start
observed = validate_policy(model, validation, branch, val_cache, ref_identity)
validation_history.append({"epoch": epoch + 1, **observed})
print(
{
"branch": branch,
"epoch": epoch + 1,
**{k: v for k, v in observed.items() if k != "rows"},
},
flush=True,
)
torch.cuda.synchronize()
elapsed = time.perf_counter() - begun
peak = torch.cuda.max_memory_allocated()
final = parameter_manifest(model)
final_digest = manifest_digest(final)
if final_digest == initial_digest:
raise ValueError("No actual learned weight change")
if (
ref_identity
and sha("assets/reference/sft.safetensors") != ref_identity["checkpoint_sha256"]
):
raise ValueError("Fixed SFT reference changed")
state = {
"identity": identity,
"epoch": stop,
"updates": updates,
"model": model.state_dict(),
"optimizer": optimizer.state_dict(),
"sampler": sampler.get_state(),
"scheduler": {
"next_update": updates,
"next_lr": learning_rate(config, branch, updates, total),
},
"rng": rng_state(),
"history": history,
"validation_history": validation_history,
}
saved = save_resume_state(destination, state)
if mode == "resume":
save_learned(
model,
(
"assets/reference/sft.safetensors"
if branch == "warmstart"
else root / f"{branch}-resume.safetensors"
),
identity,
)
save_json(
f"outputs/learning/train-{branch}-{mode}.json",
{
"branch": branch,
"mode": mode,
"identity": identity,
"epochs": stop,
"updates": updates,
"history": history,
"validation_history": validation_history,
"tokenization": rows,
"initial_parameters": initial,
"final_parameters": final,
"initial_model_digest": initial_digest,
"final_model_digest": final_digest,
"trainable_parameters": 494032768,
"reference_unchanged": True if ref_identity else None,
"checkpoint_sha256": sha(destination),
"checkpoint_bytes": destination.stat().st_size,
"checkpoint_compression": saved,
"optimization_seconds_this_process": optimization_seconds,
"training_and_validation_seconds_this_process": elapsed,
"max_torch_allocated_bytes": peak,
"test_visible": False,
},
)
def verify_resume(branch):
root = Path("assets/checkpoints")
a = load_resume_state(root / f"{branch}-continuous.pt.gz")
b = load_resume_state(root / f"{branch}-resume.pt.gz")
checks = {k: equal(a[k], b[k]) for k in a}
if a.keys() != b.keys() or not all(checks.values()):
raise ValueError("Continuous and resumed full states differ: " + str(checks))
save_json(
f"outputs/learning/resume-{branch}.json",
{
"status": "PASS",
"all_state_exact": True,
"checks": checks,
"boundary": "Complete epoch; lossless compressed local checkpoints, every model/optimizer/sampler/scheduler/RNG/history field compared",
},
)
Evaluate pairs and new answers
Offline preference versus generation
Frequent error: A chosen label appears in the generation prompt: rebuild the prompt from the case file alone.
Invariant: 180 pair scores and 180 generations per branch, with the same IDs.
File src/dpo_preferences/evaluate.py
import math, re, 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_tokenizer, reload_learned, parameter_manifest, manifest_digest
from .scores import (
format_prompt,
paired_completion_batch,
model_scores,
reference_identity,
read_cached_pair,
)
from .data import COURTESY
from .train import BRANCH_BETA
def parse_support_reply(text, finish):
sentences = [
s.strip() for s in re.split(r"(?<=[.!?])\s+", text.strip()) if s.strip()
]
fields = {}
unrecognized = []
duplicates = []
refusal = bool(
re.search(
r"\b(?:cannot help|cannot assist|unable to assist|I refuse)\b", text, re.I
)
)
for sentence in sentences:
value = sentence.rstrip(".!?").strip()
key = None
field = None
if sentence in COURTESY:
continue
match = re.fullmatch(
r"(?:Your replacement is|The replacement is|Replacement status:|I can confirm that the replacement is) (approved|pending|declined)",
value,
re.I,
)
if match:
key = "replacement"
field = match[1].lower()
else:
match = re.fullmatch(
r"The shipping date is (unknown|2026-11-\d{2})", value, re.I
)
if match:
candidate = match[1].lower()
if candidate == "unknown":
key = "shipping_date"
field = None
elif 1 <= int(candidate[-2:]) <= 30:
key = "shipping_date"
field = candidate
elif value.lower() == "i cannot confirm a shipping date":
key = "shipping_date"
field = None
elif value.lower() == "please provide your order number":
key = "request_order"
field = True
elif value.lower() == "your order number is already recorded":
key = "request_order"
field = False
if key is None:
unrecognized.append(sentence)
elif key in fields:
duplicates.append(key)
else:
fields[key] = field
covered = bool(sentences) and not unrecognized and not duplicates
complete = (
finish == "eos"
and covered
and set(fields) == {"replacement", "shipping_date", "request_order"}
and not refusal
)
return {
"recognized": fields,
"unrecognized_sentences": unrecognized,
"duplicate_fields": duplicates,
"all_text_covered_by_controlled_grammar": covered,
"complete": complete,
"refusal_pattern": refusal,
"truncated": finish != "eos",
}
def score_support_claims(rows, gold):
counts = {
"documents": len(rows),
"complete_controlled_replies": 0,
"exact_controlled_facts_and_action": 0,
"truncated": 0,
"unrecognized_prose": 0,
"refusal_patterns": 0,
"missing_order_cases": 0,
"useful_order_requests": 0,
"known_order_cases": 0,
"unnecessary_order_requests": 0,
"unknown_shipping_date_cases": 0,
"recognized_invented_dates": 0,
}
fields = dict.fromkeys(["replacement", "shipping_date", "request_order"], 0)
details = []
for row in rows:
expected = gold[row["document_id"]]["expected"]
parsed = parse_support_reply(row["text"], row["finish_reason"])
if parsed != row["parsed"]:
raise ValueError("Stored reply parse differs")
recognized = parsed["recognized"]
complete = parsed["complete"]
correct = {k: complete and recognized.get(k) == v for k, v in expected.items()}
exact = all(correct.values())
counts["complete_controlled_replies"] += complete
counts["exact_controlled_facts_and_action"] += exact
counts["truncated"] += parsed["truncated"]
counts["unrecognized_prose"] += bool(parsed["unrecognized_sentences"])
counts["refusal_patterns"] += parsed["refusal_pattern"]
for k, v in correct.items():
fields[k] += v
if expected["request_order"]:
counts["missing_order_cases"] += 1
counts["useful_order_requests"] += correct["request_order"]
else:
counts["known_order_cases"] += 1
counts["unnecessary_order_requests"] += (
recognized.get("request_order") is True
)
if expected["shipping_date"] is None:
counts["unknown_shipping_date_cases"] += 1
counts["recognized_invented_dates"] += (
recognized.get("shipping_date") is not None
)
details.append(
{
"document_id": row["document_id"],
"expected": expected,
"complete": complete,
"fields_correct": correct,
"exact": exact,
"recognized": recognized,
"unrecognized_sentences": parsed["unrecognized_sentences"],
}
)
n = len(rows)
counts.update(
{
"exact_accuracy_over_all_cases": counts["exact_controlled_facts_and_action"]
/ n,
"field_correct_over_all_cases": fields,
"field_accuracy_over_all_cases": {k: v / n for k, v in fields.items()},
"useful_request_rate_over_missing_order": counts["useful_order_requests"]
/ counts["missing_order_cases"],
"recognized_date_invention_rate_over_unknown_date": counts[
"recognized_invented_dates"
]
/ counts["unknown_shipping_date_cases"],
}
)
return {
"metrics": counts,
"rows": details,
"unrecognized_prose_is_not_certified_semantically": True,
}
def preference_accuracy(rows):
result = {}
for group in ["all", "chosen_shorter", "same_tokens", "chosen_longer"]:
selected = (
rows
if group == "all"
else [
r
for r in rows
if (
"chosen_shorter"
if r["response_length_delta"] < 0
else (
"chosen_longer"
if r["response_length_delta"] > 0
else "same_tokens"
)
)
== group
]
)
n = len(selected)
def accuracy(key):
return (
sum(1 if r[key] > 0 else 0.5 if r[key] == 0 else 0 for r in selected)
/ n
if n
else None
)
result[group] = {
"pairs": n,
"raw_sum_preference_accuracy_half_ties": accuracy("raw_sum_margin"),
"reference_adjusted_accuracy_half_ties": accuracy(
"reference_adjusted_margin"
),
"mean_token_preference_accuracy_half_ties": accuracy("mean_token_margin"),
"mean_raw_sum_margin": (
math.fsum(r["raw_sum_margin"] for r in selected) / n if n else None
),
"mean_reference_adjusted_margin": (
math.fsum(r["reference_adjusted_margin"] for r in selected) / n
if n
else None
),
}
return result
def load_policy(branch):
if branch not in ["sft", "chosen", *BRANCH_BETA]:
raise ValueError("Unknown final policy")
path = Path(
"assets/reference/sft.safetensors"
if branch == "sft"
else f"assets/checkpoints/{branch}-resume.safetensors"
)
model, metadata = reload_learned(path)
for parameter in model.parameters():
parameter.requires_grad_(False)
return model.cuda().eval(), {
"branch": branch,
"file_sha256": sha(path),
"model_digest": metadata["model_digest"],
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
}
@torch.no_grad()
def score_heldout_pairs(config, branch):
seed_all(config["seed"])
model, identity = load_policy(branch)
tokenizer = load_tokenizer()
pages = [
p for p in read_json("data/generated/input/pages.json") if p["split"] == "test"
]
gold = {g["document_id"]: g for g in read_json("data/generated/labels/gold.json")}
cache = read_json("outputs/reference-cache/test.json")
ref_identity = reference_identity()
rows = []
probes = []
torch.cuda.synchronize()
begun = time.perf_counter()
for index, page in enumerate(pages):
pair = paired_completion_batch(
tokenizer, page, gold[page["document_id"]], config
)
reference = read_cached_pair(cache, ref_identity, pair)["scores"]
scores = {}
for side in ["chosen", "rejected"]:
sums, counts, values, logits, targets = model_scores(model, [pair[side]])
scores[side] = {
"sum_logprob": float(sums[0]),
"mean_logprob": float(sums[0] / counts[0]),
"response_tokens": int(counts[0]),
"token_logprobs": values.cpu().tolist(),
"targets": targets.cpu().tolist(),
}
if index == 0:
probes.append(
{
"document_id": page["document_id"],
"side": side,
"response_position": 0,
"target": int(targets[0]),
"logits": logits[0].cpu().tolist(),
"logprob": float(values[0]),
}
)
a, b = scores["chosen"]["sum_logprob"], scores["rejected"]["sum_logprob"]
ra, rb = (
reference["chosen"]["sum_logprob"],
reference["rejected"]["sum_logprob"],
)
rows.append(
{
"document_id": page["document_id"],
"pair": pair,
"scores": scores,
"reference_scores": reference,
"response_length_delta": pair["response_length_delta"],
"raw_sum_margin": a - b,
"reference_adjusted_margin": (a - b) - (ra - rb),
"mean_token_margin": scores["chosen"]["mean_logprob"]
- scores["rejected"]["mean_logprob"],
}
)
torch.cuda.synchronize()
save_json(
f"outputs/evaluation/pairs-{branch}.json",
{
"branch": branch,
"identity": identity,
"reference_identity": ref_identity,
"reference_cache_sha256": sha("outputs/reference-cache/test.json"),
"rows": rows,
"probes": probes,
"metrics": preference_accuracy(rows),
"seconds": time.perf_counter() - begun,
"test_selection": False,
},
)
@torch.no_grad()
def generate_heldout_replies(config, branch):
if (
Path("data/generated/labels").exists()
or Path("outputs/reference-cache").exists()
):
raise ValueError("Blind generation must not mount preference labels or caches")
seed_all(config["seed"])
model, identity = load_policy(branch)
tokenizer = load_tokenizer()
pages = [
p for p in read_json("data/generated/input/pages.json") if p["split"] == "test"
]
generation = GenerationConfig(
do_sample=False,
num_beams=1,
max_new_tokens=config["max_new_tokens"],
use_cache=True,
eos_token_id=151645,
pad_token_id=151643,
bos_token_id=151643,
)
rows = []
torch.cuda.reset_peak_memory_stats()
torch.cuda.synchronize()
begun = time.perf_counter()
for index, page in enumerate(pages):
messages, ids = format_prompt(
tokenizer, page["dossier"], config["template_version"]
)
if not 0 < len(ids) <= config["max_prompt_tokens"]:
raise ValueError("No silent prompt truncation")
tensor = torch.tensor([ids], device="cuda")
torch.cuda.synchronize()
started = time.perf_counter()
output = (
model.generate(
input_ids=tensor,
attention_mask=torch.ones_like(tensor),
generation_config=generation,
)[0, len(ids) :]
.cpu()
.tolist()
)
torch.cuda.synchronize()
finish = "eos" if output and output[-1] == 151645 else "length"
text = tokenizer.decode(
output[:-1] if finish == "eos" else output,
skip_special_tokens=False,
clean_up_tokenization_spaces=False,
)
rows.append(
{
"document_id": page["document_id"],
"messages": messages,
"input_ids": ids,
"output_ids": output,
"text": text,
"finish_reason": finish,
"parsed": parse_support_reply(text, finish),
"input_tokens": len(ids),
"output_tokens": len(output),
"seconds": time.perf_counter() - started,
}
)
if (index + 1) % 30 == 0:
print({"branch": branch, "generated_test_replies": index + 1}, flush=True)
torch.cuda.synchronize()
save_json(
f"outputs/inference/predictions-{branch}.json",
{
"branch": branch,
"identity": identity,
"generation": generation.to_dict(),
"input_sha256": sha("data/generated/input/pages.json"),
"rows": rows,
"seconds": time.perf_counter() - begun,
"max_torch_allocated_bytes": torch.cuda.max_memory_allocated(),
"gold_or_reference_cache_visible": False,
"training": False,
},
)
def evaluate(config):
gold = {g["document_id"]: g for g in read_json("data/generated/labels/gold.json")}
test_ids = [
p["document_id"]
for p in read_json("data/generated/input/pages.json")
if p["split"] == "test"
]
results = {}
for branch in ["sft", "chosen", *BRANCH_BETA]:
predictions = read_json(f"outputs/inference/predictions-{branch}.json")
pairs = read_json(f"outputs/evaluation/pairs-{branch}.json")
if (
[r["document_id"] for r in predictions["rows"]] != test_ids
or [r["document_id"] for r in pairs["rows"]] != test_ids
or len(test_ids) != 180
):
raise ValueError("Every branch must retain the same180reserved IDs")
if pairs["identity"] != predictions["identity"]:
raise ValueError("Pair scoring and generation used different weights")
results[branch] = {
"free_replies": score_support_claims(predictions["rows"], gold),
"pairs": preference_accuracy(pairs["rows"]),
}
print(
{
"branch": branch,
"free_replies": results[branch]["free_replies"]["metrics"],
"pairs": results[branch]["pairs"],
},
flush=True,
)
save_json(
"outputs/evaluation/comparison.json",
{
"results": results,
"all_beta_conditions_reported": True,
"best_beta_selected": False,
"human_preferences": False,
"quality_gain_required": False,
"reference_sha256": sha("assets/reference/sft.safetensors"),
"training_compute_budgets_not_equal": True,
"no_customer_action": True,
},
)
Trigger then block a stale cache
Reference-signal integrity
Frequent error: Disabling the check to continue hides the error: retain rejection as a required test.
Invariant: The stale cache is rejected; the recalculated cache matches direct calculation.
File src/dpo_preferences/exercise.py
from pathlib import Path
import torch
from .config import read_json, save_json, sha, seed_all
from .scores import (
reference_identity,
paired_completion_batch,
read_cached_pair,
reference_cache_key,
)
from .warmstart import freeze_reference, score_pair
from .model import load_tokenizer, parameter_manifest, manifest_digest
def main_artifacts():
paths = ["assets/reference/sft.safetensors", "outputs/evaluation/comparison.json"]
for split in ["train", "validation", "test"]:
paths.append(f"outputs/reference-cache/{split}.json")
for branch in ["sft", "chosen", "dpo-005", "dpo-010", "dpo-020"]:
paths += [
f"outputs/evaluation/pairs-{branch}.json",
f"outputs/inference/predictions-{branch}.json",
]
if branch != "sft":
paths.append(f"assets/checkpoints/{branch}-resume.safetensors")
return paths
def preserve_main():
path = Path("outputs/exercise/protected.json")
if path.exists():
raise ValueError("Preserve existing exercise snapshot")
save_json(path, {name: sha(name) for name in main_artifacts()})
@torch.no_grad()
def exercise(config):
seed_all(config["seed"])
before = read_json("outputs/exercise/protected.json")
if before != {name: sha(name) for name in main_artifacts()}:
raise ValueError("Main artifacts changed before exercise")
model, metadata = freeze_reference()
tokenizer = load_tokenizer()
old = read_json("outputs/reference-cache/test.json")
identity = reference_identity(config["exercise_template_version"])
pages = [
p for p in read_json("data/generated/input/pages.json") if p["split"] == "test"
][: config["exercise_pairs"]]
gold = {r["document_id"]: r for r in read_json("data/generated/labels/gold.json")}
rows = {}
checks = []
for page in pages:
pair = paired_completion_batch(
tokenizer,
page,
gold[page["document_id"]],
config,
config["exercise_template_version"],
)
rejection = None
try:
read_cached_pair(old, identity, pair)
except ValueError as error:
rejection = str(error)
if rejection is None or not rejection.startswith("STALE_REFERENCE_CACHE"):
raise ValueError("Stale template cache was accepted")
if (
pair["chosen"]["input_ids"]
== old["rows"][page["document_id"]]["pair"]["chosen"]["input_ids"]
):
raise ValueError("Template exercise did not change actual IDs")
cached = score_pair(model, pair)
rows[page["document_id"]] = {
"key": reference_cache_key(identity, pair),
"pair": pair,
"scores": cached,
}
fresh = {"identity": identity, "rows": rows}
retrieved = read_cached_pair(fresh, identity, pair)["scores"]
direct = score_pair(model, pair)
if retrieved != direct:
raise ValueError(
"Fresh cached and separately recomputed reference scores differ"
)
checks.append(
{
"document_id": page["document_id"],
"old_cache_rejected_before_score_use": True,
"error": rejection,
"fresh_cached_scores": retrieved,
"separately_recomputed_scores": direct,
"all_scores_exact": True,
}
)
after = {name: sha(name) for name in main_artifacts()}
if (
before != after
or manifest_digest(parameter_manifest(model)) != metadata["model_digest"]
):
raise ValueError("Exercise mutated main artifacts or fixed reference")
del model
torch.cuda.empty_cache()
from .evaluate import load_policy
from .scores import model_scores, dpo_sigmoid_loss
policy, policy_identity = load_policy("dpo-010")
for check in checks:
pair = rows[check["document_id"]]["pair"]
a = model_scores(policy, [pair["chosen"]])[0]
b = model_scores(policy, [pair["rejected"]])[0]
losses = []
for field in ["fresh_cached_scores", "separately_recomputed_scores"]:
ref = check[field]
ra = torch.tensor(
[ref["chosen"]["sum_logprob"]], dtype=torch.float32, device="cuda"
)
rb = torch.tensor(
[ref["rejected"]["sum_logprob"]], dtype=torch.float32, device="cuda"
)
loss, margin = dpo_sigmoid_loss(a, b, ra, rb, 0.1)
losses.append(float(loss[0]))
if losses[0] != losses[1]:
raise ValueError("Cached and direct reference DPO losses differ")
check.update(
policy_chosen_sum=float(a[0]),
policy_rejected_sum=float(b[0]),
beta=0.1,
cached_reference_loss=losses[0],
direct_reference_loss=losses[1],
loss_exact=True,
)
if before != {name: sha(name) for name in main_artifacts()}:
raise ValueError("Main artifacts changed during policy scoring")
save_json(
"outputs/exercise/reference-cache-v2.json",
{
"identity": identity,
"reference_model_digest": metadata["model_digest"],
"rows": rows,
},
)
save_json(
"outputs/exercise/cache-invalidation.json",
{
"checks": checks,
"pairs": len(pages),
"protected_sha256": before,
"main_artifacts_unchanged": True,
"reference_unchanged": True,
"extra_training_updates": 0,
"policy_identity": policy_identity,
"old_template": old["identity"]["template_version"],
"new_template": identity["template_version"],
},
)
Sources
DPO, authors' paper, pinned original function, Qwen2.5-0.5B-Instruct. Local data, budget, cache, and grammar choices do not reproduce the paper's results.
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",
"pairs": 1200,
"warmstart_demonstrations": 210,
"test_pairs_per_policy": 180,
"free_generations": 900,
"policies": 5,
"parameters": 494032768,
"exact_resumed_phases": 5,
"exercise_pairs": 12,
"metrics": {
"sft": {
"free_replies": {
"documents": 180,
"complete_controlled_replies": 180,
"exact_controlled_facts_and_action": 180,
"truncated": 0,
"unrecognized_prose": 0,
"refusal_patterns": 0,
"missing_order_cases": 88,
"useful_order_requests": 88,
"known_order_cases": 92,
"unnecessary_order_requests": 0,
"unknown_shipping_date_cases": 88,
"recognized_invented_dates": 0,
"exact_accuracy_over_all_cases": 1.0,
"field_correct_over_all_cases": {
"replacement": 180,
"shipping_date": 180,
"request_order": 180
},
"field_accuracy_over_all_cases": {
"replacement": 1.0,
"shipping_date": 1.0,
"request_order": 1.0
},
"useful_request_rate_over_missing_order": 1.0,
"recognized_date_invention_rate_over_unknown_date": 0.0
},
"pairs": {
"all": {
"pairs": 180,
"raw_sum_preference_accuracy_half_ties": 0.8388888888888889,
"reference_adjusted_accuracy_half_ties": 0.5,
"mean_token_preference_accuracy_half_ties": 0.8444444444444444,
"mean_raw_sum_margin": 13.198352245469707,
"mean_reference_adjusted_margin": 0.0
},
"chosen_shorter": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 0.7555555555555555,
"reference_adjusted_accuracy_half_ties": 0.5,
"mean_token_preference_accuracy_half_ties": 0.5111111111111111,
"mean_raw_sum_margin": 6.847004530164931,
"mean_reference_adjusted_margin": 0.0
},
"same_tokens": {
"pairs": 90,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 0.5,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 16.468573133110638,
"mean_reference_adjusted_margin": 0.0
},
"chosen_longer": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 0.6,
"reference_adjusted_accuracy_half_ties": 0.5,
"mean_token_preference_accuracy_half_ties": 0.8666666666666667,
"mean_raw_sum_margin": 13.009258185492621,
"mean_reference_adjusted_margin": 0.0
}
}
},
"chosen": {
"free_replies": {
"documents": 180,
"complete_controlled_replies": 180,
"exact_controlled_facts_and_action": 180,
"truncated": 0,
"unrecognized_prose": 0,
"refusal_patterns": 0,
"missing_order_cases": 88,
"useful_order_requests": 88,
"known_order_cases": 92,
"unnecessary_order_requests": 0,
"unknown_shipping_date_cases": 88,
"recognized_invented_dates": 0,
"exact_accuracy_over_all_cases": 1.0,
"field_correct_over_all_cases": {
"replacement": 180,
"shipping_date": 180,
"request_order": 180
},
"field_accuracy_over_all_cases": {
"replacement": 1.0,
"shipping_date": 1.0,
"request_order": 1.0
},
"useful_request_rate_over_missing_order": 1.0,
"recognized_date_invention_rate_over_unknown_date": 0.0
},
"pairs": {
"all": {
"pairs": 180,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 0.5111111111111111,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 28.638852359494518,
"mean_reference_adjusted_margin": 15.440500114024811
},
"chosen_shorter": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 40.23522782304014,
"mean_reference_adjusted_margin": 33.38822329287521
},
"same_tokens": {
"pairs": 90,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 0.23333333333333334,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 14.97690743867877,
"mean_reference_adjusted_margin": -1.4916656944318674
},
"chosen_longer": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 0.5777777777777777,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 44.36636673758039,
"mean_reference_adjusted_margin": 31.35710855208777
}
}
},
"dpo-005": {
"free_replies": {
"documents": 180,
"complete_controlled_replies": 113,
"exact_controlled_facts_and_action": 113,
"truncated": 0,
"unrecognized_prose": 67,
"refusal_patterns": 0,
"missing_order_cases": 88,
"useful_order_requests": 43,
"known_order_cases": 92,
"unnecessary_order_requests": 0,
"unknown_shipping_date_cases": 88,
"recognized_invented_dates": 0,
"exact_accuracy_over_all_cases": 0.6277777777777778,
"field_correct_over_all_cases": {
"replacement": 113,
"shipping_date": 113,
"request_order": 113
},
"field_accuracy_over_all_cases": {
"replacement": 0.6277777777777778,
"shipping_date": 0.6277777777777778,
"request_order": 0.6277777777777778
},
"useful_request_rate_over_missing_order": 0.48863636363636365,
"recognized_date_invention_rate_over_unknown_date": 0.0
},
"pairs": {
"all": {
"pairs": 180,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 221.22210419372553,
"mean_reference_adjusted_margin": 208.0237519482558
},
"chosen_shorter": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 188.20255571719673,
"mean_reference_adjusted_margin": 181.3555511870318
},
"same_tokens": {
"pairs": 90,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 232.4231789165073,
"mean_reference_adjusted_margin": 215.95460578339666
},
"chosen_longer": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 231.83950322469076,
"mean_reference_adjusted_margin": 218.83024503919813
}
}
},
"dpo-010": {
"free_replies": {
"documents": 180,
"complete_controlled_replies": 99,
"exact_controlled_facts_and_action": 99,
"truncated": 0,
"unrecognized_prose": 81,
"refusal_patterns": 0,
"missing_order_cases": 88,
"useful_order_requests": 44,
"known_order_cases": 92,
"unnecessary_order_requests": 0,
"unknown_shipping_date_cases": 88,
"recognized_invented_dates": 0,
"exact_accuracy_over_all_cases": 0.55,
"field_correct_over_all_cases": {
"replacement": 99,
"shipping_date": 99,
"request_order": 99
},
"field_accuracy_over_all_cases": {
"replacement": 0.55,
"shipping_date": 0.55,
"request_order": 0.55
},
"useful_request_rate_over_missing_order": 0.5,
"recognized_date_invention_rate_over_unknown_date": 0.0
},
"pairs": {
"all": {
"pairs": 180,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 135.90712270219294,
"mean_reference_adjusted_margin": 122.70877045672322
},
"chosen_shorter": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 116.39779642207755,
"mean_reference_adjusted_margin": 109.55079189191262
},
"same_tokens": {
"pairs": 90,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 142.48586250808503,
"mean_reference_adjusted_margin": 126.0172893749744
},
"chosen_longer": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 142.25896937052408,
"mean_reference_adjusted_margin": 129.24971118503146
}
}
},
"dpo-020": {
"free_replies": {
"documents": 180,
"complete_controlled_replies": 146,
"exact_controlled_facts_and_action": 146,
"truncated": 0,
"unrecognized_prose": 34,
"refusal_patterns": 0,
"missing_order_cases": 88,
"useful_order_requests": 88,
"known_order_cases": 92,
"unnecessary_order_requests": 0,
"unknown_shipping_date_cases": 88,
"recognized_invented_dates": 0,
"exact_accuracy_over_all_cases": 0.8111111111111111,
"field_correct_over_all_cases": {
"replacement": 146,
"shipping_date": 146,
"request_order": 146
},
"field_accuracy_over_all_cases": {
"replacement": 0.8111111111111111,
"shipping_date": 0.8111111111111111,
"request_order": 0.8111111111111111
},
"useful_request_rate_over_missing_order": 1.0,
"recognized_date_invention_rate_over_unknown_date": 0.0
},
"pairs": {
"all": {
"pairs": 180,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 73.97669998885443,
"mean_reference_adjusted_margin": 60.77834774338472
},
"chosen_shorter": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 69.53787630928888,
"mean_reference_adjusted_margin": 62.690871779123945
},
"same_tokens": {
"pairs": 90,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 68.6144319921318,
"mean_reference_adjusted_margin": 52.14585885902117
},
"chosen_longer": {
"pairs": 45,
"raw_sum_preference_accuracy_half_ties": 1.0,
"reference_adjusted_accuracy_half_ties": 1.0,
"mean_token_preference_accuracy_half_ties": 1.0,
"mean_raw_sum_margin": 89.14005966186524,
"mean_reference_adjusted_margin": 76.13080147637261
}
}
}
},
"quality_gain_required": false,
"scope": "Independent data-only formulas, whole-reply grammar, tokenizer decoding, masks/cache identities, full-vocabulary score probes, all-case metrics, update budgets and captured exact-state comparisons; not BPE re-encoding, Torch checkpoint deserialization or full Transformer reimplementation"
}
