Il supporto fittizio AtelierDesk deve rispondere con i fatti del dossier e chiedere un'informazione mancante invece di promettere un termine inventato. Per lo stesso dossier, due risposte sono annotate come preferita/rifiutata da regole di simulazione pubbliche.
Confrontate la policy dopo SFT locale, dopo un supplemento SFT sulle risposte scelte e dopo DPO. Mostrate per ogni coppia le log-probabilità della policy e della sua riferimento, poi le risposte libere su nuovi dossier.
Scaricare il progetto completo, revisione r001
Dal dossier a una preferenza
Una preferenza confronta due risposte allo stesso dossier. Non è né un voto assoluto né una prova di soddisfazione umana. Le regole di AtelierDeskPrefs-v1 sono pubbliche e simulate: rispettare lo stato della sostituzione e la data di spedizione registrati; chiedere il numero dell'ordine solo se manca. Una data sconosciuta resta sconosciuta. Una formula cortese non rende vera una promessa inventata.
300 famiglie producono quattro coppie ciascuna: 840 train, 180 validation e 180 test, su famiglie disgiunte 0–209, 210–254 e 255–299. Una dimostrazione supplementare per famiglia train fornisce 210 esempi per l'avvio SFT. Dossier e identificativi sono distinti, ma i fatti elementari e i modelli linguistici restano condivisi. Questo non misura una generalizzazione a veri servizi clienti o a intenzioni sconosciute.
Due varianti cambiano lo stato; una cambia la data; una inverte la richiesta del numero dell'ordine. Le formule di cortesia alternano: scelta più corta, della stessa lunghezza in parole o più lunga. Un audit pubblica le presenze di ogni parola nelle due label, un classificatore binario per parola e una scelta per lunghezza con mezzo credito alle parità. Questo audit non prova l'assenza di ogni scorciatoia appresa. Gli strati di valutazione usano le lunghezze reali in token, che possono differire dai conteggi di parole. I metadati chosen/rejected e il motivo della preferenza non sono mai inclusi nei dossier di generazione.
La policy produce probabilità condizionate
Qwen2.5-0.5B-Instruct, revisione 7ae557604adf67be50417f59c2c2f167def9a775, viene caricato rigorosamente: 24 blocchi, larghezza 896, 14 teste Q, due K/V, MLP 4864, vocabolario 151936 e embedding/output condivisi. Tutti i suoi 494 032 768 parametri unici sono addestrabili in FP32. Il profilo usa attenzione eager, TF32 disattivato, microbatch di una risposta; nessuna quantizzazione, LoRA o distribuzione automatica su più GPU.
Il tokenizer applica il proprio template al messaggio di sistema e al dossier, poi aggiunge il confine assistant. La risposta è codificata separatamente e riceve esattamente un EOS=151645 finale. Il contratto è esplicito: non pretende di riprodurre una ritokenizzazione della stringa concatenata. Massimo 256 token di prompt, 128 di risposta con EOS, totale inferiore a 512; ogni superamento fallisce senza troncare.
Per predire il token in posizione t, prendiamo il logit in t−1. Il codice allinea logits[:, :-1] con labels[:, 1:]. Le label del prompt e del padding valgono −100; i target della risposta, EOS incluso, restano attivi. PAD=151643 è mascherato per posizione. Il prompt non ha una CE propria, ma i suoi token condizionano le probabilità di risposta: ignorarne le label non elimina la sua influenza.
Per una risposta y, s(x,y) è la SOMMA delle log-probabilità dei suoi token attivi. Una risposta lunga accumula più termini generalmente negativi. Il rapporto pubblica anche le medie per token per diagnosticare un bias di lunghezza; queste medie non sostituiscono mai le somme nel DPO.
SFT locale poi riferimento congelato
L'avvio impara le 210 dimostrazioni train per due epoche, con accumulazione di otto esempi: 27 aggiornamenti per epoca, di cui l'ultimo su due esempi, quindi 54 in totale. La CE è normalizzata per il numero totale di token risposta dell'aggiornamento; una media di medie darebbe un altro peso alle risposte brevi. AdamW usa lr=0,00002 e weight_decay=0,01, clipping a 1, warm-up di cinque passi poi decrescita lineare fino al 10% del tasso massimo.
Il checkpoint SFT finale viene salvato in Safetensors e ricaricato rigorosamente. Il riferimento è un'istanza congelata, eval e no_grad. Non entra mai nell'ottimizzatore. Il suo hash e il digest dei suoi parametri restano invariati. Ogni policy di preferenza riparte da un ricaricamento indipendente dello stesso SFT; non condivide un oggetto mutabile con il riferimento.
Per ogni coppia, la cache conserva le due risposte realmente tokenizzate, le loro maschere, i loro punteggi per token e le loro somme. La sua chiave lega checkpoint, tokenizer e configurazione del tokenizer, template e testo di sistema, codice di scoring, lock del modello, ID e label complete. Una cache esistente in cui una di queste identità differisce viene rifiutata prima dell'uso. Solo i file train e validation vengono caricati in learning; il file test e le uscite di generazione non gli sono accessibili.
Il costo del SFT, della cache di riferimento e di ogni ottimizzazione viene misurato separatamente. La cache evita di caricare una seconda riferimento durante ogni backward DPO; il suo costo iniziale non è gratuito.
La loss codificata e il suo gradiente
Per una coppia intervengono quattro numeri: sθ+ e sθ− per la policy, sref+ e sref− per il riferimento. Il margine è m = β[(sθ+ − sθ−) − (sref+ − sref−)]. La loss è −logsigmoid(m), mediata sulle coppie dell'aggiornamento. Il codice usa logsigmoid per la stabilità numerica. Il riferimento fornisce costanti senza gradiente.
Se policy e riferimento sono uguali, il margine vale zero e la loss log(2). Eppure, per due risposte diverse, le derivate dei punteggi della policy sono opposte e non nulle: −β/2 per sθ+ e +β/2 per sθ− su una coppia. Il modello può quindi imparare. Se le due risposte hanno esattamente gli stessi token e maschere, i due termini si compensano e non danno questo segnale; queste coppie vengono rifiutate nella costruzione.
Il valore e i gradienti FP64 vengono confrontati con la funzione preference_loss degli autori di DPO, estratta dal commit f8b8c0f49dc92a430bae41585f9d467d3618fe2f, con label_smoothing=0, ipo=False e reference_free=False. Sono forniti licenza Apache-2.0 e provenienza. Vengono usati soltanto questa funzione e i suoi import necessari; il trainer originale e le sue impostazioni globali non vengono eseguiti.
Un confronto definito prima dei risultati
Dopo il SFT, quattro policy ricevono ciascuna 840 prompt per due epoche, nello stesso ordine deterministico, cioè 210 aggiornamenti e 1680 presentazioni. Chosen-only apprende solo la risposta scelta tramite CE normalizzata in token. DPO riceve scelta e rifiutata, con β=0,05, 0,1 o 0,2. I quattro rami usano AdamW lr=0,000001, weight_decay=0,01, clipping a 1, warm-up di 21 passi poi decrescita fino al 10%.
Il numero di presentazioni dei prompt è uguale; il numero di token valutati, i forward e il calcolo non lo sono. DPO media le sue loss per coppia, chosen-only per token risposta. Tutti i beta e il checkpoint finale sono riportati: nessuna selezione del miglior beta o checkpoint usa il test. La validazione dopo ogni epoca fornisce NLL scelta e, per i rami di preferenza, margini e accuracy di preferenza. Non interrompe l'apprendimento.
Ogni fase, incluso il SFT, è eseguita con interruzione dopo un'epoca, percorso continuo di due epoche e ripresa della seconda. Modello completo, AdamW, sampler, scheduler, RNG Python/NumPy/PyTorch CPU/CUDA e cronologie finali devono essere esattamente uguali. Questo contratto riguarda il confine dell'epoca, non l'arresto nel mezzo di un backward.
I quindici stati locali sono scritti in gzip livello 1 senza perdita, in un file temporaneo poi rinominati. Vengono misurati i byte serializzati non compressi e la dimensione realmente scritta. Questa scelta non promette un tasso di compressione né una riduzione della VRAM. Gli stati Torch sono oggetti fidati prodotti localmente; non sostituiteli con un download non affidabile. Le cinque policy finali sono anch'esse conservate in Safetensors. Gli artefatti esistenti non vengono sovrascritti.
Preferire offline e rispondere liberamente
Le cinque policy SFT/chosen-only/DPO×3 valutano le stesse 180 coppie test. L'accuracy conta s+ > s−, zero per l'ordine inverso e mezzo punto per l'uguaglianza. I margini corretti dal riferimento e le medie per token sono separati. I rapporti stratificati per lunghezza impediscono di nascondere un fallimento sulle risposte scelte più lunghe dietro una media generale.
Ogni policy genera anche 180 risposte senza chosen/rejected nel proprio contesto: solo dossier con la stessa istruzione, decodifica greedy, una beam, al massimo 128 nuovi token, arresto EOS. Tutti gli ID, testi grezzi, ragioni di arresto e durate restano disponibili. Un output fermato al limite fallisce anche se le prime frasi sembrano corrette. Non viene usato alcun modello giudice, invio al cliente o azione aziendale.
Un parser indipendente dalla generazione riconosce le clausole controllate del gioco di dati: stato, data conosciuta/sconosciuta, richiesta o riconoscimento della presenza del numero. Esige la copertura di TUTTO il testo, tre campi distinti ed EOS. Una frase sconosciuta, un duplicato, una tag Markdown o un rifiuto rende la risposta incompleta. Una frase naturale corretta fuori grammatica può quindi fallire: questo punteggio ristretto non è una misura universale di fedeltà semantica.
L'esattezza richiede i tre fatti/azioni corretti sui 180 casi, senza escludere gli output non validi. Le richieste utili sono riportate sui casi in cui manca il numero. Le date inventate conteggiate sono le date riconosciute quando la data è sconosciuta; le frasi non riconosciute sono riportate separatamente e non vengono mai certificate senza invenzione. I rifiuti conteggiati seguono espressioni esplicite; questo rilevatore non identifica tutti i rifiuti possibili. Un aumento del margine DPO non prova dunque un aumento della qualità libera, tanto meno della soddisfazione del cliente.
Esercizio: un vecchio file non garantisce nulla
Sui primi dodici dossier test, il template support-v2 aggiunge un'istruzione di concisione. Gli ID condizionanti devono davvero cambiare. La vecchia cache viene rifiutata prima di leggere i suoi punteggi. Un riferimento congelato ricalcola una nuova cache poi, separatamente, gli stessi punteggi senza cache; devono essere esattamente uguali. La policy dpo-010 valuta le nuove coppie e le due vie di riferimento danno la stessa loss DPO. Non viene svolto alcun addestramento supplementare. Gli hash di pesi, cache e risultati principali restano invariati.
Un solo seme, preferenze simulate e un linguaggio controllato limitano la portata del laboratorio. Beta non corregge annotazioni contraddittorie o un corpus mal concepito. Non esiste alcun requisito di guadagno: una regressione o una generazione non valida deve essere conservata e spiegata.
Costruire coppie informative senza scorciatoie
Preferenze e bias di lunghezza
Errore frequente: un marcatore di generazione rivela chosen; rimuovete i metadati dai testi e testate la scorciatoia.
Invariante: 840/180/180 coppie e 210 dimostrazioni train, motivi e lunghezze sottoposti ad audit.
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
Valutare esattamente le risposte
Somma delle log-probabilità e maschere
Errore frequente: un EOS duplicato favorisce artificialmente una risposta; ispezionate la sequenza decodificata e i suoi ID.
Invariante: punteggi identici al calcolo token per token e PAD invarianti.
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,
},
)
Avviare la policy e congelare il riferimento
SFT locale e inizializzazione comune
Errore frequente: reference punta allo stesso oggetto mutabile di policy; copiate realmente i pesi e testate l'indipendenza.
Invariante: gli hash policy/ref sono uguali all'inizio e ref non ha alcun gradiente.
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,
},
)
Codificare DPO e la baseline chosen-only
Margine relativo, sigmoid e ottimizzazione
Errore frequente: sostituire le somme con medie cambia l'obiettivo; test numerico su lunghezze diseguali.
Invariante: i gradienti sono finiti, ref immutabile e budget di esposizione acquisiti.
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",
},
)
Valutare coppie e nuove risposte
Preferenza offline rispetto alla generazione
Errore frequente: una label chosen finisce nel prompt di generazione; ricostruite il prompt dal solo dossier.
Invariante: 180 punteggi di coppie e 180 generazioni per ramo, con gli stessi ID.
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,
},
)
Provocare poi bloccare una cache obsoleta
Integrità del segnale di riferimento
Errore frequente: disattivare la verifica per proseguire nasconde l'errore; mantenete il rifiuto come test obbligatorio.
Invariante: la cache obsoleta viene rifiutata; la cache ricalcolata corrisponde al calcolo diretto.
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"],
},
)
Fonti
DPO, articolo degli autori, funzione originale congelata, Qwen2.5-0.5B-Instruct. Le scelte locali di dati, budget, cache e grammatica non costituiscono una riproduzione dei risultati dell'articolo.
Osservazioni di sviluppo
Valori ricalcolati dalle uscite grezze di questo percorso. Non sono soglie di miglioramento; il percorso dallo ZIP li ricalcola sulle stesse popolazioni.
{
"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"
}
