El soporte ficticio AtelierDesk debe responder con los hechos del expediente y pedir información faltante en vez de prometer un plazo inventado. Para un mismo expediente, dos respuestas se anotan preferida/rechazada mediante reglas de simulación públicas.
Compare la política tras SFT local, tras un suplemento SFT sobre las respuestas elegidas y tras DPO. Muestre para cada par las log-probabilidades de la política y su referencia, y después las respuestas libres sobre expedientes nuevos.
Descargar el proyecto completo, revisión r001
Del expediente a una preferencia
Una preferencia compara dos respuestas al mismo expediente. No es ni una nota absoluta ni una prueba de satisfacción humana. Las reglas de AtelierDeskPrefs-v1 son públicas y simuladas: respetar el estado del reemplazo y la fecha de envío registrados; pedir el número de pedido solo si falta. Una fecha desconocida sigue siendo desconocida. Una fórmula cortés no convierte en verdadera una promesa inventada.
300 familias dan cuatro pares cada una: 840 train, 180 validation y 180 test, sobre familias disjuntas 0–209, 210–254 y 255–299. Una demostración adicional por familia de train proporciona 210 ejemplos para el arranque SFT. Expedientes e identificadores son distintos, pero hechos elementales y patrones de lenguaje siguen compartidos. Esto no mide la generalización a servicios de atención reales ni a intenciones desconocidas.
Dos variantes cambian el estado; una cambia la fecha; una invierte la petición del número de pedido. Las fórmulas de cortesía alternan: elegida más corta, de igual longitud en palabras o más larga. Una auditoría publica presencias de cada palabra en ambos labels, un clasificador binario por palabra y una elección por longitud con medio crédito en empates. Esta auditoría no prueba la ausencia de todo atajo aprendido. Los estratos de evaluación usan longitudes reales en tokens, que pueden diferir de los recuentos de palabras. Los metadatos chosen/rejected y el motivo de preferencia nunca se incluyen en los expedientes de generación.
La política produce probabilidades condicionales
Qwen2.5-0.5B-Instruct, revisión 7ae557604adf67be50417f59c2c2f167def9a775, se carga estrictamente: 24 bloques, anchura 896, 14 cabezas Q, dos K/V, MLP 4864, vocabulario 151936 y embeddings/salida vinculados. Sus 494 032 768 parámetros únicos son entrenables en FP32. El perfil usa atención eager, TF32 desactivado, microbatch de una respuesta; no hay cuantificación, LoRA ni distribución automática en varias GPU.
El tokenizer aplica su plantilla al mensaje de sistema y al expediente, y añade la frontera de asistente. La respuesta se codifica por separado y recibe exactamente un EOS=151645 final. El contrato es explícito: no pretende reproducir una retokenización de la cadena concatenada. Máximo 256 tokens de prompt, 128 de respuesta con EOS y total inferior a 512; todo exceso falla sin truncar.
Para predecir el token en posición t se toma el logit en t−1. El código alinea logits[:, :-1] con labels[:, 1:]. Los labels del prompt y padding valen −100; los objetivos de respuesta, incluido EOS, permanecen activos. PAD=151643 se enmascara por posición. El prompt no tiene CE propia, pero sus tokens condicionan las probabilidades de respuesta: ignorar sus labels no elimina su influencia.
Para una respuesta y, s(x,y) es la SUMA de las log-probabilidades de sus tokens activos. Una respuesta larga acumula más términos generalmente negativos. El informe también publica medias por token para diagnosticar un sesgo de longitud; estas medias nunca sustituyen las sumas en DPO.
SFT local y referencia congelada
El arranque aprende las 210 demostraciones train durante dos épocas, acumulando ocho ejemplos: 27 actualizaciones por época, de las que la última trata dos ejemplos, 54 en total. La CE se normaliza por el número total de tokens de respuesta de la actualización; una media de medias daría otro peso a respuestas cortas. AdamW usa lr=0,00002 y weight_decay=0,01, clipping a 1, warm-up de cinco pasos y descenso lineal hasta el 10 % de la tasa máxima.
El checkpoint SFT final se guarda en Safetensors y se recarga estrictamente. La referencia es una instancia congelada, eval y no_grad. Nunca entra en el optimizador. Su hash y el digest de sus parámetros siguen sin cambios. Cada política de preferencia parte de una recarga independiente del mismo SFT; no comparte un objeto mutable con la referencia.
Para cada par, la caché conserva las dos respuestas realmente tokenizadas, sus máscaras, sus puntuaciones por token y sus sumas. Su clave vincula checkpoint, tokenizer y configuración, plantilla y texto de sistema, código de puntuación, lock del modelo, ID y labels completos. Una caché existente con una identidad diferente se rechaza antes de usarse. Solo se montan train y validation en learning; test y salidas de generación no son accesibles.
El coste del SFT, de la caché de referencia y de cada optimización se mide por separado. La caché evita cargar una segunda referencia en cada backward DPO; su coste inicial no es gratuito.
La pérdida codificada y su gradiente
Para un par intervienen cuatro números: sθ+ y sθ− para la política, sref+ y sref− para la referencia. El margen es m = β[(sθ+ − sθ−) − (sref+ − sref−)]. La pérdida es −logsigmoid(m), promediada sobre los pares de la actualización. El código usa logsigmoid para la estabilidad numérica. La referencia proporciona constantes sin gradiente.
Si política y referencia son iguales, el margen vale cero y la pérdida log(2). Sin embargo, para dos respuestas diferentes las derivadas de puntuaciones de política son opuestas y no nulas: −β/2 para sθ+ y +β/2 para sθ− en un par. El modelo puede aprender. Si ambas respuestas tienen exactamente los mismos tokens y máscaras, los términos se compensan y no dan esa señal; estos pares se rechazan al construirlos.
El valor y gradientes FP64 se comparan con la función preference_loss de los autores de DPO, extraída del commit f8b8c0f49dc92a430bae41585f9d467d3618fe2f, con label_smoothing=0, ipo=False y reference_free=False. Se proporcionan licencia Apache-2.0 y procedencia. Solo se usan esta función y los imports necesarios; no se ejecutan el trainer original ni sus ajustes globales.
Una comparación definida antes de los resultados
Tras SFT, cuatro políticas reciben cada una 840 prompts durante dos épocas, en el mismo orden determinista: 210 actualizaciones y 1680 presentaciones. Chosen-only aprende solo la respuesta elegida por CE normalizada en tokens. DPO recibe elegida y rechazada, con β=0,05, 0,1 o 0,2. Las cuatro ramas usan AdamW lr=0,000001, weight_decay=0,01, clipping a 1, warm-up de 21 pasos y descenso hasta el 10 %.
El número de presentaciones de prompts es igual; los tokens puntuados, forwards y cálculo no lo son. DPO promedia pérdidas por par, chosen-only por token de respuesta. Se informan todos los beta y el checkpoint final: ninguna selección del mejor beta o checkpoint usa test. La validación tras cada época ofrece NLL elegida y, para preferencias, márgenes y accuracy. No detiene el aprendizaje.
Cada fase, incluido SFT, se ejecuta con interrupción después de una época, recorrido continuo de dos épocas y reanudación de la segunda. Modelo completo, AdamW, sampler, scheduler, RNG Python/NumPy/PyTorch CPU/CUDA e historiales finales deben ser exactamente iguales. Este contrato se refiere al límite de época, no a una parada en medio de un backward.
Los quince estados locales se escriben en gzip nivel 1 sin pérdida, en archivo temporal y después se renombran. Se miden los bytes serializados sin comprimir y el tamaño realmente escrito. Esta elección no promete una tasa de compresión ni reducción de VRAM. Los estados Torch son objetos de confianza producidos localmente; no deben sustituirse por una descarga no fiable. Las cinco políticas finales también se conservan en Safetensors. Los artefactos existentes no se sobrescriben.
Preferir sin conexión y responder libremente
Las cinco políticas SFT/chosen-only/DPO×3 puntúan los mismos 180 pares test. La accuracy cuenta s+ > s−, cero en orden inverso y medio punto en igualdad. Márgenes corregidos de referencia y medias por token se separan. Informes estratificados por longitud evitan ocultar un fallo sobre respuestas elegidas más largas tras una media general.
Cada política también genera 180 respuestas sin chosen/rejected en contexto: solo expediente con la misma instrucción, decodificación greedy, un beam, máximo 128 tokens nuevos y parada EOS. ID, textos brutos, razones de parada y duraciones permanecen disponibles. Una salida detenida en el límite falla aunque las primeras frases parezcan correctas. No se usa modelo juez, envío al cliente ni acción de negocio.
Un analizador independiente de la generación reconoce las cláusulas controladas: estado, fecha conocida/desconocida, petición o acuse de presencia del número. Exige cobertura de TODO el texto, tres campos distintos y EOS. Una frase desconocida, un duplicado, una etiqueta Markdown o un rechazo vuelve incompleta la respuesta. Una frase natural correcta fuera de gramática puede fallar: esta puntuación estrecha no es una medida universal de fidelidad semántica.
La exactitud exige los tres hechos/acciones correctos en los 180 casos sin excluir salidas inválidas. Las peticiones útiles se informan en casos donde falta el número. Las fechas inventadas contadas son fechas reconocidas cuando la fecha es desconocida; frases no reconocidas se informan separadamente y nunca se certifican sin invención. Los rechazos contados siguen expresiones explícitas; este detector no identifica todos los rechazos posibles. Una subida del margen DPO no prueba una subida de calidad libre ni de satisfacción cliente.
Ejercicio: un archivo antiguo no garantiza nada
En los doce primeros expedientes test, la plantilla support-v2 añade una instrucción de concisión. Los ID condicionantes deben cambiar realmente. La caché antigua se rechaza antes de leer sus puntuaciones. Una referencia congelada recalcula una nueva caché y, por separado, las mismas puntuaciones sin caché; deben ser exactamente iguales. La política dpo-010 puntúa los nuevos pares y las dos vías de referencia dan la misma pérdida DPO. No se realiza entrenamiento adicional. Los hashes de pesos, cachés y resultados principales no cambian.
Una sola semilla, preferencias simuladas y lenguaje controlado limitan el alcance del TP. Beta no corrige anotaciones contradictorias ni un corpus mal diseñado. No existe exigencia de ganancia: una regresión o una generación inválida debe conservarse y explicarse.
Construir pares informativos sin atajo
Preferencias y sesgo de longitud
Error frecuente: Un marcador de generación revela chosen: retirar metadatos de textos y probar el atajo.
Invariante: 840/180/180 pares y 210 demostraciones train, motivos y longitudes auditados.
Archivo 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
Puntuar exactamente las respuestas
Suma de log-probabilidades y máscaras
Error frecuente: Un EOS duplicado favorece artificialmente una respuesta: inspeccionar la secuencia decodificada y sus ID.
Invariante: Puntuaciones idénticas al cálculo token por token y PAD invariantes.
Archivo 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,
},
)
Arrancar la política y congelar la referencia
SFT local e inicialización común
Error frecuente: Reference apunta al mismo objeto mutable que policy: copiar realmente los pesos y probar la independencia.
Invariante: Los hashes policy/ref son iguales al inicio y ref no tiene ningún gradiente.
Archivo 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,
},
)
Codificar DPO y la baseline chosen-only
Margen relativo, sigmoide y optimización
Error frecuente: Sustituir las sumas por medias cambia el objetivo: prueba numérica sobre longitudes desiguales.
Invariante: Los gradientes son finitos, ref inmutable y presupuestos de exposición capturados.
Archivo 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",
},
)
Evaluar pares y respuestas nuevas
Preferencia sin conexión frente a generación
Error frecuente: Un label chosen termina en el prompt de generación: reconstruir el prompt solo desde el expediente.
Invariante: 180 puntuaciones de pares y 180 generaciones por rama, con los mismos ID.
Archivo 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,
},
)
Provocar y bloquear una caché caducada
Integridad de la señal de referencia
Error frecuente: Desactivar la verificación para continuar oculta el error: conservar el rechazo como prueba obligatoria.
Invariante: La caché caducada se rechaza; la caché recalculada coincide con el cálculo directo.
Archivo 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"],
},
)
Fuentes
DPO, artículo de los autores, función original fijada, Qwen2.5-0.5B-Instruct. Las decisiones locales de datos, presupuesto, caché y gramática no constituyen una reproducción de los resultados del artículo.
Observaciones de desarrollo
Valores recalculados a partir de las salidas brutas de este recorrido. No son umbrales de ganancia; el recorrido desde el ZIP los recalcula sobre las mismas poblaciones.
{
"status": "PASS",
"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"
}
