Un ejercicio ficticio de inventario describe entradas y salidas de lotes. El modelo debe devolver un pequeño JSON con el stock final como entero. Un verificador independiente calcula la respuesta a partir de los datos estructurados, sin ejecutar el texto del modelo.
Visualice cuatro respuestas al mismo problema, sus recompensas, las ventajas relativas y una actualización. Compare el SFT de arranque, el SFT adicional y GRPO en problemas nuevos, incluido el coste de las generaciones.
Descargar el proyecto completo, revisión r001
La respuesta debe coincidir con el cálculo
StockArithmetic-v1 contiene 1400 problemas: 1000 de entrenamiento, 200 de validación y 200 de prueba. Cada problema proporciona un stock inicial, dos entregas y dos retiradas; todos los stocks intermedios siguen siendo no negativos. Las tuplas se deduplican tras ordenar por separado las entregas y las retiradas, para que una permutación de los mismos movimientos no atraviese las particiones. Los patrones de lenguaje y las operaciones aritméticas se comparten: este pequeño corpus no representa inventarios reales.
Los operandos de entrenamiento y validación están entre 0 y 50. La prueba contiene 100 problemas de ese intervalo y 100 con al menos un operando entre 51 y 100, marcados como OOD100. Los resultados de ambas poblaciones se separan. Los primeros 128 objetivos de entrenamiento sirven para el arranque; los otros 872 para el SFT adicional o para las extracciones de prompts de RL. Los objetivos se calculan, no los anota una persona ni los produce un LLM.
El modelo solo ve el enunciado y una instrucción de formato. El verificador recibe por separado la estructura del problema. Exige un único objeto JSON completo, limitado a 256 caracteres, que contenga exactamente final_stock con un entero entre 0 y 300. Se rechazan booleanos, flotantes, campos adicionales, duplicados, Markdown, varios objetos y números no finitos. Un ID generado que no exista en el tokenizer también invalida la respuesta; no puede desaparecer silenciosamente durante la decodificación. No se ejecuta ningún texto generado. Una respuesta detenida por el límite de tokens recibe cero, aunque su inicio contenga un JSON correcto.
La recompensa vale uno para una respuesta completa y correcta, y cero en todos los demás casos. No hay prima de estilo, recompensa parcial ni decodificación JSON restringida. El solucionador determinista ya sabe calcular todos los objetivos: un aprendizaje costoso puede ser inútil para esta tarea.
Tres políticas diferentes
Qwen2.5-0.5B-Instruct se adquiere en la revisión 7ae557604adf67be50417f59c2c2f167def9a775. Sus 494032768 parámetros únicos se optimizan todos en FP32, con atención eager y TF32 desactivado. Se controlan la arquitectura y los pesos compartidos entre embeddings/salida. Los logits predicen el siguiente token; solo se supervisan los tokens de respuesta, incluido EOS 151645. El prompt y el padding se enmascaran por posición. El prompt tiene como máximo 256 tokens, la respuesta 64 y el total 320; ningún exceso de los ejemplos supervisados se trunca silenciosamente.
El arranque SFT aprende 128 ejemplos durante tres épocas: 48 actualizaciones de ocho ejemplos. El SFT adicional parte de ese estado y aprende 872 ejemplos durante dos épocas: 218 actualizaciones. Ambos usan AdamW a 2e−5. La CE se normaliza sobre todos los tokens activos de la acumulación, no como una media de medias de respuestas.
GRPO parte del mismo arranque. La referencia fija es una copia de ese estado recargada de forma estricta, congelada y sin gradientes. La política antigua designa el estado que produjo las respuestas de una iteración; sus log-probabilidades se capturan durante la generación y permanecen separadas. La política actual recibe los gradientes. No hay un crítico de valor aprendido. Estos tres papeles no son intercambiables.
Del grupo a la ventaja
Una iteración elige cuatro prompts de entrenamiento. Para cada uno, muestrea G=4 respuestas secuenciales a temperatura 1, sin top-k, sin truncamiento top-p y sin retirar tokens del vocabulario. Las probabilidades de comportamiento proceden de los logits reales en el momento de la extracción. Se conservan tokens, puntuaciones, estado de la política, motivos de parada y tiempos. El modo eval desactiva el dropout durante la generación Y durante las actualizaciones; eval no prohíbe autograd.
En cada grupo se calcula A=(r−media)/(desviación estándar+1e−8), con desviación estándar poblacional, correction=0. Las recompensas y ventajas se separan. Para [1,0,1,0], las ventajas son aproximadamente [1,−1,1,−1]; epsilon explica que no sean exactamente esos enteros. Si las cuatro recompensas son idénticas, todas las ventajas son exactamente cero. Se mide la frecuencia de estos grupos sin inventarles un ganador.
En cada token de respuesta, rho=exp(logp_actual−logp_antigua). Antes de la primera actualización, la puntuación de la secuencia completa debe recuperar las puntuaciones capturadas en los prefijos con una diferencia de hasta 2e−4; las formas de los kernels pueden producir una pequeña diferencia numérica. La puntuación antigua nunca se sustituye por una separación de los nuevos logits tras el aprendizaje.
El término que se maximiza es min(rho A, clip(rho,0,8,1,2) A). El signo de A es esencial: las pruebas controlan ambos lados y los gradientes de las ramas recortadas. La pérdida es el opuesto de ese término, promediado primero sobre los tokens de cada respuesta, después sobre las respuestas y por último sobre los prompts. Una respuesta larga no obtiene automáticamente más peso total; esta normalización conserva, sin embargo, sesgos de longitud y dificultad que un experimento pequeño no elimina.
La KL exacta en los prefijos visitados
La variante se llama «GRPO pedagógico, KL exacta en prefijos». Para cada prefijo realmente visitado, se calcula suma_v p_actual(v) × [log p_actual(v)−log p_referencia(v)] sobre todo el vocabulario. Se recorre en bloques de 4096, con los normalizadores logsumexp completos. Normalizar cada bloque por separado cambiaría la divergencia.
Esta suma categórica es exacta salvo redondeo para cada prefijo. Los prefijos siguen siendo los de los rollouts: no enumera todas las secuencias posibles. El coeficiente es 0,02. Su reducción sigue la misma media tokens/respuesta/prompt que la surrogate. Valores y gradientes en tensores pequeños se comparan con kl_div de PyTorch en FP64 y FP32; algunas distribuciones completas reales sirven como sondas independientes.
El artículo utiliza una estimación distinta de la penalización KL. TRL propone varias reducciones, entre ellas DAPO; no afirmamos usar ni sus valores predeterminados ni la identidad de su trainer. El TP implementa explícitamente su variante y no instala TRL. Los grupos con ventaja nula no tienen gradiente de recompensa, pero la regularización KL y las reglas del optimizador aún pueden actuar.
Presupuesto, costes y reanudación
El mínimo de GRPO realiza 100 iteraciones, cuatro prompts por iteración, cuatro respuestas por prompt y dos actualizaciones sobre cada conjunto de rollouts: 1600 respuestas reales y 200 actualizaciones. AdamW usa lr=1e−6, weight_decay=0,01, recorte de gradiente a 1, calentamiento del 10 % y después una disminución lineal hasta el 10 % de la tasa máxima. El arranque, SFT y RL no consumen los mismos ejemplos, tokens, forwards ni segundos de GPU; no se presupone igualdad de coste.
La generación sin caché KV favorece la legibilidad de la puntuación. Se distinguen los tiempos de generación, la comprobación de puntuaciones antiguas, la optimización con referencia, el bucle completo y los comandos. Las huellas de estado también tienen un coste. Los costes de la comparación entre ejecución continua y reanudación deben permanecer visibles además de los del entrenamiento continuo de referencia.
La reanudación de GRPO interviene tras una iteración completamente optimizada, es decir, después de las dos actualizaciones. Se guardan el modelo completo, AdamW, scheduler, permutación/cursor de los prompts, RNG de generación, RNG del sistema y la lista de rollouts consumidos. Interrumpir en 40, continuar hasta 100 y reanudar las 60 restantes deben terminar de forma exactamente idéntica. Un rollout no puede consumirse dos veces. Las duraciones reales quedan fuera del estado numérico comparable, pero se conservan sus pruebas. SFT también compara una interrupción después de una época, la ejecución continua y la reanudación.
Los quince estados completos de las cinco ramas permanecen locales en gzip nivel 1 sin pérdida; también se recargan cinco modelos finales en Safetensors. Se miden los tamaños, sin prometer una tasa de compresión. Cargar objetos Torch solo se permite aquí porque se produjeron localmente; no los sustituya por un archivo de terceros.
Evaluar una política, no seleccionar su mejor respuesta
Las políticas de arranque, SFT y GRPO generan cada una una respuesta codiciosa en los mismos 200 problemas de prueba: es el pass@1 codicioso. También producen cuatro muestras estocásticas independientes por problema con semillas fijadas. La tasa de éxito por muestra usa las 800 respuestas; no es el éxito de la mejor de las cuatro. Las poblaciones ID50, OOD100 y su total se mantienen separadas.
La validez JSON, la exactitud, la ausencia de EOS, las longitudes y los tiempos se calculan sobre todas las salidas. La validación SFT mide la NLL sin elegir un checkpoint; los estados finales se fijan antes de los ensayos. Las respuestas de prueba no entran en las actualizaciones. Una reducción de loss no demuestra una ganancia de exactitud, y un resultado desfavorable para GRPO es publicable.
Ejercicio: dos tamaños de grupo
Los grupos constantes se estudian primero con recompensas construidas, claramente separadas de las respuestas del modelo. Después se entrenan realmente G=2 y G=8 durante 20 iteraciones desde el mismo arranque, con las mismas 80 presentaciones de prompts y 40 actualizaciones. Producen respectivamente 160 y 640 rollouts: el mismo número de actualizaciones no significa el mismo coste. Las dos políticas se evalúan en los mismos 200 problemas y con cuatro muestras.
El G=4 principal tiene 100 iteraciones. Compararlo directamente con las ramas de 20 iteraciones no sería una ablación que aislara solo el tamaño del grupo; esta limitación se indica. El ejercicio protege los SHA-256 de los pesos, resultados y configuración principales. El verificador no cambia, aunque todos los grupos sean incorrectos.
Escribir el problema y su verificador independiente
Recompensa que no depende del modelo
Error que diagnosticar: una salida contiene dos JSON, uno de ellos correcto; exigir una sola respuesta completa.
Invariante: 1 000/200/200 problemas distintos; solucionador exacto en fixtures y casos límite.
Archivo src/grpo_recompense_verifiable/verifier.py
import json, random
def compute_final_stock(problem):
if not isinstance(problem, dict) or set(problem) != {
"initial",
"deliveries",
"withdrawals",
}:
raise ValueError("Expected initial, deliveries and withdrawals only")
if type(problem["initial"]) is not int or not 0 <= problem["initial"] <= 100:
raise ValueError("Initial stock must be an integer in 0..100")
stock = problem["initial"]
for key, sign in [("deliveries", 1), ("withdrawals", -1)]:
values = problem[key]
if (
not isinstance(values, list)
or len(values) != 2
or any(type(v) is not int or not 0 <= v <= 100 for v in values)
):
raise ValueError("Exactly two bounded integer movements required")
for value in values:
stock += sign * value
if stock < 0:
raise ValueError("An intermediate stock cannot be negative")
return stock
def normalized_problem(problem):
compute_final_stock(problem)
return (
problem["initial"],
tuple(sorted(problem["deliveries"])),
tuple(sorted(problem["withdrawals"])),
)
def generate_stock_problem(rng, maximum=50):
if maximum not in [50, 100]:
raise ValueError("Named ID50 or OOD100 range required")
initial = rng.randint(0, maximum)
deliveries = [rng.randint(0, maximum) for _ in range(2)]
available = initial + sum(deliveries)
first = rng.randint(0, min(maximum, available))
second = rng.randint(0, min(maximum, available - first))
return {
"initial": initial,
"deliveries": deliveries,
"withdrawals": [first, second],
}
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError("Duplicate JSON field")
result[key] = value
return result
def reject_constant(value):
raise ValueError("Non-finite JSON number forbidden")
def parse_generated_json(text, finish_reason):
if finish_reason != "eos":
return {"valid": False, "value": None, "reason": "completion_without_eos"}
if not isinstance(text, str) or len(text) > 256:
return {"valid": False, "value": None, "reason": "oversized_or_nontext"}
try:
obj = json.loads(
text, object_pairs_hook=unique_object, parse_constant=reject_constant
)
if type(obj) is not dict or set(obj) != {"final_stock"}:
raise ValueError("One exact field required")
value = obj["final_stock"]
if type(value) is not int or not 0 <= value <= 300:
raise ValueError("A bounded integer excluding bool/float is required")
return {"valid": True, "value": value, "reason": None}
except (ValueError, TypeError, RecursionError):
return {"valid": False, "value": None, "reason": "invalid_json_schema"}
def score_generated_json(text, finish_reason, problem, tokenization_valid=True):
expected = compute_final_stock(problem)
parsed = parse_generated_json(text, finish_reason)
if type(tokenization_valid) is not bool:
raise ValueError("Explicit tokenization validity required")
if not tokenization_valid:
parsed = {"valid": False, "value": None, "reason": "undecodable_token"}
return {
**parsed,
"reward": int(parsed["valid"] and parsed["value"] == expected),
"expected": expected,
}
Formar un grupo e inspeccionar sus puntuaciones
Rollouts y probabilidades de comportamiento
Error que diagnosticar: un dropout activo vuelve incoherentes las log-probs; poner la política en modo de evaluación y permitir a la vez autograd en las actualizaciones.
Invariante: cada grupo tiene un prompt_id único y cuatro respuestas con puntuación compatible.
Archivo src/grpo_recompense_verifiable/rollout.py
import time
import torch
from .scores import format_prompt, rollout_completion_row, model_scores
from .loss import group_advantages
from .verifier import score_generated_json
def capture_behavior_logprobs(logits, token):
# Probability captured at the actual sampling prefix, not reconstructed from
# the current policy after an optimizer step. Temperature is fixed to one.
return float(torch.log_softmax(logits.float(), dim=-1)[token].detach())
@torch.no_grad()
def sample_response(
model,
tokenizer,
page,
config,
generator,
mode="sample",
policy_id=None,
draw=0,
capture_probe=False,
):
if model.training:
raise ValueError("Sampling requires eval mode; dropout must be off")
if config["temperature"] != 1.0 or config["top_p"] != 1.0 or config["top_k"] != 0:
raise ValueError(
"The public protocol requires full support and temperature one"
)
if mode not in ["sample", "greedy"]:
raise ValueError("Named generation mode required")
messages, prompt = format_prompt(
tokenizer, page["statement"], config["template_version"]
)
if not prompt or len(prompt) > config["max_prompt_tokens"]:
raise ValueError("Prompt exceeds the declared budget; no truncation")
ids = torch.tensor([prompt], device="cuda")
response = []
behavior = []
first_distribution = None
torch.cuda.synchronize()
start = time.perf_counter()
for position in range(config["max_new_tokens"]):
# No KV cache: sampling and teacher-forced scoring both use eager FP32
# causal attention. Prefix/full-sequence kernels may differ numerically.
logits = (
model(
input_ids=ids,
attention_mask=torch.ones_like(ids),
use_cache=False,
return_dict=True,
)
.logits[0, -1]
.float()
)
probabilities = torch.softmax(logits, dim=-1)
if not torch.isfinite(probabilities).all():
raise ValueError("Non-finite sampling distribution")
token = (
int(torch.argmax(logits))
if mode == "greedy"
else int(torch.multinomial(probabilities, 1, generator=generator))
)
behavior.append(capture_behavior_logprobs(logits, token))
response.append(token)
if position == 0 and capture_probe:
first_distribution = {
"prefix_ids": prompt,
"logits": logits.cpu().tolist(),
"target": token,
"selected_logprob": behavior[-1],
}
ids = torch.cat([ids, torch.tensor([[token]], device="cuda")], dim=1)
if token == tokenizer.eos_token_id:
break
torch.cuda.synchronize()
seconds = time.perf_counter() - start
finish = "eos" if response[-1] == tokenizer.eos_token_id else "length"
# Keep non-EOS special tokens in decoded text: silently dropping generated
# controls could turn an invalid response into apparently valid JSON.
decoded_ids = response[:-1] if finish == "eos" else response
text = tokenizer.decode(
decoded_ids, skip_special_tokens=False, clean_up_tokenization_spaces=False
)
tokenization_valid = all(
tokenizer.convert_ids_to_tokens(token) is not None for token in response
)
return {
"document_id": page["document_id"],
"messages": messages,
"prompt_ids": prompt,
"response_ids": response,
"old_token_logprobs": behavior,
"response_tokens": len(response),
"finish_reason": finish,
"text": text,
"tokenization_valid": tokenization_valid,
"generation_seconds": seconds,
"mode": mode,
"draw": draw,
"policy_id": policy_id,
"temperature": 1.0,
"top_p": 1.0,
"top_k": 0,
"dropout_active": False,
"first_prefix_probe": first_distribution,
}
@torch.no_grad()
def sample_group(
model,
tokenizer,
page,
problem,
config,
generator,
policy_id,
group_id,
group_size=None,
capture_probe=False,
):
size = config["group_size"] if group_size is None else group_size
if type(size) is not int or size < 2:
raise ValueError("At least two actual stochastic responses required")
responses = []
for member in range(size):
row = sample_response(
model,
tokenizer,
page,
config,
generator,
policy_id=policy_id,
draw=member,
capture_probe=capture_probe and member < 2,
)
row["rollout_id"] = f"{group_id}/response-{member:02d}"
row["verifier"] = score_generated_json(
row["text"], row["finish_reason"], problem, row["tokenization_valid"]
)
responses.append(row)
rewards = torch.tensor(
[[r["verifier"]["reward"] for r in responses]], dtype=torch.float32
)
advantages, mean, std = group_advantages(rewards, config["advantage_epsilon"])
return {
"group_id": group_id,
"document_id": page["document_id"],
"behavior_policy_id": policy_id,
"responses": responses,
"rewards": rewards[0].tolist(),
"advantages": advantages[0].tolist(),
"reward_mean": float(mean[0, 0]),
"population_std": float(std[0, 0]),
"zero_variance": bool(std[0, 0] == 0),
"std_correction": 0,
"epsilon": config["advantage_epsilon"],
}
@torch.no_grad()
def verify_behavior_before_update(model, group, tolerance=2e-4):
if model.training:
raise ValueError("Ratio verification requires eval mode")
maximum = 0.0
for response in group["responses"]:
_, _, actual, _, _ = model_scores(model, [rollout_completion_row(response)])
recorded = torch.tensor(
response["old_token_logprobs"], device=actual.device, dtype=actual.dtype
)
error = float((actual - recorded).abs().max())
maximum = max(maximum, error)
if error > tolerance:
raise ValueError(
"Behavior/full-prefix scoring disagreement before update: " + str(error)
)
return {
"maximum_absolute_logprob_difference": maximum,
"tolerance": tolerance,
"old_logprobs_source": "captured at actual generation, unchanged",
"before_any_update": True,
}
Arrancar y comparar un SFT directo
Referencia de aprendizaje sin RL
Error que diagnosticar: descargar un checkpoint de un TP anterior sustituye el mínimo; rehacer el arranque en este proyecto.
Invariante: ningún ejemplo de prueba en el arranque, con costes y hash inicial capturados.
Archivo src/grpo_recompense_verifiable/warmstart.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,
save_learned,
parameter_manifest,
manifest_digest,
)
from .scores import completion_row, model_scores
from .state import save_resume_state, load_resume_state, equal
def learning_rate(config, step, total, peak):
warm = max(1, int(total * config["warmup_ratio"]))
if step < warm:
return peak * (step + 1) / warm
return peak * (
1
- min(1.0, (step - warm) / max(1, total - warm - 1))
* (1 - config["minimum_lr_ratio"])
)
def assert_learning_scope():
for name in [
"data/generated/input",
"data/generated/labels/gold.json",
"outputs/inference",
]:
if Path(name).exists():
raise ValueError(
"Learning must not mount reserved inputs, gold or inference outputs: "
+ name
)
def encoded_split(config, tokenizer, split):
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")
}
return [
completion_row(
tokenizer,
p["statement"],
labels[p["document_id"]]["response"],
config,
p["document_id"],
)
for p in pages
]
@torch.no_grad()
def validation_nll(model, rows):
model.eval()
result = []
for row in rows:
sums, counts, values, _, targets = model_scores(model, [row])
result.append(
{
"document_id": row["document_id"],
"sum_logprob": float(sums[0]),
"tokens": int(counts[0]),
"token_logprobs": values.cpu().tolist(),
"targets": targets.cpu().tolist(),
}
)
return {
"examples": len(rows),
"tokens": sum(r["tokens"] for r in result),
"nll": -sum(r["sum_logprob"] for r in result)
/ sum(r["tokens"] for r in result),
"rows": result,
"used_for_selection": False,
}
def freeze_initial_reference():
model, metadata = reload_learned("assets/checkpoints/warmstart-resume.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 be frozen without gradients")
return model, metadata
def train_warmstart(config, branch="warmstart", mode="continuous"):
if branch not in ["warmstart", "sft"] or mode not in [
"interrupt",
"continuous",
"resume",
]:
raise ValueError("Unknown SFT phase")
assert_learning_scope()
seed_all(config["seed"])
tokenizer = load_tokenizer()
split = "warmstart" if branch == "warmstart" else "train"
rows = encoded_split(config, tokenizer, split)
validation = encoded_split(config, tokenizer, "validation")
if len(rows) != (128 if branch == "warmstart" else 872) or len(validation) != 200:
raise ValueError("Frozen SFT population changed")
model = (
load_base()
if branch == "warmstart"
else reload_learned("assets/checkpoints/warmstart-resume.safetensors")[0]
)
model = model.cuda().eval()
parameters = list(model.parameters())
initial = parameter_manifest(model)
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
):
raise ValueError("Expected full FP32 policy")
epochs = (
config["warmstart_epochs"] if branch == "warmstart" else config["sft_epochs"]
)
peak = (
config["warmstart_learning_rate"]
if branch == "warmstart"
else config["sft_learning_rate"]
)
total = math.ceil(len(rows) / config["accumulation_examples"]) * epochs
identity = {
"branch": branch,
"config": config,
"initial_model_digest": digest,
"model_sha256": sha("assets/model/model.safetensors"),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"data_sha256": {
name: sha("data/generated/" + name)
for name in [
f"{split}/pages.json",
f"labels/{split}.json",
"validation/pages.json",
"labels/validation.json",
]
},
"code_sha256": {
name: sha("src/grpo_recompense_verifiable/" + name)
for name in ["model.py", "scores.py", "warmstart.py", "state.py"]
},
}
optimizer = torch.optim.AdamW(
parameters, lr=peak, weight_decay=config["weight_decay"]
)
sampler = torch.Generator().manual_seed(config["seed"] + 100)
root = Path("assets/checkpoints")
destination = root / f"{branch}-{mode}.pt.gz"
if destination.exists():
raise ValueError("Preserve existing full checkpoint")
updates = start = 0
history = []
validation_history = []
if mode == "resume":
state = load_resume_state(root / f"{branch}-interrupt.pt.gz")
if state["identity"] != identity:
raise ValueError("SFT resume 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"])
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, peak),
}:
raise ValueError("SFT scheduler changed")
del state
stop = 1 if mode == "interrupt" else epochs
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()
start_epoch = time.perf_counter()
for offset in range(0, len(rows), config["accumulation_examples"]):
selected = order[offset : offset + config["accumulation_examples"]]
denominator = sum(rows[i]["response_tokens"] for i in selected)
lr = learning_rate(config, updates, total, peak)
for group in optimizer.param_groups:
group["lr"] = lr
optimizer.zero_grad(set_to_none=True)
details = []
for index in selected:
sums, counts, *_ = model_scores(model, [rows[index]])
loss = -sums.sum() / denominator
if not torch.isfinite(loss):
raise ValueError("Nonfinite SFT objective")
loss.backward()
details.append(
{
"document_id": rows[index]["document_id"],
"sum_logprob": float(sums.detach()[0]),
"response_tokens": int(counts[0]),
}
)
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 full-policy gradients required")
optimizer.step()
updates += 1
history.append(
{
"update": updates,
"epoch": epoch + 1,
"document_ids": [rows[i]["document_id"] for i in selected],
"response_token_denominator": denominator,
"nll": -sum(r["sum_logprob"] for r in details) / denominator,
"learning_rate": lr,
"gradient_norm_before_clip": float(norm),
"rows": details,
}
)
torch.cuda.synchronize()
optimization_seconds += time.perf_counter() - start_epoch
observed = validation_nll(model, validation)
validation_history.append({"epoch": epoch + 1, **observed})
print(
{
"branch": branch,
"epoch": epoch + 1,
"updates": updates,
"validation_nll": observed["nll"],
},
flush=True,
)
torch.cuda.synchronize()
elapsed = time.perf_counter() - begun
peak_memory = torch.cuda.max_memory_allocated()
final = parameter_manifest(model)
final_digest = manifest_digest(final)
if final_digest == digest:
raise ValueError("No actual SFT weight update")
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, peak),
},
"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": 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_memory,
"test_visible": False,
"dropout_active": 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("Full resume differs: " + str(checks))
save_json(
f"outputs/learning/resume-{branch}.json",
{
"status": "PASS",
"all_state_exact": True,
"checks": checks,
"boundary": "Full epoch for SFT; completed rollout iteration with both optimizer updates for GRPO. Timings live outside resumable numerical state.",
},
)
Codificar ventajas, recorte y KL
Actualización en línea por grupo
Error que diagnosticar: usar las log-probs actuales separadas como antiguas en cada microbatch borra el historial; capturarlas durante la generación.
Invariante: se superan las pruebas de signos del recorte, grupos constantes y gradientes; la política se actualiza realmente.
Archivo src/grpo_recompense_verifiable/train.py
import 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_tokenizer,
reload_learned,
save_learned,
parameter_manifest,
manifest_digest,
)
from .scores import rollout_completion_row, model_scores
from .rollout import sample_group, verify_behavior_before_update
from .loss import group_advantages, clipped_group_surrogate, exact_prefix_kl
from .warmstart import assert_learning_scope, freeze_initial_reference, learning_rate
from .state import save_resume_state, load_resume_state
def next_prompt_indices(sampler, population, order, cursor, count):
selected = []
while len(selected) < count:
if cursor == len(order):
order = torch.randperm(population, generator=sampler).tolist()
cursor = 0
selected.append(order[cursor])
cursor += 1
return selected, order, cursor
def train_grpo(config, branch="grpo", mode="continuous"):
if branch not in ["grpo", "grpo-g2", "grpo-g8"] or mode not in [
"interrupt",
"continuous",
"resume",
]:
raise ValueError("Unknown GRPO branch or process")
group_size = (
config["group_size"] if branch == "grpo" else int(branch.rsplit("g", 1)[1])
)
iterations = (
config["iterations"] if branch == "grpo" else config["exercise_iterations"]
)
boundary = config["interrupt_iteration"] if branch == "grpo" else 8
if config["std_correction"] != 0 or config["updates_per_rollout"] != 2:
raise ValueError(
"Frozen variant requires population std and two updates per rollout"
)
assert_learning_scope()
seed_all(config["seed"])
tokenizer = load_tokenizer()
pages = read_json("data/generated/train/pages.json")
gold = {r["document_id"]: r for r in read_json("data/generated/labels/train.json")}
if (
len(pages) != 872
or any(p["split"] != "train" for p in pages)
or any(r["warmstart"] for r in gold.values())
):
raise ValueError("RL must use only the 872 remaining train problems")
model, metadata = reload_learned("assets/checkpoints/warmstart-resume.safetensors")
model = model.cuda().eval()
reference, ref_metadata = freeze_initial_reference()
parameters = list(model.parameters())
initial = parameter_manifest(model)
initial_digest = manifest_digest(initial)
if (
initial_digest != ref_metadata["model_digest"]
or sum(p.numel() for p in parameters) != 494032768
or any(not p.requires_grad or p.dtype != torch.float32 for p in parameters)
):
raise ValueError(
"Policy and fixed reference must start from the same full FP32 local SFT"
)
identity = {
"branch": branch,
"config": config,
"group_size": group_size,
"iterations": iterations,
"initial_model_digest": initial_digest,
"reference_sha256": sha("assets/checkpoints/warmstart-resume.safetensors"),
"reference_model_digest": ref_metadata["model_digest"],
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"variant": "GRPO pédagogique, KL exacte sur préfixes",
"data_sha256": {
n: sha("data/generated/" + n)
for n in ["train/pages.json", "labels/train.json"]
},
"code_sha256": {
n: sha("src/grpo_recompense_verifiable/" + n)
for n in [
"model.py",
"scores.py",
"loss.py",
"rollout.py",
"verifier.py",
"train.py",
"state.py",
]
},
}
optimizer = torch.optim.AdamW(
parameters, lr=config["learning_rate"], weight_decay=config["weight_decay"]
)
sampler = torch.Generator().manual_seed(config["seed"] + 100)
generation_rng = torch.Generator(device="cuda").manual_seed(config["seed"] + 200)
root = Path("assets/checkpoints")
destination = root / f"{branch}-{mode}.pt.gz"
if destination.exists():
raise ValueError("Preserve existing local GRPO state")
start = updates = cursor = 0
order = []
history = []
consumed = []
total_updates = iterations * config["updates_per_rollout"]
if mode == "resume":
state = load_resume_state(root / f"{branch}-interrupt.pt.gz")
if state["identity"] != identity:
raise ValueError("GRPO resume identity differs")
model.load_state_dict(state["model"], strict=True)
optimizer.load_state_dict(state["optimizer"])
sampler.set_state(state["sampler"])
generation_rng.set_state(state["generation_rng"])
restore_rng(state["rng"])
start = state["iteration"]
updates = state["updates"]
cursor = state["cursor"]
order = state["order"]
history = state["history"]
consumed = state["consumed_rollout_ids"]
if (
updates != start * config["updates_per_rollout"]
or state["pending_rollouts"] != []
):
raise ValueError("Resume only at a completed rollout boundary")
if state["scheduler"] != {
"next_update": updates,
"next_lr": learning_rate(
config, updates, total_updates, config["learning_rate"]
),
}:
raise ValueError("GRPO scheduler identity differs")
del state
stop = boundary if mode == "interrupt" else iterations
torch.cuda.reset_peak_memory_stats()
torch.cuda.synchronize()
begun = time.perf_counter()
generation_seconds = scoring_seconds = optimization_seconds = 0.0
for iteration in range(start, stop):
selected, order, cursor = next_prompt_indices(
sampler, len(pages), order, cursor, config["prompts_per_iteration"]
)
behavior_digest = manifest_digest(parameter_manifest(model))
groups = []
for prompt_position, index in enumerate(selected):
group_id = (
f"{branch}/iteration-{iteration:03d}/prompt-{prompt_position:02d}"
)
group = sample_group(
model,
tokenizer,
pages[index],
gold[pages[index]["document_id"]]["problem"],
config,
generation_rng,
behavior_digest,
group_id,
group_size=group_size,
capture_probe=iteration == 0 and prompt_position == 0,
)
generation_seconds += sum(
r["generation_seconds"] for r in group["responses"]
)
torch.cuda.synchronize()
score_start = time.perf_counter()
group["before_update"] = verify_behavior_before_update(model, group)
torch.cuda.synchronize()
scoring_seconds += time.perf_counter() - score_start
groups.append(group)
rollout_path = Path(
f"outputs/learning/rollouts/{branch}/{mode}/iteration-{iteration:03d}.json"
)
if rollout_path.exists():
raise ValueError("Existing actual rollout evidence must be preserved")
save_json(
rollout_path,
{
"branch": branch,
"mode": mode,
"iteration": iteration,
"behavior_model_digest": behavior_digest,
"groups": groups,
"reference_model_digest": identity["reference_model_digest"],
},
)
current_ids = [r["rollout_id"] for g in groups for r in g["responses"]]
if len(set(current_ids)) != len(current_ids) or set(current_ids) & set(
consumed
):
raise ValueError("Duplicate rollout would be optimized twice")
update_rows = []
torch.cuda.synchronize()
update_start = time.perf_counter()
for inner_update in range(config["updates_per_rollout"]):
lr = learning_rate(config, updates, total_updates, config["learning_rate"])
for parameter_group in optimizer.param_groups:
parameter_group["lr"] = lr
optimizer.zero_grad(set_to_none=True)
details = []
for group in groups:
for member, response in enumerate(group["responses"]):
row = rollout_completion_row(response)
with torch.no_grad():
_, _, reference_values, reference_logits, _ = model_scores(
reference, [row]
)
_, _, current_values, current_logits, targets = model_scores(
model, [row]
)
old = torch.tensor(
response["old_token_logprobs"],
device="cuda",
dtype=current_values.dtype,
).unsqueeze(0)
advantage = torch.tensor(
[group["advantages"][member]],
device="cuda",
dtype=current_values.dtype,
)
mask = torch.ones_like(old, dtype=torch.bool)
surrogate, audit = clipped_group_surrogate(
current_values.unsqueeze(0),
old,
advantage,
mask,
config["clip_epsilon"],
)
kl = exact_prefix_kl(
current_logits, reference_logits, config["kl_vocabulary_block"]
)
contribution = (
surrogate.mean() + config["kl_beta"] * kl.mean()
) / (len(groups) * group_size)
if not torch.isfinite(contribution) or not torch.isfinite(kl).all():
raise ValueError("Nonfinite actual GRPO objective")
detail = {
"rollout_id": response["rollout_id"],
"tokens": len(response["response_ids"]),
"old_token_logprobs": response["old_token_logprobs"],
"current_token_logprobs": current_values.detach()
.cpu()
.tolist(),
"reference_token_logprobs": reference_values.cpu().tolist(),
"advantages": float(advantage[0]),
"token_ratios": audit["ratio"][0].cpu().tolist(),
"surrogate": float(surrogate.detach()[0]),
"exact_prefix_kl": kl.detach().cpu().tolist(),
"active_clip_fraction": float(audit["active_clip_fraction"][0]),
"outside_interval_fraction": float(
audit["outside_interval_fraction"][0]
),
"scaled_contribution": float(contribution.detach()),
}
if iteration == 0 and inner_update == 0 and len(details) < 2:
detail["kl_probe"] = {
"current_logits": current_logits[0].detach().cpu().tolist(),
"reference_logits": reference_logits[0].cpu().tolist(),
"target": int(targets[0]),
"exact_kl": float(kl.detach()[0]),
}
contribution.backward()
details.append(detail)
del (
current_logits,
reference_logits,
current_values,
reference_values,
kl,
contribution,
surrogate,
)
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("Full policy gradients missing")
if (
any(
p.requires_grad or p.grad is not None
for p in reference.parameters()
)
or reference.training
or model.training
):
raise ValueError("Reference/dropout invariant broken")
optimizer.step()
updates += 1
update_rows.append(
{
"update": updates,
"inner_update": inner_update,
"learning_rate": lr,
"gradient_norm_before_clip": float(norm),
"prompt_denominator": len(groups),
"group_denominator": group_size,
"surrogate": sum(r["surrogate"] for r in details) / len(details),
"mean_exact_kl": sum(
sum(r["exact_prefix_kl"]) / r["tokens"] for r in details
)
/ len(details),
"active_clip_fraction": sum(
r["active_clip_fraction"] for r in details
)
/ len(details),
"total_loss": sum(r["scaled_contribution"] for r in details),
"responses": details,
}
)
torch.cuda.synchronize()
optimization_seconds += time.perf_counter() - update_start
consumed.extend(current_ids)
# Keep timing out of resumable state: wall-clock durations cannot be equal
# across processes. Actual raw rollouts and timings are retained above.
summary = {
"iteration": iteration,
"behavior_model_digest": behavior_digest,
"document_ids": [pages[i]["document_id"] for i in selected],
"group_size": group_size,
"groups": [
{
"group_id": g["group_id"],
"rewards": g["rewards"],
"advantages": g["advantages"],
"population_std": g["population_std"],
"zero_variance": g["zero_variance"],
"before_update": g["before_update"],
"response_ids": [r["response_ids"] for r in g["responses"]],
}
for g in groups
],
"updates": update_rows,
}
history.append(summary)
print(
{
"branch": branch,
"iteration": iteration + 1,
"updates": updates,
"reward_mean": sum(g["reward_mean"] for g in groups) / len(groups),
"zero_variance_fraction": sum(g["zero_variance"] for g in groups)
/ len(groups),
"kl": update_rows[-1]["mean_exact_kl"],
},
flush=True,
)
torch.cuda.synchronize()
elapsed = time.perf_counter() - begun
peak_memory = torch.cuda.max_memory_allocated()
final = parameter_manifest(model)
final_digest = manifest_digest(final)
ref_final = manifest_digest(parameter_manifest(reference))
if (
ref_final != identity["reference_model_digest"]
or sha("assets/checkpoints/warmstart-resume.safetensors")
!= identity["reference_sha256"]
):
raise ValueError("Frozen reference changed")
# Zero reward variation can leave a policy unchanged; this is a valid measured
# limitation. Never require a fabricated improvement or weight change here.
state = {
"identity": identity,
"iteration": stop,
"updates": updates,
"model": model.state_dict(),
"optimizer": optimizer.state_dict(),
"sampler": sampler.get_state(),
"generation_rng": generation_rng.get_state(),
"order": order,
"cursor": cursor,
"scheduler": {
"next_update": updates,
"next_lr": learning_rate(
config, updates, total_updates, config["learning_rate"]
),
},
"rng": rng_state(),
"history": history,
"consumed_rollout_ids": consumed,
"pending_rollouts": [],
}
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,
"iterations": stop,
"updates": updates,
"history": history,
"consumed_rollout_ids": consumed,
"initial_parameters": initial,
"final_parameters": final,
"initial_model_digest": initial_digest,
"final_model_digest": final_digest,
"reference_model_digest_after": ref_final,
"weights_changed": final_digest != initial_digest,
"trainable_parameters": 494032768,
"reference_trainable_parameters": 0,
"reference_gradients": 0,
"checkpoint_sha256": sha(destination),
"checkpoint_bytes": destination.stat().st_size,
"checkpoint_compression": compression,
"generation_seconds_this_process": generation_seconds,
"behavior_scoring_seconds_this_process": scoring_seconds,
"optimization_and_reference_scoring_seconds_this_process": optimization_seconds,
"total_loop_seconds_this_process": elapsed,
"max_torch_allocated_bytes": peak_memory,
"test_visible": False,
"dropout_active": False,
},
)
Evaluar la calidad y el coste total
Pass@1 y ausencia de señal
Error que diagnosticar: publicar la mejor muestra como pass@1; indicar el presupuesto exacto de muestreo.
Invariante: 200 ID por rama, salidas no válidas conservadas y coste de generación no omitido.
Archivo src/grpo_recompense_verifiable/evaluate.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, manifest_digest, parameter_manifest
from .rollout import sample_response
from .verifier import score_generated_json
MAIN = ["warmstart", "sft", "grpo"]
EXERCISE = ["grpo-g2", "grpo-g8"]
@torch.no_grad()
def generate_policy(config, branch):
if branch not in MAIN + EXERCISE:
raise ValueError("Unknown evaluation policy")
if Path("data/generated/labels").exists() or Path("outputs/learning").exists():
raise ValueError("Inference must not mount labels or learning outputs")
seed_all(config["seed"])
tokenizer = load_tokenizer()
model, metadata = reload_learned(f"assets/checkpoints/{branch}-resume.safetensors")
model = model.cuda().eval()
pages = [
p for p in read_json("data/generated/input/pages.json") if p["split"] == "test"
]
if len(pages) != 200:
raise ValueError("Frozen test population changed")
root = Path("outputs/inference") / branch
if root.exists():
raise ValueError("Preserve existing actual predictions")
torch.cuda.reset_peak_memory_stats()
torch.cuda.synchronize()
begun = time.perf_counter()
files = {}
for mode in ["greedy", "sample"]:
results = []
for index, page in enumerate(pages):
for draw in range(1 if mode == "greedy" else config["evaluation_draws"]):
# Fixed independent draws, common seeds across policies; no
# selection of the best response and no label-dependent retry.
seed = 480000 + index * config["evaluation_draws"] + draw
generator = torch.Generator(device="cuda").manual_seed(seed)
row = sample_response(
model,
tokenizer,
page,
config,
generator,
mode=mode,
policy_id=metadata["model_digest"],
draw=draw,
capture_probe=index < 2 and draw == 0,
)
row["sampling_seed"] = seed if mode == "sample" else None
row["population"] = page["population"]
results.append(row)
if (index + 1) % 50 == 0:
print(
{"policy": branch, "mode": mode, "problems": index + 1}, flush=True
)
path = root / f"{mode}.json"
save_json(
path,
{
"branch": branch,
"mode": mode,
"problems": 200,
"draws_per_problem": (
1 if mode == "greedy" else config["evaluation_draws"]
),
"model_digest": metadata["model_digest"],
"rows": results,
"labels_visible": False,
},
)
files[mode] = {"sha256": sha(path), "responses": len(results)}
torch.cuda.synchronize()
elapsed = time.perf_counter() - begun
if manifest_digest(parameter_manifest(model)) != metadata["model_digest"]:
raise ValueError("Evaluation changed the policy")
save_json(
root / "audit.json",
{
"branch": branch,
"model_digest": metadata["model_digest"],
"seconds": elapsed,
"max_torch_allocated_bytes": torch.cuda.max_memory_allocated(),
"files": files,
"labels_visible": False,
"generation_support": "full vocabulary; temperature1; no top-k/top-p truncation; no JSON constraint",
},
)
def metrics(rows, labels):
result = {}
for population in ["ALL", "ID50", "OOD100"]:
selected = [
r for r in rows if population == "ALL" or r["population"] == population
]
scored = []
for row in selected:
observed = score_generated_json(
row["text"],
row["finish_reason"],
labels[row["document_id"]]["problem"],
row["tokenization_valid"],
)
scored.append(
{
"document_id": row["document_id"],
"draw": row["draw"],
"response_tokens": row["response_tokens"],
"finish_reason": row["finish_reason"],
**observed,
}
)
n = len(scored)
if n == 0:
raise ValueError("A declared evaluation population is empty")
result[population] = {
"responses": n,
"problems": len({r["document_id"] for r in scored}),
"correct": sum(r["reward"] for r in scored),
"success_rate": sum(r["reward"] for r in scored) / n,
"valid_json": sum(r["valid"] for r in scored),
"json_validity": sum(r["valid"] for r in scored) / n,
"without_eos": sum(r["finish_reason"] != "eos" for r in scored),
"mean_response_tokens": sum(r["response_tokens"] for r in scored) / n,
"rows": scored,
}
return result
def evaluate(config, exercise=False):
branches = EXERCISE if exercise else MAIN
labels = {r["document_id"]: r for r in read_json("data/generated/labels/gold.json")}
results = {}
for branch in branches:
root = Path("outputs/inference") / branch
greedy = read_json(root / "greedy.json")
sample = read_json(root / "sample.json")
if (
len(greedy["rows"]) != 200
or len(sample["rows"]) != 200 * config["evaluation_draws"]
):
raise ValueError("Incomplete actual evaluation")
for source, draws in [(greedy, 1), (sample, config["evaluation_draws"])]:
keys = [(r["document_id"], r["draw"]) for r in source["rows"]]
if len(set(keys)) != len(keys) or any(
sum(r["document_id"] == identifier for r in source["rows"]) != draws
for identifier, _ in keys
):
raise ValueError("Missing/duplicated test draw")
results[branch] = {
"greedy_pass_at_1": metrics(greedy["rows"], labels),
"empirical_per_draw_success": metrics(sample["rows"], labels),
"draws_per_problem": config["evaluation_draws"],
"best_of_draws_reported_as_pass_at_1": False,
"generation": read_json(root / "audit.json"),
}
save_json(
"outputs/exercise/results.json" if exercise else "outputs/results.json",
{
"policies": results,
"deterministic_solver": "Exact on all generated problems; a production stock calculation does not need a language model",
"shared_templates": "Synthetic controlled arithmetic, not real operational evidence",
"quality_gain_required": False,
},
)
return results
Interpretar grupos sin ganador
Señal nula sin fabricar una recompensa
Error que diagnosticar: está prohibido añadir una bonificación arbitraria para forzar una curva descendente; registrar el resultado nulo.
Invariante: ventajas nulas para los grupos constantes; ninguna modificación del verificador.
Archivo src/grpo_recompense_verifiable/exercise.py
from pathlib import Path
import torch
from .config import read_json, save_json, sha
from .loss import group_advantages, clipped_group_surrogate
def main_artifacts():
names = ["configs/minimum.yaml", "outputs/results.json"]
for branch in ["warmstart", "sft", "grpo"]:
names += [
f"assets/checkpoints/{branch}-resume.safetensors",
f"outputs/learning/resume-{branch}.json",
f"outputs/inference/{branch}/greedy.json",
f"outputs/inference/{branch}/sample.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 run_group_size_ablation(config):
protected = read_json("outputs/exercise/protected.json")
if protected != {name: sha(name) for name in main_artifacts()}:
raise ValueError("Exercise changed the main artifacts")
analytic = []
for rewards in [[0.0, 0.0, 0.0, 0.0], [1.0, 1.0, 1.0, 1.0], [1.0, 0.0, 1.0, 0.0]]:
advantages, mean, std = group_advantages(
torch.tensor([rewards], dtype=torch.float64), config["advantage_epsilon"]
)
current = torch.full((4, 2), -1.0, dtype=torch.float64, requires_grad=True)
old = current.detach().clone()
loss, _ = clipped_group_surrogate(
current, old, advantages[0], torch.ones_like(current, dtype=torch.bool)
)
loss.mean().backward()
analytic.append(
{
"rewards": rewards,
"advantages": advantages[0].tolist(),
"population_std": float(std[0, 0]),
"reward_surrogate": float(loss.detach().mean()),
"token_gradients": current.grad.tolist(),
"source": "Constructed illustrative rewards, not actual model outputs",
}
)
runs = {
b: read_json(f"outputs/learning/train-{b}-continuous.json")
for b in ["grpo-g2", "grpo-g8"]
}
if any(r["updates"] != 40 or r["iterations"] != 20 for r in runs.values()):
raise ValueError("Frozen group-size ablation budget changed")
if len({r["initial_model_digest"] for r in runs.values()}) != 1:
raise ValueError("Ablation initialization differs")
if [[h["document_ids"] for h in r["history"]] for r in runs.values()][0] != [
[h["document_ids"] for h in r["history"]] for r in runs.values()
][1]:
raise ValueError("Ablation prompt sequence differs")
budgets = {
b: {
"updates": r["updates"],
"prompts": sum(len(h["document_ids"]) for h in r["history"]),
"rollouts": len(r["consumed_rollout_ids"]),
"seconds": r["total_loop_seconds_this_process"],
}
for b, r in runs.items()
}
result = read_json("outputs/exercise/results.json")
if set(result["policies"]) != {"grpo-g2", "grpo-g8"}:
raise ValueError("Missing actual heldout ablation generation")
save_json(
"outputs/exercise/group-size.json",
{
"analytic_constant_groups": analytic,
"actual_fresh_training": True,
"budgets": budgets,
"same_initialization_and_prompt_order": True,
"equal_compute_claimed": False,
"main_100_iteration_g4_is_not_a_controlled_group_size_only_comparison": True,
"main_artifacts_unchanged": protected
== {name: sha(name) for name in main_artifacts()},
"protected_sha256": protected,
"verifier_changed": False,
"quality_gain_required": False,
},
)
Fuentes
DeepSeekMath v3 describe el objetivo relativo de grupo. TRL v0.23.1 documenta diferentes reducciones y variantes. Estas fuentes aclaran las elecciones; no se reproduce ningún resultado publicado ni un trainer completo. Base Qwen fijada.
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",
"metrics": {
"warmstart": {
"greedy_pass_at_1": {
"responses": 200,
"problems": 200,
"correct": 3,
"success_rate": 0.015,
"valid_json": 200,
"json_validity": 1.0,
"without_eos": 0,
"mean_response_tokens": 7.85
},
"empirical_per_draw_success": {
"responses": 800,
"problems": 200,
"correct": 15,
"success_rate": 0.01875,
"valid_json": 799,
"json_validity": 0.99875,
"without_eos": 0,
"mean_response_tokens": 7.78125
}
},
"sft": {
"greedy_pass_at_1": {
"responses": 200,
"problems": 200,
"correct": 16,
"success_rate": 0.08,
"valid_json": 200,
"json_validity": 1.0,
"without_eos": 0,
"mean_response_tokens": 7.865
},
"empirical_per_draw_success": {
"responses": 800,
"problems": 200,
"correct": 30,
"success_rate": 0.0375,
"valid_json": 799,
"json_validity": 0.99875,
"without_eos": 0,
"mean_response_tokens": 7.81125
}
},
"grpo": {
"greedy_pass_at_1": {
"responses": 200,
"problems": 200,
"correct": 3,
"success_rate": 0.015,
"valid_json": 200,
"json_validity": 1.0,
"without_eos": 0,
"mean_response_tokens": 7.825
},
"empirical_per_draw_success": {
"responses": 800,
"problems": 200,
"correct": 14,
"success_rate": 0.0175,
"valid_json": 799,
"json_validity": 0.99875,
"without_eos": 0,
"mean_response_tokens": 7.74375
}
},
"grpo-g2": {
"greedy_pass_at_1": {
"responses": 200,
"problems": 200,
"correct": 4,
"success_rate": 0.02,
"valid_json": 200,
"json_validity": 1.0,
"without_eos": 0,
"mean_response_tokens": 7.9
},
"empirical_per_draw_success": {
"responses": 800,
"problems": 200,
"correct": 11,
"success_rate": 0.01375,
"valid_json": 800,
"json_validity": 1.0,
"without_eos": 0,
"mean_response_tokens": 7.805
}
},
"grpo-g8": {
"greedy_pass_at_1": {
"responses": 200,
"problems": 200,
"correct": 3,
"success_rate": 0.015,
"valid_json": 200,
"json_validity": 1.0,
"without_eos": 0,
"mean_response_tokens": 7.96
},
"empirical_per_draw_success": {
"responses": 800,
"problems": 200,
"correct": 13,
"success_rate": 0.01625,
"valid_json": 799,
"json_validity": 0.99875,
"without_eos": 0,
"mean_response_tokens": 7.79875
}
}
},
"training_branches": 5,
"full_state_resume_checks": 5,
"actual_test_generations": 5000,
"scope": "Scalar reward/ratio/categorical-probe/metrics plus captured full-state comparisons; no scalar full Transformer or CUDA multinomial replay claimed",
"quality_gain_required": false
}
