Uno strumento fittizio trasforma fatti di incidente in un breve resoconto che separa osservazione e ipotesi. Le coppie illustrano una sintesi fedele e una sintesi che aggiunge una causa non stabilita. Il compito non è né medico né un controllo di apparecchiature.
Confrontate SFT chosen-only e ORPO dagli stessi pesi instruct. Visualizzate NLL scelta, log-odds e qualità dei riassunti, con un esempio in cui una formula numericamente ingenua fallisce.
Scaricare il progetto completo, revisione r003
Prima il fatto, poi il resoconto
IncidentNotesPrefs-v1 contiene 300 famiglie di quattro coppie, cioè 1200 esempi. Le famiglie 0–209 producono 840 coppie train; 210–254 producono 180 validation; 255–299 producono 180 test. Ogni dossier nomina un'unità fittizia, il suo stato stopped/running, una pressione nota o non misurata e una causa confermata o sconosciuta. Una vecchia nota fornisce uno stato contraddittorio e una causa suggerita; la rilevazione corrente la sostituisce esplicitamente. Non viene proposta alcuna azione su un'apparecchiatura reale.
La risposta scelta esprime i fatti correnti. Quella rifiutata inverte una sola informazione: causa in due formulazioni, pressione in una, stato in una. Alcune cause sono confermate: scrivere sistematicamente « sconosciuta » è allora falso. Una risposta cortese può comunque inventare una causa. Le formule di cortesia si alternano per produrre scelta più corta, della stessa lunghezza in parole o più lunga. Le parole e questi regimi sono sottoposti ad audit; gli strati finali usano i token reali.
Tutte le varianti di una famiglia restano in una partizione. Gli identificativi sono distinti, ma fatti elementari e forme linguistiche sono condivisi: questa separazione non prova una generalizzazione a incidenti o domini reali. Le annotazioni sono simulate, senza alcun feedback umano dichiarato. Un audit per parola non dimostra l'assenza di ogni scorciatoia semantica.
Una distribuzione sui token
La base Qwen2.5-0.5B-Instruct è acquisita alla revisione 7ae557604adf67be50417f59c2c2f167def9a775. Tutti i suoi 494 032 768 parametri unici sono addestrabili in FP32. L'architettura è controllata: 24 blocchi, larghezza 896, 14 teste Q e due K/V, MLP 4864, vocabolario 151936, embedding/output condivisi. Attenzione eager e TF32 disattivato; una sola scheda scelta esplicitamente. Nessuna LoRA, quantizzazione o copia di riferimento.
Il template del tokenizer assembla sistema, incidente e confine assistant. La risposta è codificata separatamente, con un solo EOS=151645 finale. Questo contratto non afferma un'equivalenza alla ritokenizzazione dell'intera stringa. I limiti sono 256 token di prompt, 128 di risposta incluso EOS, totale al massimo 512. Ogni superamento provoca un errore identificato, mai un troncamento dei dati di apprendimento.
I logits in posizione t−1 predicono il token t. Dopo lo spostamento, restano attive solo le label della risposta; prompt e padding valgono −100. PAD=151643 ed EOS sono distinti e le posizioni reali determinano le maschere. Il prompt non è supervisionato direttamente, ma condiziona ogni punteggio di risposta.
Per una risposta di n token attivi, calcoliamo una somma s di log-probabilità, poi ell=s/n. q=exp(ell) è la media geometrica delle probabilità dei token, non la probabilità congiunta della risposta. EOS partecipa al conteggio n. Una risposta vuota viene rifiutata. A differenza del DPO, l'obiettivo ORPO adottato usa ell, non s.
Dalle probabilità alle quote
La quota di q vale q/(1−q). Il suo logaritmo vale ell−log(1−exp(ell)). Per scelta e rifiutata, sottraiamo le due log-quote; una differenza positiva favorisce la scelta. La penalità per coppia è −logsigmoid(differenza). La loss totale aggiunge una supervisione: NLL scelta + lambda × media delle penalità di coppia.
Due riduzioni distinte sono importanti. La NLL è la somma di tutte le CE scelte divisa per tutti i loro token attivi. La preferenza è una media per coppia. In un'accumulazione di otto coppie di lunghezze diverse, ogni CE scelta condivide il denominatore totale dei token di queste otto coppie, mentre ogni penalità è divisa per otto. Una media di medie di CE cambierebbe l'obiettivo.
Lambda=0 deve ritrovare la loss e i gradienti SFT con queste stesse maschere e riduzioni. Il ramo SFT reale non calcola i rifiutati; il test lambda=0 può calcolarli per verificare il loro gradiente nullo. Risposte identiche danno una penalità relativa log(2), senza segnale relativo utile; la NLL scelta resta presente. Le coppie identiche vengono rifiutate nel corpus.
Evitare una sottrazione imprecisa
Calcolare direttamente log(1−exp(ell)) può dare log(0) vicino a zero: exp(ell) viene arrotondato a 1. Usiamo log1p(−exp(ell)) se ell < −log(2), e log(−expm1(ell)) nell'altro ramo. Solo il ramo utile viene valutato per ogni valore; valutare entrambi e poi selezionare con torch.where potrebbe lasciare un gradiente NaN proveniente dal ramo inattivo.
I due lati sono impilati prima di questa operazione comune. Questa simmetria consente ai gradienti opposti di una stessa voce di annullarsi esattamente nel punto in cui si incontrano, senza cambiare la formula né allentare il test.
Il dominio è esplicito: FP32 o FP64, ell finito strettamente negativo, con una magnitudine almeno uguale a finfo.tiny. Zero e le magnitudini subnormali vengono rifiutati, poiché la log-quota o la sua derivata non è rappresentabile nel contratto adottato. Non esiste alcun clamp silenzioso che crei un plateau. I test includono −1000, il confine −log(2), −1e−300 in FP64 e −1e−30 in FP32. I fallimenti della formula ingenua sono conservati con uno stato e null per i numeri non finiti; mai un falso risultato numerico JSON.
Il valore e i gradienti del termine di preferenza vengono confrontati con odds_ratio_loss di TRL v0.23.1, commit 4529a1c8b1813480a85b02c9c8e7f75a29085d65, nel dominio numerico ordinario: tolleranza assoluta 1e−12 in FP64 e 2e−6 in FP32. Questo estratto Apache-2.0 è fornito con provenienza e licenza; il trainer non è importato. La NLL mascherata viene confrontata separatamente con CrossEntropyLoss di PyTorch.
Questo confronto non riproduce l'intero trainer TRL: nella versione ispezionata, ricostruisce le label scelte da ID e attenzione, includendo il prompt. Il laboratorio usa esplicitamente label della sola risposta. È stato ispezionato anche il repository originale degli autori; le sue maschere e la conversione finale BF16 non sono riprese. Confrontiamo formula di preferenza e riduzioni dichiarate, senza rivendicare l'identità di un'intera infrastruttura di addestramento.
Tre apprendimenti e una base comune
Il minimo confronta la base acquisita, SFT chosen-only e ORPO lambda=0,1. Le due policy addestrate ripartono indipendentemente dallo stesso file instruct, senza SFT importato né avvio di un altro tutorial. Ogni policy vede gli stessi 840 prompt per due epoche, nello stesso ordine deterministico: 1680 presentazioni e 210 aggiornamenti, microbatch di una coppia e accumulazione di otto.
AdamW usa lr=0,000002, weight_decay=0,01, clipping a 1, warm-up di 21 aggiornamenti poi decrescita lineare fino al 10% del tasso massimo. Il seme è 47. I parametri completi e i gradienti FP32 sono controllati. La validazione è misurata dopo ogni epoca, senza selezionare un checkpoint o lambda sul test. L'ORPO tratta anche le risposte rifiutate; la stessa esposizione di prompt non significa lo stesso budget di token, forward o tempo.
Ogni fase comprende interruzione dopo un'epoca, addestramento continuo di due epoche e ripresa della seconda. Gli stati finali di modello, AdamW, sampler, scheduler, RNG Python/NumPy/PyTorch CPU/CUDA e cronologie devono essere esattamente uguali. I checkpoint completi sono serializzati localmente in gzip livello 1 senza perdita, con dimensioni realmente registrate, poi pubblicati tramite rinomina atomica. La loro compressione non riduce la VRAM e non garantisce un tasso di riduzione. Sono fidati perché prodotti localmente; non caricate al loro posto un oggetto Torch scaricato. I modelli finali sono conservati anche in Safetensors e ricaricati rigorosamente.
L'esercizio aggiunge una terza policy addestrata, chiamata sum-ablation, con gli stessi pesi iniziali, ordine dei prompt, ottimizzatore e aggiornamenti. Il suo termine di quote riceve le SOMME invece delle medie: è un'ablazione denominata, non l'ORPO di riferimento. I nove stati completi delle tre fasi restano locali. Le durate di ottimizzazione, validazione, comandi completi e picchi misurati sono distinti.
Valutare candidati e testi liberi
Ogni stato base/SFT/ORPO/sum-ablation valuta le stesse 180 coppie test. Il rapporto pubblica NLL scelta per token, margine di log-quote, penalità e accuracy con mezzo credito alle parità. I risultati sono separati secondo la lunghezza scelta più corta, uguale o più lunga in token. Per confrontare le policy, questa valutazione usa sempre le quote delle medie, anche sui pesi di sum-ablation; il nome del modello non modifica di nascosto la metrica.
Ogni stato genera anche 180 resoconti liberi, con il solo incidente nel prompt: greedy, una beam, al massimo 128 nuovi token, EOS=151645. ID, testo grezzo, ragione di arresto, lunghezza e durate restano disponibili. Una risposta fermata al limite è un fallimento anche se il suo inizio pare completo. Nessun output viene inviato né applicato a un sistema reale.
La rubrica è una grammatica indipendente dalla generazione. Riconosce le clausole controllate di unità/stato, pressione e causa. Ogni frase non riconosciuta, campo ripetuto, tag o arresto al limite rende la nota incompleta. Una parafrasi naturale corretta fuori grammatica può dunque fallire. Il punteggio non è un giudice generale di fedeltà né una misura di soddisfazione.
L'esattezza richiede i quattro campi corretti su TUTTI i 180 casi. Cause note e sconosciute hanno denominatori separati. Una causa riconosciuta diversa dalla rilevazione è conteggiata come non comprovata. Un'astensione riconosciuta su una causa nota viene conteggiata a parte; un campo assente in una prosa incomprensibile non è automaticamente chiamato astensione. Una pressione riconosciuta quando non esiste alcuna misura viene segnalata. Le frasi non riconosciute restano non certificate, anche se questi contatori parziali sono zero. I fallimenti non vengono mai rimossi dai denominatori.
Esercizio sulla lunghezza
Le prime righe dell'esercizio sono analitiche: log-probabilità costanti −1 per i token scelti e −1,2 per quelli rifiutati, con lunghezze 4, 16 e 64. Medie, somme, quote e gradienti sono confrontati per le nove coppie di lunghezze. Con le medie, la qualità media costante non dipende dalla lunghezza; con le somme, una risposta scelta lunga può perdere contro una risposta rifiutata corta. Non è una misura su testi aziendali.
Poi i punteggi reali delle 180 coppie della policy ORPO principale sono sottoposti alle due formule senza modificare i pesi. Infine, la policy sum-ablation viene davvero addestrata e valutata in generazione libera. I suoi risultati sono separati dall'esperimento analitico e dal semplice ricalcolo della formula. Gli SHA-256 di modelli, risultati principali e configurazione minima sono protetti; la versione principale conserva le medie.
Una loss numericamente stabile può imparare cattive preferenze. Un margine positivo su due candidati non garantisce un buon nuovo testo. Un solo seme e modelli sintetici limitano fortemente le conclusioni; non è richiesto alcun miglioramento di ORPO su SFT.
Definire preferenze senza una parola rivelatrice
Osservazione e causalità
Errore frequente: tutti i chosen dicono 'unconfirmed'; aggiungete cause confermate e parafrasi senza modificare il test dopo l'esecuzione.
Invariante: 840/180/180 coppie, nessuna fuga di incidente, diversità dei motivi.
File src/orpo_sans_reference/data.py
import hashlib, re
from pathlib import Path
from .config import save_json, sha
CAUSES = ["filter blockage", "power loss", "sensor fault"]
COURTESY = [
"This is the recorded incident note.",
"Further details may be documented separately.",
]
def split_incident_families(family):
return "train" if family < 210 else "validation" if family < 255 else "test"
def facts_for(family):
return {
"unit": f"P{family+1000}",
"state": "stopped" if family % 3 else "running",
"pressure": None if family % 4 < 2 else 10 + (family // 4) % 7,
"cause": None if family % 2 == 0 else CAUSES[(family // 2) % 3],
}
def incident_text(family, j, facts):
previous = "running" if facts["state"] == "stopped" else "stopped"
old_cause = CAUSES[(family // 2 + 1) % 3]
pressure = (
"not measured" if facts["pressure"] is None else str(facts["pressure"]) + " kPa"
)
cause = "not confirmed" if facts["cause"] is None else "confirmed " + facts["cause"]
return f"Incident N{family:03d}-{j}. Superseded note: state {previous}; suggested cause {old_cause}. Current verified record: unit {facts['unit']}; state {facts['state']}; pressure {pressure}; cause {cause}. Use only the current record."
def reply(facts, j, courtesy=False):
first = [
f"{facts['unit']} is {facts['state']}.",
f"Unit {facts['unit']} is {facts['state']}.",
f"The state of {facts['unit']} is {facts['state']}.",
f"The current state of {facts['unit']} is {facts['state']}.",
][j]
pressure = (
f"The pressure is {facts['pressure']} kPa."
if facts["pressure"] is not None
else ("Pressure was not measured." if j % 2 else "The pressure is unknown.")
)
cause = (
f"The confirmed cause is {facts['cause']}."
if facts["cause"] is not None
else (
"The cause remains unconfirmed."
if j % 2
else "The confirmed cause is not known."
)
)
clauses = [first, pressure, cause]
if courtesy:
clauses = COURTESY[:1] + clauses + COURTESY[1:]
return " ".join(clauses)
def audit_causal_labels(labels):
n = len(labels)
lengths = {"chosen_shorter": 0, "same_words": 0, "chosen_longer": 0}
known = sum(r["expected"]["cause"] is not None for r in labels)
features = []
words = sorted(
{
w
for r in labels
for side in ["chosen", "rejected"]
for w in re.findall("[a-z]+", r[side].lower())
}
)
for row in labels:
a, b = len(row["chosen"].split()), len(row["rejected"].split())
lengths[
"chosen_shorter" if a < b else "chosen_longer" if a > b else "same_words"
] += 1
for word in words:
a, b = [
sum(word in set(re.findall("[a-z]+", r[side].lower())) for r in labels)
for side in ["chosen", "rejected"]
]
accuracy = (a + n - b) / (2 * n)
features.append(
{
"word": word,
"chosen_presence": a,
"rejected_presence": b,
"best_binary_label_accuracy": max(accuracy, 1 - accuracy),
}
)
return {
"pairs": n,
"known_cause": known,
"unknown_cause": n - known,
"unknown_pressure": sum(r["expected"]["pressure"] is None for r in labels),
"word_length_regimes": lengths,
"best_length_preference_accuracy_half_ties": max(
(lengths["chosen_longer"] + 0.5 * lengths["same_words"]) / n,
(lengths["chosen_shorter"] + 0.5 * lengths["same_words"]) / n,
),
"word_features": features,
"maximum_word_accuracy": max(r["best_binary_label_accuracy"] for r in features),
"scope": "Original words, not every possible semantic/BPE shortcut",
}
def generate_incident_pairs():
pages = []
labels = []
for family in range(300):
facts = facts_for(family)
for j in range(4):
identifier = f"N{family:03d}-{j}"
text = incident_text(family, j, facts)
wrong = dict(facts)
reason = ["cause", "pressure", "state", "cause"][j]
if reason == "cause":
wrong["cause"] = (
CAUSES[(family // 2) % 3] if facts["cause"] is None else None
)
elif reason == "pressure":
wrong["pressure"] = (
10 + (family // 4) % 7 if facts["pressure"] is None else None
)
else:
wrong["state"] = "running" if facts["state"] == "stopped" else "stopped"
chosen = reply(facts, j, courtesy=j == 1)
rejected = reply(wrong, j, courtesy=j == 2)
if chosen == rejected:
raise ValueError("Preference pair is identical")
pages.append(
{
"document_id": identifier,
"family": family,
"formulation": j,
"split": split_incident_families(family),
"incident": text,
"record": facts,
"content_sha256": hashlib.sha256(text.encode()).hexdigest(),
}
)
labels.append(
{
"document_id": identifier,
"chosen": chosen,
"rejected": rejected,
"expected": facts,
"preference_reason": reason,
"simulated_preference": True,
}
)
if len({p["incident"] for p in pages}) != 1200:
raise ValueError("Duplicate incident")
return pages, labels
def prepare(config):
root = Path("data/generated")
if (root / "manifest.json").exists():
raise ValueError("Preserve existing corpus; use a new extraction")
pages, labels = generate_incident_pairs()
gold = {r["document_id"]: r for r in labels}
save_json(root / "input/pages.json", pages)
save_json(root / "labels/gold.json", labels)
counts = {}
audits = {}
for split in ["train", "validation", "test"]:
selected = [p for p in pages if p["split"] == split]
chosen = [gold[p["document_id"]] for p in selected]
counts[split] = {
"pairs": len(selected),
"families": len({p["family"] for p in selected}),
}
audits[split] = audit_causal_labels(chosen)
if split != "test":
save_json(root / f"{split}/pages.json", selected)
save_json(root / f"labels/{split}.json", chosen)
save_json(
"outputs/data-audit.json",
{
"dataset": "IncidentNotesPrefs-v1",
"rights": "CC0-1.0",
"counts": counts,
"shortcuts": audits,
"human_preferences": False,
"shared_semantic_and_language_primitives": True,
"current_record_supersedes_previous_note": True,
"reference_model_or_cache": False,
},
)
save_json(
root / "manifest.json",
{
"dataset": "IncidentNotesPrefs-v1",
"seed": config["seed"],
"counts": counts,
"files": {
str(f.relative_to(root)): sha(f) for f in root.rglob("*") if f.is_file()
},
},
)
return counts
Calcolare le medie dei token
Maschere, lunghezza e punteggio q
Errore frequente: un batch medio pondera per errore ogni frase; nominate separatamente media token e media coppie.
Invariante: PAD invarianti, risposte non vuote, riduzione controllata.
File src/orpo_sans_reference/scores.py
import torch
from .config import save_json
SYSTEM = "Write an incident note using only the current verified record. State the unit and its current state, the pressure if measured, and the confirmed cause if known. Distinguish observations from causal conclusions. A superseded suggestion is not a current fact. Do not add facts, actions or explanations."
def format_prompt(tokenizer, incident, version):
if version != "incident-v1":
raise ValueError("Unknown incident template")
messages = [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": incident},
]
return messages, tokenizer.apply_chat_template(
messages, tokenize=True, add_generation_prompt=True
)
def completion_row(tokenizer, incident, response, config, identifier, version=None):
version = version or config["template_version"]
messages, prompt = format_prompt(tokenizer, incident, 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["incident"],
label["chosen"],
config,
page["document_id"],
version,
)
rejected = completion_row(
tokenizer,
page["incident"],
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 mean_completion_logprob(logits, labels):
sums, counts, values, selected, targets = sequence_logprob(logits, labels)
return sums / counts, counts, values, selected, targets
def chosen_token_nll(logits, labels):
sums, counts, _, _, _ = sequence_logprob(logits, labels)
return -sums.sum() / counts.sum()
def inspect_pair_lengths(config):
from .model import load_tokenizer
from .data import generate_incident_pairs
tokenizer = load_tokenizer()
pages, labels = generate_incident_pairs()
pairs = [
paired_completion_batch(tokenizer, pages[i], labels[i], config)
for i in range(4)
]
save_json(
"outputs/tokenization-mechanism.json",
{
"pairs": pairs,
"preference_reduction": "mean_response_logprob_including_EOS",
"nll_reduction": "all_chosen_active_tokens",
"prompt_and_padding_excluded": True,
},
)
Codificare le quote in modo stabile
log1mexp e sigmoid
Errore frequente: un clamp silenzioso crea un plateau del gradiente; rendetelo esplicito e testatene la zona d'azione.
Invariante: test FP64/FP32 e concordanza di riferimento secondo tolleranze prefissate.
File src/orpo_sans_reference/loss.py
import math
from types import SimpleNamespace
import torch
from .config import save_json
def log1mexp(a):
if a.dtype not in [torch.float32, torch.float64] or not torch.isfinite(a).all():
raise ValueError("Use finite FP32/FP64 log probabilities")
if (a >= 0).any() or (a.abs() < torch.finfo(a.dtype).tiny).any():
raise ValueError(
"Log probability must be strictly negative with a representable finite derivative; no silent clamp"
)
# Compute only the selected branch: torch.where on two evaluated formulas
# could propagate NaN gradients from its inactive near-zero branch.
far = a < -math.log(2)
result = torch.empty_like(a)
result[far] = torch.log1p(-torch.exp(a[far]))
result[~far] = torch.log(-torch.expm1(a[~far]))
return result
def orpo_preference_loss(chosen_logps, rejected_logps):
if chosen_logps.shape != rejected_logps.shape or not chosen_logps.numel():
raise ValueError("Matching nonempty paired score vectors required")
paired = torch.stack([chosen_logps, rejected_logps])
odds = paired - log1mexp(paired)
margin = odds[0] - odds[1]
return -torch.nn.functional.logsigmoid(margin), margin
def orpo_total_loss(
chosen_sums,
rejected_sums,
chosen_counts,
rejected_counts,
coefficient=0.1,
normalization="mean",
):
if (
coefficient < 0
or not math.isfinite(coefficient)
or normalization not in ["mean", "sum"]
):
raise ValueError("Explicit nonnegative lambda and named normalization required")
if (chosen_counts <= 0).any() or (rejected_counts <= 0).any():
raise ValueError("Responses must have active tokens")
chosen = chosen_sums / chosen_counts if normalization == "mean" else chosen_sums
rejected = (
rejected_sums / rejected_counts if normalization == "mean" else rejected_sums
)
nll = -chosen_sums.sum() / chosen_counts.sum()
preference, margins = orpo_preference_loss(chosen, rejected)
return nll + coefficient * preference.mean(), nll, preference, margins
def numeric_mechanism(config):
from .reference_loss import odds_ratio_loss
reference_rows = []
for dtype, tolerance in [(torch.float64, 1e-12), (torch.float32, 2e-6)]:
chosen = torch.tensor([-1.0, -3.0, -0.1], dtype=dtype, requires_grad=True)
rejected = torch.tensor([-2.0, -1.0, -0.5], dtype=dtype, requires_grad=True)
loss, margin = orpo_preference_loss(chosen, rejected)
control = SimpleNamespace(
beta=config["lambda_orpo"], accelerator=SimpleNamespace(device="cpu")
)
reference = -odds_ratio_loss(control, chosen, rejected)[0]
a = torch.autograd.grad(
config["lambda_orpo"] * loss.mean(), (chosen, rejected), retain_graph=True
)
b = torch.autograd.grad(reference.mean(), (chosen, rejected))
checks = [
torch.allclose(
config["lambda_orpo"] * loss, reference, atol=tolerance, rtol=0
),
*(torch.allclose(x, y, atol=tolerance, rtol=0) for x, y in zip(a, b)),
]
if not all(checks):
raise ValueError("Pinned ORPO reference value/gradient comparison failed")
reference_rows.append(
{
"dtype": str(dtype),
"chosen_mean_logps": chosen.tolist(),
"rejected_mean_logps": rejected.tolist(),
"preference_losses": loss.tolist(),
"log_odds_margins": margin.tolist(),
"weighted_reference_losses": reference.tolist(),
"chosen_gradient": a[0].tolist(),
"rejected_gradient": a[1].tolist(),
"reference_chosen_gradient": b[0].tolist(),
"reference_rejected_gradient": b[1].tolist(),
"lambda": config["lambda_orpo"],
"atol": tolerance,
}
)
stability = []
for dtype, values in [
(
torch.float64,
[-1000.0, -100.0, -1.0, -math.log(2), -0.1, -1e-10, -1e-100, -1e-300],
),
(
torch.float32,
[-1000.0, -100.0, -1.0, -math.log(2), -0.1, -1e-5, -1e-10, -1e-30],
),
]:
x = torch.tensor(values, dtype=dtype, requires_grad=True)
stable = log1mexp(x)
gradient = torch.autograd.grad(stable.sum(), x)[0]
naive = torch.log(1 - torch.exp(x))
naive_grad = torch.autograd.grad(naive.sum(), x)[0]
if not torch.isfinite(stable).all() or not torch.isfinite(gradient).all():
raise ValueError("Stable numerical branch is not finite")
for a, s, g, n, ng in zip(
x.tolist(),
stable.tolist(),
gradient.tolist(),
naive.tolist(),
naive_grad.tolist(),
):
expected = (
math.log1p(-math.exp(a))
if a < -math.log(2)
else math.log(-math.expm1(a))
)
stability.append(
{
"dtype": str(dtype),
"log_probability": a,
"stable": s,
"stable_gradient": g,
"analytic_float64_value": expected,
"absolute_error": abs(s - expected),
"naive": n if math.isfinite(n) else None,
"naive_gradient": ng if math.isfinite(ng) else None,
"naive_finite": math.isfinite(n) and math.isfinite(ng),
}
)
save_json(
"outputs/numeric-mechanism.json",
{
"reference": reference_rows,
"stability": stability,
"data_source": "Controlled analytical log probabilities, not incident quality measurements",
"zero_or_subnormal_magnitude_policy": "Explicit rejection, no clamp",
"reference_model_loaded": False,
},
)
Addestrare SFT e ORPO dalla stessa base
Due obiettivi comparabili
Errore frequente: copiare una configurazione DPO importa una ref inutilizzata; sottoponete ad audit oggetti e allocazioni.
Invariante: aggiornamenti reali, checkpoint e ripresa verificati; nessun modello ref creato.
File src/orpo_sans_reference/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,
parameter_manifest,
manifest_digest,
save_learned,
)
from .scores import paired_completion_batch, model_scores
from .loss import orpo_preference_loss
from .state import save_resume_state, load_resume_state, equal
BRANCHES = ["sft", "orpo", "sum-ablation"]
def learning_rate(config, step, total=210):
peak = 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_split(config, tokenizer, split):
pages = read_json(f"data/generated/{split}/pages.json")
gold = {
r["document_id"]: r for r in read_json(f"data/generated/labels/{split}.json")
}
if any(p["split"] != split for p in pages):
raise ValueError("Family partition differs")
return [
paired_completion_batch(tokenizer, p, gold[p["document_id"]], config)
for p in pages
]
@torch.no_grad()
def validate_policy(model, pairs):
model.eval()
rows = []
total_nll = 0.0
total_tokens = 0
wins = 0.0
for pair in pairs:
chosen, nc, *_ = model_scores(model, [pair["chosen"]])
rejected, nr, *_ = model_scores(model, [pair["rejected"]])
a = chosen / nc
b = rejected / nr
preference, margin = orpo_preference_loss(a, b)
total_nll -= float(chosen[0])
total_tokens += int(nc[0])
wins += 1 if float(margin[0]) > 0 else 0.5 if float(margin[0]) == 0 else 0
rows.append(
{
"document_id": pair["document_id"],
"chosen_sum": float(chosen[0]),
"rejected_sum": float(rejected[0]),
"chosen_tokens": int(nc[0]),
"rejected_tokens": int(nr[0]),
"chosen_mean": float(a[0]),
"rejected_mean": float(b[0]),
"mean_log_odds_margin": float(margin[0]),
"mean_odds_preference_loss": float(preference[0]),
}
)
model.train()
return {
"pairs": len(pairs),
"chosen_tokens": total_tokens,
"chosen_nll_nats_per_token": total_nll / total_tokens,
"mean_odds_accuracy_half_ties": wins / len(pairs),
"rows": rows,
"used_for_checkpoint_selection": False,
"evaluation_normalization": "mean_even_for_sum_ablation",
}
def train_policy(config, branch="orpo", mode="continuous"):
if branch not in BRANCHES or mode not in ["interrupt", "continuous", "resume"]:
raise ValueError("Unknown training branch/mode")
normalization = "sum" if branch == "sum-ablation" else "mean"
if config.get("normalization", "mean") != normalization:
raise ValueError("Sum ablation requires its explicit exercise config")
if (
Path("data/generated/input").exists()
or Path("data/generated/labels/gold.json").exists()
or Path("assets/reference").exists()
or Path("outputs/reference-cache").exists()
or Path("outputs/inference").exists()
):
raise ValueError(
"Learning must not mount heldout outputs, labels or any reference model/cache"
)
seed_all(config["seed"])
tokenizer = load_tokenizer()
rows = encoded_split(config, tokenizer, "train")
validation = encoded_split(config, tokenizer, "validation")
if len(rows) != 840 or len(validation) != 180:
raise ValueError("Frozen population differs")
model = load_base().cuda()
parameters = list(model.parameters())
initial = parameter_manifest(model)
initial_digest = manifest_digest(initial)
if (
sum(p.numel() for p in parameters) != 494032768
or any(not p.requires_grad or p.dtype != torch.float32 for p in parameters)
or len({id(p) for p in parameters}) != len(parameters)
):
raise ValueError("All unique FP32 policy parameters must be optimized")
identity = {
"branch": branch,
"normalization": normalization,
"lambda": 0.0 if branch == "sft" else config["lambda_orpo"],
"config": config,
"initial_model_digest": initial_digest,
"model_sha256": sha("assets/model/model.safetensors"),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"implementation_sha256": {
name: sha("src/orpo_sans_reference/" + name)
for name in ["model.py", "scores.py", "loss.py", "train.py", "state.py"]
},
"data_sha256": {
name: sha("data/generated/" + name)
for name in [
"train/pages.json",
"labels/train.json",
"validation/pages.json",
"labels/validation.json",
]
},
"reference_models": 0,
"reference_caches": 0,
}
optimizer = torch.optim.AdamW(
parameters, lr=config["learning_rate"], weight_decay=config["weight_decay"]
)
sampler = torch.Generator().manual_seed(config["seed"] + 100)
per_epoch = math.ceil(len(rows) / config["accumulation_pairs"])
total = per_epoch * config["epochs"]
root = Path("assets/checkpoints")
destination = root / f"{branch}-{mode}.pt.gz"
updates = start = 0
history = []
validation_history = []
if destination.exists():
raise ValueError("Preserve existing local checkpoint")
if mode == "resume":
state = load_resume_state(root / f"{branch}-interrupt.pt.gz")
if state["identity"] != identity:
raise ValueError("Resume identity changed, including lambda/normalization")
model.load_state_dict(state["model"], strict=True)
optimizer.load_state_dict(state["optimizer"])
sampler.set_state(state["sampler"])
restore_rng(state["rng"])
updates = state["updates"]
start = state["epoch"]
history = state["history"]
validation_history = state["validation_history"]
if state["scheduler"] != {
"next_update": updates,
"next_lr": learning_rate(config, updates, total),
}:
raise ValueError("Scheduler identity changed")
del state
stop = config["interrupt_epoch"] if mode == "interrupt" else config["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):
order = torch.randperm(len(rows), generator=sampler).tolist()
torch.cuda.synchronize()
epoch_begun = time.perf_counter()
for offset in range(0, len(rows), config["accumulation_pairs"]):
selected = order[offset : offset + config["accumulation_pairs"]]
chosen_denominator = sum(
rows[i]["chosen"]["response_tokens"] for i in selected
)
lr = learning_rate(config, updates, total)
for group in optimizer.param_groups:
group["lr"] = lr
optimizer.zero_grad(set_to_none=True)
details = []
nll_numerator = preference_numerator = 0.0
policy_tokens = 0
for index in selected:
pair = rows[index]
chosen, nc, *_ = model_scores(model, [pair["chosen"]])
nll = -chosen.sum()
contribution = nll / chosen_denominator
policy_tokens += int(nc[0])
detail = {
"document_id": pair["document_id"],
"chosen_sum": float(chosen.detach()[0]),
"chosen_tokens": int(nc[0]),
"rejected_sum": None,
"rejected_tokens": 0,
"preference_loss": None,
"log_odds_margin": None,
"preference_chosen_logp": None,
"preference_rejected_logp": None,
}
if branch != "sft":
rejected, nr, *_ = model_scores(model, [pair["rejected"]])
a = chosen / nc if normalization == "mean" else chosen
b = rejected / nr if normalization == "mean" else rejected
preference, margin = orpo_preference_loss(a, b)
contribution = contribution + config[
"lambda_orpo"
] * preference.sum() / len(selected)
preference_numerator += float(preference.detach()[0])
policy_tokens += int(nr[0])
detail.update(
rejected_sum=float(rejected.detach()[0]),
rejected_tokens=int(nr[0]),
preference_loss=float(preference.detach()[0]),
log_odds_margin=float(margin.detach()[0]),
preference_chosen_logp=float(a.detach()[0]),
preference_rejected_logp=float(b.detach()[0]),
)
if not torch.isfinite(contribution):
raise ValueError("Nonfinite objective")
contribution.backward()
nll_numerator += float(nll.detach())
details.append(detail)
norm = torch.nn.utils.clip_grad_norm_(
parameters, config["gradient_clip"], error_if_nonfinite=True
)
if any(p.grad is None or p.grad.dtype != torch.float32 for p in parameters):
raise ValueError("All policy gradients must exist in FP32")
optimizer.step()
updates += 1
nll = nll_numerator / chosen_denominator
preference = (
preference_numerator / len(selected) if branch != "sft" else 0.0
)
history.append(
{
"update": updates,
"epoch": epoch + 1,
"document_ids": [rows[i]["document_id"] for i in selected],
"chosen_token_denominator": chosen_denominator,
"preference_pair_denominator": (
len(selected) if branch != "sft" else 0
),
"policy_response_tokens": policy_tokens,
"chosen_nll": nll,
"preference_loss": preference,
"total_loss": nll + identity["lambda"] * preference,
"learning_rate": lr,
"gradient_norm_before_clip": float(norm),
"pairs": 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_begun
observed = validate_policy(model, validation)
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 real learned parameter change")
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, updates, total),
},
"rng": rng_state(),
"history": history,
"validation_history": validation_history,
}
compression = save_resume_state(destination, state)
if mode == "resume":
save_learned(model, 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,
"checkpoint_sha256": sha(destination),
"checkpoint_bytes": destination.stat().st_size,
"checkpoint_compression": compression,
"optimization_seconds_this_process": optimization_seconds,
"training_and_validation_seconds_this_process": elapsed,
"max_torch_allocated_bytes": peak,
"test_visible": False,
"reference_models_loaded": 0,
"policy_models_loaded": 1,
},
)
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/resumed full states differ: " + str(checks))
save_json(
f"outputs/learning/resume-{branch}.json",
{
"status": "PASS",
"all_state_exact": True,
"checks": checks,
"boundary": "Full epoch; model/AdamW/scheduler/sampler/RNG/history compared exactly after lossless decompression",
},
)
Osservare cause inventate e astensioni
Qualità della generazione libera
Errore frequente: valutare solo le cause sconosciute premia l'astensione sistematica; conservate i due strati.
Invariante: 180 output per ramo, cause note/sconosciute presentate separatamente.
File src/orpo_sans_reference/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_base,
load_tokenizer,
reload_learned,
parameter_manifest,
manifest_digest,
)
from .scores import format_prompt, paired_completion_batch, model_scores
from .loss import orpo_preference_loss
from .data import COURTESY
def parse_note(text, finish):
sentences = [
s.strip() for s in re.split(r"(?<=[.!?])\s+", text.strip()) if s.strip()
]
fields = {}
unknown = []
duplicates = []
for sentence in sentences:
if sentence in COURTESY:
continue
value = sentence.rstrip(".!?").strip()
items = None
match = re.fullmatch(r"(?:Unit )?(P\d{4}) is (stopped|running)", value, re.I)
if match:
items = {"unit": match[1].upper(), "state": match[2].lower()}
else:
match = re.fullmatch(
r"The (?:current )?state of (P\d{4}) is (stopped|running)", value, re.I
)
if match:
items = {"unit": match[1].upper(), "state": match[2].lower()}
else:
match = re.fullmatch(r"The pressure is (\d{1,4}) kPa", value, re.I)
if match:
items = {"pressure": int(match[1])}
elif value.lower() in [
"pressure was not measured",
"the pressure is unknown",
]:
items = {"pressure": None}
elif value.lower() in [
"the cause remains unconfirmed",
"no cause has been confirmed",
"the confirmed cause is not known",
]:
items = {"cause": None}
else:
match = re.fullmatch(
r"The confirmed cause is (filter blockage|power loss|sensor fault)",
value,
re.I,
)
if match:
items = {"cause": match[1].lower()}
if items is None:
unknown.append(sentence)
else:
for key, item in items.items():
if key in fields:
duplicates.append(key)
else:
fields[key] = item
covered = bool(sentences) and not unknown and not duplicates
return {
"recognized": fields,
"unrecognized_sentences": unknown,
"duplicate_fields": duplicates,
"all_text_covered": covered,
"complete": finish == "eos"
and covered
and set(fields) == {"unit", "state", "pressure", "cause"},
"truncated": finish != "eos",
}
def score_causal_claims(rows, gold):
metrics = {
"documents": len(rows),
"complete_controlled_notes": 0,
"exact_facts": 0,
"truncated": 0,
"unrecognized_prose": 0,
"known_cause_cases": 0,
"unknown_cause_cases": 0,
"known_cause_correct": 0,
"unknown_cause_correct": 0,
"recognized_unsupported_cause": 0,
"recognized_known_cause_abstention": 0,
"unknown_pressure_cases": 0,
"recognized_invented_pressure": 0,
}
fields = dict.fromkeys(["unit", "state", "pressure", "cause"], 0)
details = []
for row in rows:
expected = gold[row["document_id"]]["expected"]
parsed = parse_note(row["text"], row["finish_reason"])
if row["parsed"] != parsed:
raise ValueError("Stored parser result differs")
actual = parsed["recognized"]
valid = parsed["complete"]
correct = {k: valid and actual.get(k) == v for k, v in expected.items()}
exact = all(correct.values())
metrics["complete_controlled_notes"] += valid
metrics["exact_facts"] += exact
metrics["truncated"] += parsed["truncated"]
metrics["unrecognized_prose"] += bool(parsed["unrecognized_sentences"])
for key, value in correct.items():
fields[key] += value
known = expected["cause"] is not None
metrics["known_cause_cases" if known else "unknown_cause_cases"] += 1
metrics["known_cause_correct" if known else "unknown_cause_correct"] += correct[
"cause"
]
metrics["recognized_unsupported_cause"] += (
actual.get("cause") is not None and actual["cause"] != expected["cause"]
)
metrics["recognized_known_cause_abstention"] += (
known and "cause" in actual and actual["cause"] is None
)
if expected["pressure"] is None:
metrics["unknown_pressure_cases"] += 1
metrics["recognized_invented_pressure"] += (
actual.get("pressure") is not None
)
details.append(
{
"document_id": row["document_id"],
"expected": expected,
"recognized": actual,
"complete": valid,
"fields_correct": correct,
"exact": exact,
"unrecognized_sentences": parsed["unrecognized_sentences"],
}
)
n = len(rows)
metrics.update(
exact_accuracy_over_all_cases=metrics["exact_facts"] / n,
field_correct_over_all_cases=fields,
field_accuracy_over_all_cases={k: v / n for k, v in fields.items()},
known_cause_accuracy=metrics["known_cause_correct"]
/ metrics["known_cause_cases"],
unknown_cause_accuracy=metrics["unknown_cause_correct"]
/ metrics["unknown_cause_cases"],
recognized_known_cause_abstention_rate=metrics[
"recognized_known_cause_abstention"
]
/ metrics["known_cause_cases"],
mean_generated_tokens=math.fsum(r["output_tokens"] for r in rows) / n,
)
return {
"metrics": metrics,
"rows": details,
"unrecognized_prose_is_not_certified_faithful": True,
}
def pair_metrics(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)
count = sum(r["scores"]["chosen"]["response_tokens"] for r in selected)
result[group] = {
"pairs": n,
"chosen_response_tokens": count,
"chosen_nll_nats_per_token": (
-math.fsum(r["scores"]["chosen"]["sum_logprob"] for r in selected)
/ count
if count
else None
),
"mean_log_odds_margin": (
math.fsum(r["mean_log_odds_margin"] for r in selected) / n
if n
else None
),
"mean_odds_preference_accuracy_half_ties": (
sum(
(
1
if r["mean_log_odds_margin"] > 0
else 0.5 if r["mean_log_odds_margin"] == 0 else 0
)
for r in selected
)
/ n
if n
else None
),
"mean_odds_preference_loss": (
math.fsum(r["mean_odds_preference_loss"] for r in selected) / n
if n
else None
),
}
return result
def load_policy(branch):
if branch not in ["base", "sft", "orpo", "sum-ablation"]:
raise ValueError("Unknown evaluation branch")
path = Path(
"assets/model/model.safetensors"
if branch == "base"
else f"assets/checkpoints/{branch}-resume.safetensors"
)
model = load_base() if branch == "base" else reload_learned(path)[0]
digest = manifest_digest(parameter_manifest(model))
for parameter in model.parameters():
parameter.requires_grad_(False)
return model.cuda().eval(), {
"branch": branch,
"file_sha256": sha(path),
"model_digest": digest,
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"reference_models_loaded": 0,
}
@torch.no_grad()
def score_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")}
rows = []
probes = []
torch.cuda.reset_peak_memory_stats()
torch.cuda.synchronize()
begun = time.perf_counter()
for index, page in enumerate(pages):
pair = paired_completion_batch(
tokenizer, page, gold[page["document_id"]], config
)
scores = {}
mean_tensors = {}
for side in ["chosen", "rejected"]:
sums, counts, values, logits, targets = model_scores(model, [pair[side]])
mean_tensors[side] = sums / counts
scores[side] = {
"sum_logprob": float(sums[0]),
"mean_logprob": float(mean_tensors[side][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]),
}
)
loss, margin = orpo_preference_loss(
mean_tensors["chosen"], mean_tensors["rejected"]
)
rows.append(
{
"document_id": page["document_id"],
"pair": pair,
"scores": scores,
"response_length_delta": pair["response_length_delta"],
"mean_log_odds_margin": float(margin[0]),
"mean_odds_preference_loss": float(loss[0]),
}
)
torch.cuda.synchronize()
save_json(
f"outputs/evaluation/pairs-{branch}.json",
{
"branch": branch,
"identity": identity,
"rows": rows,
"probes": probes,
"metrics": pair_metrics(rows),
"seconds": time.perf_counter() - begun,
"max_torch_allocated_bytes": torch.cuda.max_memory_allocated(),
"evaluation_normalization": "mean_even_for_sum_ablation",
"test_selection": False,
},
)
@torch.no_grad()
def generate_notes(config, branch):
if (
Path("data/generated/labels").exists()
or Path("assets/reference").exists()
or Path("outputs/reference-cache").exists()
):
raise ValueError("Blind generation must not mount labels or a reference")
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["incident"], 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_note(text, finish),
"input_tokens": len(ids),
"output_tokens": len(output),
"seconds": time.perf_counter() - started,
}
)
if (index + 1) % 30 == 0:
print({"branch": branch, "generated_notes": 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_visible": False,
"reference_models_loaded": 0,
"training": False,
},
)
def evaluate_notes(config, exercise=False):
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 ["sum-ablation"] if exercise else ["base", "sft", "orpo"]:
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
or predictions["identity"] != pairs["identity"]
):
raise ValueError("Same180testcases and model weights required")
results[branch] = {
"free_notes": score_causal_claims(predictions["rows"], gold),
"pairs": pair_metrics(pairs["rows"]),
}
print(
{
"branch": branch,
"free_notes": results[branch]["free_notes"]["metrics"],
"pairs": results[branch]["pairs"],
},
flush=True,
)
save_json(
(
"outputs/evaluation/exercise-comparison.json"
if exercise
else "outputs/evaluation/comparison.json"
),
{
"results": results,
"lambda_selected_before_training": config["lambda_orpo"],
"human_preferences": False,
"quality_gain_required": False,
"checkpoint_selection": False,
"reference_models_loaded": 0,
"same_prompt_presentations_not_same_compute": True,
"exercise_sum_is_not_reference_ORPO": exercise,
"no_real_incident_action": True,
},
)
Isolare l'effetto della lunghezza
Ablazione di normalizzazione
Errore frequente: mescolare dati analitici e risultati addestrati nello stesso grafico senza legenda; separate le fonti.
Invariante: le due varianti sono nominate e l'ORPO finale conserva la media.
File src/orpo_sans_reference/exercise.py
from pathlib import Path
import torch
from .config import read_json, save_json, sha
from .loss import orpo_preference_loss
def main_artifacts():
names = [
"configs/minimum.yaml",
"outputs/evaluation/comparison.json",
"outputs/numeric-mechanism.json",
]
for branch in ["base", "sft", "orpo"]:
names += [
f"outputs/evaluation/pairs-{branch}.json",
f"outputs/inference/predictions-{branch}.json",
]
if branch != "base":
names += [
f"assets/checkpoints/{branch}-resume.safetensors",
f"outputs/learning/resume-{branch}.json",
]
return names
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()})
def compare_sum_mean_odds():
rows = []
for chosen_length in [4, 16, 64]:
for rejected_length in [4, 16, 64]:
for normalization in ["mean", "sum"]:
a = torch.full(
(chosen_length,), -1.0, dtype=torch.float64, requires_grad=True
)
b = torch.full(
(rejected_length,), -1.2, dtype=torch.float64, requires_grad=True
)
pa = a.mean() if normalization == "mean" else a.sum()
pb = b.mean() if normalization == "mean" else b.sum()
loss, margin = orpo_preference_loss(pa.reshape(1), pb.reshape(1))
ga, gb = torch.autograd.grad(loss.sum(), (a, b))
rows.append(
{
"normalization": normalization,
"chosen_length": chosen_length,
"rejected_length": rejected_length,
"chosen_token_logprob": -1.0,
"rejected_token_logprob": -1.2,
"chosen_score": float(pa.detach()),
"rejected_score": float(pb.detach()),
"log_odds_margin": float(margin.detach()[0]),
"preference_loss": float(loss.detach()[0]),
"chosen_token_gradients": ga.tolist(),
"rejected_token_gradients": gb.tolist(),
}
)
return rows
def run_length_ablation(config):
if config.get("normalization") != "sum":
raise ValueError("Exercise must identify the sum ablation")
protected = read_json("outputs/exercise/protected.json")
if protected != {name: sha(name) for name in main_artifacts()}:
raise ValueError("Exercise changed main weights/results/config")
actual = read_json("outputs/evaluation/pairs-orpo.json")
rescored = []
for row in actual["rows"]:
a = torch.tensor(
[row["scores"]["chosen"]["sum_logprob"]], dtype=torch.float32, device="cuda"
)
b = torch.tensor(
[row["scores"]["rejected"]["sum_logprob"]],
dtype=torch.float32,
device="cuda",
)
na = row["scores"]["chosen"]["response_tokens"]
nb = row["scores"]["rejected"]["response_tokens"]
mean, mean_margin = orpo_preference_loss(a / na, b / nb)
summed, sum_margin = orpo_preference_loss(a, b)
if (
float(mean[0]) != row["mean_odds_preference_loss"]
or float(mean_margin[0]) != row["mean_log_odds_margin"]
):
raise ValueError("Mean scoring changed in the exercise")
rescored.append(
{
"document_id": row["document_id"],
"chosen_tokens": na,
"rejected_tokens": nb,
"mean_log_odds_margin": float(mean_margin[0]),
"sum_log_odds_margin": float(sum_margin[0]),
"mean_preference_loss": float(mean[0]),
"sum_preference_loss": float(summed[0]),
"scope": "Same already scored main ORPO policy, formula ablation only",
}
)
branches = {
b: read_json(f"outputs/learning/train-{b}-continuous.json")
for b in ["sft", "orpo", "sum-ablation"]
}
initial = {r["initial_model_digest"] for r in branches.values()}
schedules = [
[
(r["document_ids"], r["chosen_token_denominator"], r["learning_rate"])
for r in run["history"]
]
for run in branches.values()
]
if (
len(initial) != 1
or not all(schedule == schedules[0] for schedule in schedules)
or any(run["updates"] != 210 for run in branches.values())
):
raise ValueError("Ablation initial model/example budgets differ")
result = read_json("outputs/evaluation/exercise-comparison.json")
if not result["exercise_sum_is_not_reference_ORPO"]:
raise ValueError("Sum variant must not be labelled reference ORPO")
save_json(
"outputs/exercise/odds-length.json",
{
"analytic_rows": compare_sum_mean_odds(),
"analytic_source": "Constructed constant token log probabilities, not generated incident notes",
"actual_main_policy_rescoring": rescored,
"actual_fresh_sum_training": True,
"same_base_and_prompt_order": True,
"updates_per_branch": 210,
"main_normalization_restored": "mean",
"main_artifacts_unchanged": protected
== {name: sha(name) for name in main_artifacts()},
"protected_sha256": protected,
"sum_variant_is_not_reference_ORPO": True,
"quality_gain_required": False,
},
)
Fonti
ORPO, articolo originale, repository degli autori congelato, funzione di controllo TRL congelata, Qwen2.5-0.5B-Instruct. Dati, maschere e budget locali non riproducono i risultati pubblicati da questi autori.
Riferimento misurato
Consultate i risultati di sviluppo r2 (RESULTATS_DEVELOPPEMENT.md, incluso nello ZIP) e la loro provenienza esplicita (OBSERVATIONS.json, incluso nello ZIP). Il codice scientifico resta identico in r3; viene corretta solo una legenda.
