Un exercice d'inventaire fictif décrit des entrées et sorties de lots. Le modèle doit rendre un petit JSON avec le stock final entier. Un vérificateur indépendant calcule la réponse depuis les données structurées, sans exécuter le texte du modèle.
Visualiser quatre réponses au même problème, leurs récompenses, les avantages relatifs et une mise à jour. Comparer SFT d'amorçage, SFT supplémentaire et GRPO sur de nouveaux problèmes, avec le coût des générations.
Télécharger le projet complet, révision r001
La réponse doit correspondre au calcul
StockArithmetic-v1 contient 1400 problèmes : 1000 train, 200 validation et 200 test. Chaque problème donne un stock initial, deux livraisons puis deux retraits ; tous les stocks intermédiaires restent non négatifs. Les tuples sont dédupliqués après tri séparé des livraisons et retraits, afin qu’une permutation des mêmes mouvements ne traverse pas les partitions. Les patrons de langage et les opérations arithmétiques sont partagés : ce petit corpus ne représente pas des inventaires réels.
Les opérandes train et validation sont compris entre 0 et 50. Le test contient 100 problèmes de cette plage et 100 avec au moins un opérande entre 51 et 100, marqués OOD100. Les résultats de ces deux populations sont séparés. Les 128 premières cibles train servent à l’amorçage ; les 872 autres servent au SFT supplémentaire ou aux tirages de prompts RL. Les cibles sont calculées, pas annotées par un humain ni produites par un LLM.
Le modèle voit uniquement l’énoncé et une instruction de format. Le vérificateur reçoit séparément la structure du problème. Il exige un seul objet JSON entier, borné à 256 caractères, contenant exactement final_stock avec un entier entre 0 et 300. Booléens, flottants, champs supplémentaires, doublons, Markdown, plusieurs objets et nombres non finis sont rejetés. Un ID généré absent du tokenizer invalide aussi la réponse ; il ne peut pas disparaître silencieusement au décodage. Aucun texte généré n’est exécuté. Une réponse arrêtée à la limite de tokens reçoit zéro, même si son début contient un JSON correct.
La récompense vaut un pour une réponse complète correcte, zéro dans tous les autres cas. Il n’y a ni prime de style, ni récompense partielle, ni décodage JSON contraint. Le solveur déterministe sait déjà calculer toutes les cibles : un apprentissage coûteux peut être inutile pour cette tâche.
Trois politiques différentes
Qwen2.5-0.5B-Instruct est acquis à la révision 7ae557604adf67be50417f59c2c2f167def9a775. Ses 494032768 paramètres uniques sont tous optimisés en FP32, avec attention eager et TF32 désactivé. L’architecture et les poids liés embeddings/sortie sont contrôlés. Les logits prédisent le token suivant ; seuls les tokens de réponse sont supervisés, avec EOS 151645 inclus. Le prompt et le padding sont masqués par position. Le prompt a au plus 256 tokens, la réponse 64, le total 320 ; aucun dépassement des exemples supervisés n’est tronqué silencieusement.
L’amorçage SFT apprend 128 exemples pendant trois époques, soit 48 mises à jour de huit exemples. Le SFT supplémentaire repart de cet état et apprend 872 exemples pendant deux époques, soit 218 mises à jour. Les deux utilisent AdamW à 2e−5. La CE est normalisée sur tous les tokens actifs de l’accumulation, pas comme une moyenne de moyennes de réponses.
GRPO repart du même amorçage. La référence fixe est une copie strictement rechargée de cet état, gelée et sans gradients. L’ancienne politique désigne l’état qui a produit les réponses d’une itération ; ses log-probabilités sont capturées à la génération et restent détachées. La politique actuelle reçoit les gradients. Il n’y a aucun critique de valeur appris. Ces trois rôles ne sont pas interchangeables.
Du groupe à l’avantage
Une itération choisit quatre prompts train. Pour chacun, elle échantillonne G=4 réponses séquentielles à température 1, sans top-k, sans troncature top-p et sans suppression de token du vocabulaire. Les probabilités de comportement proviennent des logits réels au moment du tirage. Les tokens, scores, état de politique, motifs d’arrêt et temps sont conservés. Le mode eval désactive le dropout pendant la génération ET pendant les mises à jour ; eval n’interdit pas autograd.
Dans chaque groupe, on calcule A=(r−moyenne)/(écart-type+1e−8), avec écart-type de population, correction=0. Les récompenses et avantages sont détachés. Pour [1,0,1,0], les avantages sont environ [1,−1,1,−1] ; l’epsilon explique qu’ils ne soient pas exactement ces entiers. Si les quatre récompenses sont identiques, tous les avantages sont exactement zéro. On mesure la fréquence de ces groupes sans leur inventer un gagnant.
À chaque token de réponse, rho=exp(logp_actuel−logp_ancien). Avant la première mise à jour, le scoring de la séquence entière doit retrouver les scores capturés aux préfixes à 2e−4 près ; les formes de kernels peuvent produire un petit écart numérique. Le score ancien n’est jamais remplacé par un détachement des nouveaux logits après apprentissage.
Le terme à maximiser est min(rho A, clip(rho,0,8,1,2) A). Le signe de A est essentiel : les tests contrôlent les deux côtés et les gradients des branches clippées. La perte est l’opposé de ce terme, moyenné d’abord sur les tokens de chaque réponse, puis sur les réponses et enfin sur les prompts. Une réponse longue n’obtient pas automatiquement davantage de poids total ; cette normalisation comporte néanmoins des biais de longueur et de difficulté qu’une petite expérience ne supprime pas.
La KL exacte sur les préfixes visités
La variante s’appelle « GRPO pédagogique, KL exacte sur préfixes ». Pour chaque préfixe effectivement visité, on calcule somme_v p_actuel(v) × [log p_actuel(v)−log p_référence(v)] sur tout le vocabulaire. On parcourt celui-ci par blocs de 4096, avec les normalisateurs logsumexp complets. Normaliser chaque bloc séparément changerait la divergence.
Cette somme catégorielle est exacte à l’arrondi près pour chaque préfixe. Les préfixes restent ceux des rollouts : elle n’énumère pas toutes les séquences possibles. Le coefficient vaut 0,02. Sa réduction suit la même moyenne tokens/réponse/prompt que la surrogate. Des valeurs et gradients sur de petits tenseurs sont comparés à kl_div de PyTorch en FP64 et FP32 ; quelques distributions complètes réelles servent de sondes indépendantes.
Le papier utilise une estimation différente de la pénalité KL. TRL propose plusieurs réductions, dont DAPO ; nous ne revendiquons ni ses defaults ni l’identité de son trainer. Le TP implémente explicitement sa variante et n’installe pas TRL. Les groupes à avantage nul n’ont aucun gradient de récompense, mais la régularisation KL et les règles d’optimiseur peuvent encore agir.
Budget, coûts et reprise
Le minimum GRPO fait 100 itérations, quatre prompts par itération, quatre réponses par prompt et deux mises à jour sur chaque ensemble de rollouts : 1600 réponses réelles et 200 mises à jour. AdamW utilise lr=1e−6, weight_decay=0,01, clipping du gradient à 1, warm-up de 10 % puis décroissance linéaire jusqu’à 10 % du taux maximal. L’amorçage, le SFT et le RL ne consomment pas les mêmes exemples, tokens, forwards ni GPU-secondes ; aucune égalité de coût n’est supposée.
La génération sans cache KV privilégie la lisibilité du scoring. Les temps de génération, vérification des scores anciens, optimisation avec référence, boucle complète et commandes sont distingués. Les empreintes d’état ont aussi un coût. Les coûts de la comparaison continu/reprise doivent rester visibles en plus de ceux de l’entraînement continu de référence.
La reprise GRPO intervient après une itération entièrement optimisée, soit après les deux mises à jour. Modèle complet, AdamW, scheduler, permutation/cursor des prompts, RNG de génération, RNG système et liste des rollouts consommés sont sauvegardés. Interruption à 40, continu à 100 et reprise des 60 restantes doivent finir exactement identiques. Un rollout ne peut pas être consommé deux fois. Les durées réelles sont hors état numérique comparable, mais leurs preuves sont conservées. SFT compare aussi interruption après une époque, continu et reprise.
Les quinze états complets des cinq branches restent locaux en gzip niveau 1 sans perte ; cinq modèles finaux sont aussi rechargés en Safetensors. Les tailles sont mesurées, aucun taux de compression n’est promis. Charger les objets Torch n’est permis ici que parce qu’ils ont été produits localement ; ne pas les remplacer par un fichier tiers.
Évaluer une politique, pas sélectionner sa meilleure réponse
Les politiques amorçage, SFT et GRPO génèrent chacune une réponse gloutonne sur les mêmes 200 problèmes test : c’est le pass@1 glouton. Elles produisent aussi quatre tirages stochastiques indépendants par problème avec graines fixées. Le taux de réussite par tirage utilise les 800 réponses ; il n’est pas le succès du meilleur des quatre. Les populations ID50, OOD100 et leur total restent distinctes.
Validité JSON, exactitude, absence d’EOS, longueurs et temps sont calculés sur toutes les sorties. La validation SFT mesure la NLL sans choisir un checkpoint ; les états finaux sont fixés avant les essais. Les réponses test n’entrent pas dans les mises à jour. Une baisse de loss ne prouve pas un gain d’exactitude, et un résultat défavorable à GRPO est publiable.
Exercice : deux tailles de groupe
Les groupes constants sont d’abord étudiés sur des récompenses construites, clairement séparées des réponses du modèle. Puis G=2 et G=8 sont réellement entraînés pendant 20 itérations depuis le même amorçage, avec les mêmes 80 présentations de prompts et 40 mises à jour. Ils produisent respectivement 160 et 640 rollouts : même nombre de mises à jour ne signifie pas même coût. Les deux politiques sont évaluées sur les mêmes 200 problèmes et quatre tirages.
Le G=4 principal a 100 itérations. Le comparer directement aux branches de 20 itérations ne serait pas une ablation isolant seulement la taille de groupe ; cette limite est indiquée. L’exercice protège les SHA-256 des poids, résultats et configuration principaux. Le vérificateur reste inchangé, même si tous les groupes sont faux.
Écrire le problème et son vérificateur indépendant
Récompense qui ne dépend pas du modèle
Erreur à diagnostiquer : Une sortie contient deux JSON dont un bon : exiger une seule réponse entière.
Invariant : 1 000/200/200 problèmes distincts ; solveur exact sur fixtures et cas limites.
Fichier 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,
}
Former un groupe et inspecter ses scores
Rollouts et probabilités de comportement
Erreur à diagnostiquer : Un dropout actif rend les log-probs incohérents : mettre la politique en mode évaluation tout en autorisant autograd aux updates.
Invariant : Chaque groupe a un prompt_id unique et quatre réponses avec scoring compatible.
Fichier 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,
}
Amorcer et comparer un SFT direct
Référence d'apprentissage sans RL
Erreur à diagnostiquer : Télécharger un checkpoint d'un ancien TP remplace le minimum : refaire l'amorçage dans ce projet.
Invariant : Aucun exemple test dans l'amorçage, coûts et hash initial capturés.
Fichier 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.",
},
)
Coder avantages, clipping et KL
Mise à jour en ligne par groupe
Erreur à diagnostiquer : Utiliser les log-probs actuels détachés comme old à chaque microbatch efface l'historique : capturer à la génération.
Invariant : Tests des signes du clipping, groupes constants et gradients réussis ; politique réellement mise à jour.
Fichier 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,
},
)
Évaluer qualité et coût total
Pass@1 et absence de signal
Erreur à diagnostiquer : Publier le meilleur tirage comme pass@1 : nommer le budget d'échantillonnage exact.
Invariant : 200 IDs par branche, sorties invalides conservées et coût génération non omis.
Fichier 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
Interpréter les groupes sans gagnant
Signal nul sans fabrication de récompense
Erreur à diagnostiquer : Ajouter un bonus arbitraire pour forcer une courbe descendante est interdit : consigner le résultat nul.
Invariant : Avantages nuls pour les groupes constants ; aucune modification du vérificateur.
Fichier 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,
},
)
Sources
DeepSeekMath v3 décrit l’objectif relatif de groupe. TRL v0.23.1 documente différentes réductions et variantes. Ces sources éclairent les choix ; aucun résultat publié ni trainer complet n’est reproduit. Base Qwen figée.
Observations de développement
Valeurs recalculées depuis les sorties brutes de ce parcours. Elles ne sont pas des seuils de gain ; le parcours depuis le ZIP les recalcule sur les mêmes populations.
{
"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
}
