Un esercizio su un inventario fittizio descrive entrate e uscite di lotti. Il modello deve restituire un piccolo JSON con lo stock finale intero. Un verificatore indipendente calcola la risposta a partire dai dati strutturati, senza eseguire il testo del modello.
Osservate quattro risposte allo stesso problema, le loro ricompense, i vantaggi relativi e un aggiornamento. Confrontate il warm start SFT, l'SFT aggiuntivo e GRPO su nuovi problemi, includendo il costo delle generazioni.
Scaricare il progetto completo, revisione r001
La risposta deve corrispondere al calcolo
StockArithmetic-v1 contiene 1400 problemi: 1000 train, 200 validation e 200 test. Ogni problema fornisce uno stock iniziale, due consegne e due prelievi; tutti gli stock intermedi restano non negativi. Le tuple vengono deduplicate dopo aver ordinato separatamente consegne e prelievi, così una permutazione degli stessi movimenti non attraversa le partizioni. Gli schemi linguistici e le operazioni aritmetiche sono condivisi: questo piccolo corpus non rappresenta inventari reali.
Gli operandi train e validation sono compresi tra 0 e 50. Il test contiene 100 problemi in questo intervallo e 100 con almeno un operando tra 51 e 100, contrassegnati OOD100. I risultati di queste due popolazioni sono separati. I primi 128 target train servono al warm start; gli altri 872 servono all'SFT aggiuntivo o alle estrazioni di prompt RL. I target sono calcolati, non annotati da una persona né prodotti da un LLM.
Il modello vede soltanto il testo del problema e un'istruzione sul formato. Il verificatore riceve separatamente la struttura del problema. Richiede un solo oggetto JSON completo, limitato a 256 caratteri, che contenga esattamente final_stock con un intero tra 0 e 300. Booleani, float, campi aggiuntivi, duplicati, Markdown, più oggetti e numeri non finiti vengono rifiutati. Anche un ID generato assente dal tokenizer invalida la risposta: non può scomparire silenziosamente durante la decodifica. Nessun testo generato viene eseguito. Una risposta interrotta al limite di token riceve zero, anche se il suo inizio contiene un JSON corretto.
La ricompensa vale uno per una risposta completa corretta, zero in tutti gli altri casi. Non esistono premi di stile, ricompense parziali né decodifica JSON vincolata. Il risolutore deterministico sa già calcolare tutti i target: un apprendimento costoso può essere inutile per questo compito.
Tre politiche diverse
Qwen2.5-0.5B-Instruct viene acquisito alla revisione 7ae557604adf67be50417f59c2c2f167def9a775. Tutti i suoi 494032768 parametri unici sono ottimizzati in FP32, con attenzione eager e TF32 disattivato. L'architettura e i pesi condivisi tra embeddings e output vengono controllati. I logits predicono il token successivo; vengono supervisionati solo i token di risposta, con EOS 151645 incluso. Il prompt e il padding sono mascherati per posizione. Il prompt contiene al massimo 256 token, la risposta 64, per un totale di 320; nessun superamento negli esempi supervisionati viene troncato silenziosamente.
Il warm start SFT apprende 128 esempi per tre epoche, cioè 48 aggiornamenti di otto esempi. L'SFT aggiuntivo riparte da questo stato e apprende 872 esempi per due epoche, cioè 218 aggiornamenti. Entrambi usano AdamW a 2e−5. La CE viene normalizzata su tutti i token attivi dell'accumulo, non come media delle medie delle risposte.
GRPO riparte dallo stesso warm start. Il riferimento fisso è una copia di questo stato ricaricata in modo rigoroso, congelata e senza gradienti. La vecchia politica designa lo stato che ha prodotto le risposte di un'iterazione; le sue log-probabilità sono catturate alla generazione e restano scollegate. La politica attuale riceve i gradienti. Non esiste alcun critico del valore appreso. Questi tre ruoli non sono intercambiabili.
Dal gruppo al vantaggio
Un'iterazione seleziona quattro prompt train. Per ciascuno campiona G=4 risposte sequenziali a temperatura 1, senza top-k, troncamento top-p né soppressione di token dal vocabolario. Le probabilità di comportamento provengono dai logits reali al momento del campionamento. Token, punteggi, stato della politica, motivi di arresto e tempi vengono conservati. La modalità eval disattiva il dropout durante la generazione E durante gli aggiornamenti; eval non vieta autograd.
In ogni gruppo si calcola A=(r−media)/(deviazione standard+1e−8), con deviazione standard della popolazione, correction=0. Ricompense e vantaggi sono scollegati. Per [1,0,1,0], i vantaggi sono circa [1,−1,1,−1]; l'epsilon spiega perché non siano esattamente questi interi. Se le quattro ricompense sono identiche, tutti i vantaggi sono esattamente zero. Misuriamo la frequenza di questi gruppi senza inventare un vincitore.
A ogni token di risposta, rho=exp(logp_attuale−logp_vecchia). Prima del primo aggiornamento, lo scoring dell'intera sequenza deve ritrovare i punteggi catturati sui prefissi entro 2e−4; le forme dei kernel possono produrre una piccola differenza numerica. Il punteggio vecchio non viene mai sostituito da uno scollegamento dei nuovi logits dopo l'apprendimento.
Il termine da massimizzare è min(rho A, clip(rho,0,8,1,2) A). Il segno di A è essenziale: i test controllano entrambi i lati e i gradienti dei rami sottoposti a clipping. La loss è l'opposto di questo termine, mediata prima sui token di ogni risposta, poi sulle risposte e infine sui prompt. Una risposta lunga non ottiene automaticamente più peso complessivo; questa normalizzazione conserva tuttavia bias di lunghezza e difficoltà che un piccolo esperimento non elimina.
La KL esatta sui prefissi visitati
La variante si chiama « GRPO didattico, KL esatta sui prefissi ». Per ogni prefisso effettivamente visitato, calcoliamo somma_v p_attuale(v) × [log p_attuale(v)−log p_riferimento(v)] sull'intero vocabolario. Lo percorriamo in blocchi di 4096, con i normalizzatori logsumexp completi. Normalizzare ogni blocco separatamente cambierebbe la divergenza.
Questa somma categoriale è esatta fino all'arrotondamento per ogni prefisso. I prefissi restano quelli dei rollout: non enumera tutte le sequenze possibili. Il coefficiente vale 0,02. La sua riduzione segue la stessa media token/risposta/prompt della surrogate. Valori e gradienti su piccoli tensori vengono confrontati con kl_div di PyTorch in FP64 e FP32; alcune distribuzioni complete reali servono da sonde indipendenti.
L'articolo usa una stima diversa della penalità KL. TRL propone diverse riduzioni, tra cui DAPO; non rivendichiamo né i suoi default né l'identità del suo trainer. Il tutorial implementa esplicitamente la propria variante e non installa TRL. I gruppi a vantaggio nullo non hanno gradiente di ricompensa, ma la regolarizzazione KL e le regole dell'ottimizzatore possono ancora agire.
Budget, costi e ripresa
Il minimo GRPO esegue 100 iterazioni, quattro prompt per iterazione, quattro risposte per prompt e due aggiornamenti su ogni insieme di rollout: 1600 risposte reali e 200 aggiornamenti. AdamW usa lr=1e−6, weight_decay=0,01, clipping del gradiente a 1, warm-up del 10% poi decrescita lineare fino al 10% del tasso massimo. Warm start, SFT e RL non consumano gli stessi esempi, token, forward né secondi GPU; non viene ipotizzata alcuna uguaglianza di costo.
La generazione senza cache KV privilegia la leggibilità dello scoring. I tempi di generazione, verifica dei vecchi punteggi, ottimizzazione con riferimento, ciclo completo e comandi sono distinti. Anche le impronte dello stato hanno un costo. I costi del confronto continuo/ripresa devono restare visibili oltre a quelli dell'addestramento continuo di riferimento.
La ripresa GRPO avviene dopo un'iterazione completamente ottimizzata, cioè dopo i due aggiornamenti. Modello completo, AdamW, scheduler, permutazione/cursore dei prompt, RNG di generazione, RNG di sistema e lista dei rollout consumati vengono salvati. L'interruzione a 40, il percorso continuo a 100 e la ripresa delle 60 rimanenti devono concludersi in modo esattamente identico. Un rollout non può essere consumato due volte. Le durate reali sono fuori dallo stato numerico comparabile, ma le loro prove vengono conservate. Anche SFT confronta l'interruzione dopo un'epoca, il percorso continuo e la ripresa.
I quindici stati completi dei cinque rami restano locali in gzip livello 1 senza perdita; vengono inoltre ricaricati cinque modelli finali in Safetensors. Le dimensioni vengono misurate, senza promettere alcun tasso di compressione. Caricare oggetti Torch è consentito qui solo perché sono stati prodotti localmente; non sostituiteli con un file di terzi.
Valutare una politica, non selezionare la sua risposta migliore
Le politiche warm start, SFT e GRPO generano ciascuna una risposta greedy sugli stessi 200 problemi test: è il pass@1 greedy. Producono inoltre quattro campioni stocastici indipendenti per problema con seed fissati. Il tasso di successo per estrazione usa le 800 risposte; non è il successo della migliore delle quattro. Le popolazioni ID50, OOD100 e il loro totale restano distinti.
Validità JSON, esattezza, assenza di EOS, lunghezze e tempi vengono calcolati su tutte le uscite. La validazione SFT misura la NLL senza scegliere un checkpoint; gli stati finali sono fissati prima delle prove. Le risposte test non entrano negli aggiornamenti. Una diminuzione della loss non prova un guadagno di esattezza, e un risultato sfavorevole a GRPO è pubblicabile.
Esercizio: due dimensioni di gruppo
I gruppi costanti vengono dapprima studiati su ricompense costruite, chiaramente separate dalle risposte del modello. Poi G=2 e G=8 vengono realmente addestrati per 20 iterazioni dallo stesso warm start, con le stesse 80 presentazioni di prompt e 40 aggiornamenti. Producono rispettivamente 160 e 640 rollout: lo stesso numero di aggiornamenti non significa lo stesso costo. Le due politiche vengono valutate sugli stessi 200 problemi e quattro estrazioni.
Il G=4 principale ha 100 iterazioni. Confrontarlo direttamente con i rami da 20 iterazioni non sarebbe un'ablazione che isola soltanto la dimensione del gruppo; questo limite viene indicato. L'esercizio protegge gli SHA-256 dei pesi, dei risultati e della configurazione principale. Il verificatore resta invariato, anche se tutti i gruppi sono errati.
Scrivere il problema e il suo verificatore indipendente
Ricompensa che non dipende dal modello
Errore da diagnosticare: un output contiene due JSON, di cui uno corretto: esigere una sola risposta completa.
Invariante: 1 000/200/200 problemi distinti; risolutore esatto su fixture e casi limite.
File 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,
}
Formare un gruppo e ispezionare i suoi punteggi
Rollout e probabilità di comportamento
Errore da diagnosticare: un dropout attivo rende incoerenti le log-probabilità: impostare la politica in modalità valutazione consentendo al contempo l'autograd.
Invariante: ogni gruppo ha un prompt_id univoco e quattro risposte con scoring compatibile.
File 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,
}
Avviare e confrontare un SFT diretto
Riferimento di apprendimento senza RL
Errore da diagnosticare: scaricare un checkpoint di un vecchio tutorial sostituisce il minimo: rifare il warm start in questo progetto.
Invariante: nessun esempio test nel warm start, costi e hash iniziale catturati.
File 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.",
},
)
Codificare vantaggi, clipping e KL
Aggiornamento online per gruppo
Errore da diagnosticare: usare le log-probabilità attuali scollegate come old a ogni microbatch cancella la storia: catturarle al momento della generazione.
Invariante: i test dei segni del clipping, dei gruppi costanti e dei gradienti riescono; la politica viene realmente aggiornata.
File 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,
},
)
Valutare qualità e costo totale
Pass@1 e assenza di segnale
Errore da diagnosticare: pubblicare la migliore estrazione come pass@1: indicare il budget di campionamento esatto.
Invariante: 200 ID per ramo, uscite non valide conservate e costo di generazione non omesso.
File 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
Interpretare i gruppi senza vincitore
Segnale nullo senza fabbricare una ricompensa
Errore da diagnosticare: aggiungere un bonus arbitrario per forzare una curva discendente è vietato: registrare il risultato nullo.
Invariante: vantaggi nulli per i gruppi costanti; nessuna modifica del verificatore.
File 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,
},
)
Fonti
DeepSeekMath v3 descrive l'obiettivo relativo di gruppo. TRL v0.23.1 documenta diverse riduzioni e varianti. Queste fonti chiariscono le scelte; non vengono riprodotti né risultati pubblicati né un trainer completo. Base Qwen fissata.
Osservazioni di sviluppo
Valori ricalcolati dalle uscite grezze di questo percorso. Non sono soglie di guadagno; il percorso dallo ZIP li ricalcola sulle stesse popolazioni.
{
"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
}
