Come raccogliere dati di process reward per addestrare coding agent migliori
Guida passo passo alla raccolta di segnali di reward per singolo passo per l'addestramento di PRM con Potato. Copre la modalità primo errore, l'annotazione passo per passo e l'esportazione verso le pipeline di addestramento.
Che cosa sono i process reward model?
Due modi di etichettare i process reward
Gli outcome reward model (ORM) guardano solo la fine della traiettoria di un coding agent: il codice compila, i test passano, il problema è risolto? I process reward model (PRM) assegnano invece un punteggio a ogni passo intermedio. Con un segnale di reward a ogni passo, i metodi di addestramento possono individuare con precisione dove l'agente ha sbagliato, il che tende a rendere l'apprendimento più efficiente in termini di campioni e aiuta la generalizzazione.
Il lavoro recente spinge in questa direzione. AgentPRM ridefinisce le ricompense di processo per i compiti d'agente: ogni azione viene valutata per quanto avvicina all'obiettivo, non per la sua correttezza, e riporta un'efficienza di calcolo oltre 8× migliore rispetto alle baseline con cui si confronta. ToolRM ha rilevato che i modelli di ricompensa addestrati su output in linguaggio naturale giudicano male le chiamate agli strumenti, e ha costruito modelli di ricompensa specifici per gli strumenti insieme a FC-RewardBench per valutarli. Per contrasto, DeepSWE addestra un agente di codice con la sola ricompensa di esito sparsa, cioè se i test passano, e raggiunge il 42,2% di Pass@1 e il 59% con scaling al momento del test su SWE-bench Verified. È proprio l'impostazione basata solo sull'esito che la supervisione di processo cerca di migliorare.
Quello che serve a tutti questi approcci è una buona annotazione umana a livello di passo, ed è di solito lì che si crea il collo di bottiglia. Gli schemi di process reward di Potato sono fatti per rendere più rapida la raccolta di quei dati. Per lo schema sottostante, vedi la documentazione sulla valutazione delle traiettorie, e per i dettagli sul formato di input delle tracce la documentazione sulle tracce degli agenti.
Due modalità di annotazione
Potato ha due modalità di annotazione PRM che scambiano velocità con granularità. Scegli quella che sta nel tuo budget di dati e nei tuoi obiettivi.
Modalità primo errore
Nella modalità primo errore l'annotatore legge la traiettoria dall'alto verso il basso e clicca sul primo passo in cui l'agente sbaglia. Potato segna poi come corretti tutti i passi precedenti e come sbagliati tutti quelli dal passo cliccato in avanti.
È veloce perché l'annotatore deve trovare un solo punto di decisione. Funziona bene quando gli errori si propagano a cascata, cioè quando un agente, una volta uscito di strada, difficilmente si riprende, che è il caso più comune nella pratica.
annotation_schemes:
- annotation_type: process_reward
name: prm_first_error
mode: "first_error"
description: >
Review the agent's steps from top to bottom. Click on the
first step where the agent makes a mistake. All steps before
your selection will be marked correct; all steps after
(including the selected step) will be marked incorrect.Il flusso di annotazione in modalità primo errore è questo:
- L'annotatore apre una traccia e vede tutti i passi renderizzati dal componente CodingTraceDisplay.
- Legge i passi in sequenza, esaminando diff, output del terminale e ragionamento.
- Quando trova il primo passo sbagliato, clicca sul marcatore di errore accanto a esso.
- I passi da 0 a N-1 diventano verdi (corretti), i passi da N alla fine diventano rossi (sbagliati).
- L'annotatore rivede l'etichettatura automatica e clicca su «Submit» per confermare.
Se l'intera traccia è corretta (l'agente ha risolto il task alla perfezione), l'annotatore clicca su «All Correct». Se già il primo passo è sbagliato, clicca sul passo 0 oppure usa «All Incorrect».
Ecco l'interfaccia di annotazione PRM in funzione:
Nella modalità primo errore basta cliccare il primo passo sbagliato e tutti i passi successivi vengono segnati in automatico
Modalità passo per passo
Nella modalità passo per passo ogni passo riceve la propria etichetta. Ne escono dati più ricchi, perché così si colgono i casi in cui un agente si riprende in parte da un errore, prende una deviazione inutile ma innocua, o compie un passo di per sé corretto ma sbagliato nel contesto.
annotation_schemes:
- annotation_type: process_reward
name: prm_per_step
mode: "per_step"Impostare un progetto di annotazione PRM
Passo 1: prepara i dati delle tracce
I dati di input devono essere un file JSONL in cui ogni riga è un oggetto JSON che contiene la traiettoria dell'agente. I campi principali sono id, issue_description e structured_turns:
{
"id": "trace_001",
"issue_description": "Fix the TypeError in parse_config() when config file is empty",
"repo": "myorg/myproject",
"base_commit": "a1b2c3d",
"structured_turns": [
{
"step_idx": 0,
"type": "thinking",
"content": "I need to understand what parse_config does and where the TypeError occurs. Let me read the file first."
},
{
"step_idx": 1,
"type": "file_read",
"path": "src/config.py",
"content": "def parse_config(path):\n with open(path) as f:\n data = json.load(f)\n return data['settings']",
"start_line": 1,
"end_line": 4
},
{
"step_idx": 2,
"type": "thinking",
"content": "The bug is on line 4. When the file is empty, json.load returns None, and None['settings'] raises TypeError. I should add a check."
},
{
"step_idx": 3,
"type": "file_edit",
"path": "src/config.py",
"diff": "--- a/src/config.py\n+++ b/src/config.py\n@@ -1,4 +1,6 @@\n def parse_config(path):\n with open(path) as f:\n data = json.load(f)\n+ if data is None:\n+ return {}\n return data['settings']"
},
{
"step_idx": 4,
"type": "bash_command",
"command": "python -m pytest tests/test_config.py -v",
"output": "tests/test_config.py::test_parse_config_empty PASSED\ntests/test_config.py::test_parse_config_valid PASSED\n\n2 passed in 0.12s",
"exit_code": 0
}
]
}Se stai convertendo da un formato di agente già esistente, usa lo strumento di conversione delle tracce:
# Convert Claude Code traces
potato convert-traces \
--format claude_code \
--input ./raw_traces/ \
--output ./data/traces.jsonl
# Convert SWE-Agent trajectories
potato convert-traces \
--format swe_agent \
--input ./swe_agent_output/ \
--output ./data/traces.jsonlPotato renderizza le tracce dei coding agent con l'evidenziazione corretta dei diff:
Diff del codice, output del terminale e letture di file vengono renderizzati con evidenziazione della sintassi
Passo 2: crea la configurazione
Ecco una configurazione di progetto completa per l'annotazione PRM in modalità primo errore:
# config.yaml
project_name: "PRM Data Collection - SWE-bench Traces"
port: 8000
data:
source: "local"
input_path: "./data/traces.jsonl"
data_format: "coding_trace"
coding_agent:
display:
diff_style: "unified"
context_lines: 3
syntax_highlighting: true
terminal_theme: "dark"
file_tree:
enabled: true
position: "left"
collapsible:
auto_collapse_thinking: true
auto_collapse_long_output: true
long_output_threshold: 50
annotation_schemes:
- annotation_type: process_reward
name: step_reward
mode: "first_error"
description: >
Review the agent's trajectory step by step. Click the first
step where the agent makes an error. If the entire trajectory
is correct, click "All Correct."
- annotation_type: radio
name: outcome
labels:
- value: "resolved"
text: "Fully Resolved"
- value: "partial"
text: "Partially Resolved"
- value: "not_resolved"
text: "Not Resolved"
- annotation_type: text
name: error_description
description: "If incorrect, briefly describe the error"
placeholder: "e.g., Agent edited the wrong file..."
output:
path: "./output/"
format: "jsonl"
quality_control:
inter_annotator_agreement: true
overlap_percentage: 15
minimum_time_per_instance: 20
annotators:
- username: "reviewer1"
- username: "reviewer2"
- username: "reviewer3"Passo 3: avvia il server di annotazione
# Start the annotation server
potato start config.yaml -p 8000
# Or run in the background
nohup potato start config.yaml -p 8000 > potato.log 2>&1 &Vai su http://localhost:8000, accedi con uno degli account annotatore configurati e comincia a rivedere le tracce.
Passo 4: tieni d'occhio l'avanzamento
Mentre l'annotazione procede, controlla avanzamento e accordo:
# Check annotation progress
potato status config.yaml
# View inter-annotator agreement
potato agreement config.yaml --metric krippendorff_alphaEsportare verso i formati di addestramento
Finita l'annotazione, esporta i dati nel formato che si aspetta la tua pipeline di addestramento.
Formato PRM per l'addestramento del reward model
L'esportazione in formato PRM produce un oggetto JSON per traccia con le etichette a livello di passo:
potato export \
--format prm \
--project ./output/ \
--output ./training_data/prm_labels.jsonlL'output è così:
{
"trace_id": "trace_001",
"issue_description": "Fix the TypeError in parse_config() when config file is empty",
"total_steps": 5,
"first_error_step": null,
"all_correct": true,
"steps": [
{"step_idx": 0, "type": "thinking", "label": "correct", "reward": 1.0},
{"step_idx": 1, "type": "file_read", "label": "correct", "reward": 1.0},
{"step_idx": 2, "type": "thinking", "label": "correct", "reward": 1.0},
{"step_idx": 3, "type": "file_edit", "label": "correct", "reward": 1.0},
{"step_idx": 4, "type": "bash_command", "label": "correct", "reward": 1.0}
]
}Coppie di preferenza DPO/RLHF
Quando hai più tracce per lo stesso problema (per esempio da agenti diversi o da esecuzioni diverse), Potato può generare coppie di preferenza a partire dalle etichette PRM:
potato export \
--format preference_pairs \
--project ./output/ \
--output ./training_data/preferences.jsonl \
--pair_by "issue_id"L'esportazione delle coppie di preferenza confronta le tracce che hanno affrontato lo stesso task e sceglie la migliore in base alle etichette a livello di passo:
{
"prompt": "Fix the TypeError in parse_config() when config file is empty",
"chosen_trace_id": "trace_001",
"rejected_trace_id": "trace_002",
"chosen_first_error": null,
"rejected_first_error": 3,
"chosen_steps": 5,
"rejected_steps": 7,
"margin": 0.8
}Risultati compatibili con SWE-bench
Esporta in formato SWE-bench per il benchmarking:
potato export \
--format swe_bench \
--project ./output/ \
--output ./training_data/swe_bench_results.jsonEsempi di analisi
Una volta raccolte le annotazioni, usa questi frammenti Python per analizzare i dati e individuare gli schemi ricorrenti.
Accuratezza a livello di passo per tipo di passo
import json
from collections import defaultdict
# Load PRM annotations
with open("training_data/prm_labels.jsonl") as f:
traces = [json.loads(line) for line in f]
# Compute accuracy by step type
type_stats = defaultdict(lambda: {"correct": 0, "total": 0})
for trace in traces:
for step in trace["steps"]:
step_type = step["type"]
type_stats[step_type]["total"] += 1
if step["label"] == "correct":
type_stats[step_type]["correct"] += 1
print("Step-Level Accuracy by Type:")
print("-" * 45)
for step_type, stats in sorted(type_stats.items()):
acc = stats["correct"] / stats["total"] * 100
print(f" {step_type:<20} {acc:5.1f}% ({stats['correct']}/{stats['total']})")Trovare i punti di fallimento più frequenti
import json
from collections import Counter
with open("training_data/prm_labels.jsonl") as f:
traces = [json.loads(line) for line in f]
# Analyze where errors first occur
error_positions = []
error_types_at_first_error = Counter()
for trace in traces:
if trace["first_error_step"] is not None:
pos = trace["first_error_step"]
total = trace["total_steps"]
# Normalize position to 0-1 range
error_positions.append(pos / total)
# Track what type of step caused the first error
error_step = trace["steps"][pos]
error_types_at_first_error[error_step["type"]] += 1
if error_positions:
avg_pos = sum(error_positions) / len(error_positions)
print(f"Average first-error position: {avg_pos:.2f} (0=start, 1=end)")
print(f"Traces with errors: {len(error_positions)}/{len(traces)}")
print()
print("Most common step types at first error:")
for step_type, count in error_types_at_first_error.most_common(5):
print(f" {step_type}: {count}")Calcolare l'accordo tra annotatori sulle etichette PRM
import json
import numpy as np
from sklearn.metrics import cohen_kappa_score
def load_annotations(annotator_file):
"""Load annotations from a single annotator's output file."""
with open(annotator_file) as f:
data = {item["trace_id"]: item for item in
(json.loads(line) for line in f)}
return data
ann1 = load_annotations("output/reviewer1/annotations.jsonl")
ann2 = load_annotations("output/reviewer2/annotations.jsonl")
# Find overlapping traces
overlap_ids = set(ann1.keys()) & set(ann2.keys())
print(f"Overlapping traces: {len(overlap_ids)}")
# Compare first-error step labels
labels1 = []
labels2 = []
for trace_id in overlap_ids:
fe1 = ann1[trace_id].get("first_error_step", -1)
fe2 = ann2[trace_id].get("first_error_step", -1)
# Bin into: all_correct, early_error (first half), late_error (second half)
total = ann1[trace_id]["total_steps"]
for fe, labels in [(fe1, labels1), (fe2, labels2)]:
if fe is None or fe == -1:
labels.append("all_correct")
elif fe < total / 2:
labels.append("early_error")
else:
labels.append("late_error")
kappa = cohen_kappa_score(labels1, labels2)
print(f"Cohen's kappa (binned first-error): {kappa:.3f}")Consigli per raccogliere dati PRM in modo efficiente
Usa la modalità primo errore per andare veloce. Se stai addestrando un PRM per guidare una ricerca (MCTS, campionamento best-of-N), la modalità primo errore ti dà segnale a sufficienza a una velocità di annotazione 2-3 volte superiore a quella della modalità passo per passo. Tanto la maggior parte degli agenti fallisce a cascata: un errore porta a una catena di passi sbagliati.
Usa la modalità passo per passo quando ti serve il dettaglio. Se ti interessano i recuperi parziali, le deviazioni innocue, o se stai costruendo un reward model a livello di passo con più di due etichette, la modalità passo per passo ripaga il tempo in più che costa.
Combina il PRM con il confronto a coppie. Etichetta le tracce una per una con il PRM, poi fai un confronto a coppie sulle tracce che hanno affrontato lo stesso problema. Un solo giro di annotazione ti dà sia i reward a livello di passo sia le coppie di preferenza.
Parti da annotatori esperti. Annotare per il PRM vuol dire leggere codice, diff e output del terminale. Comincia con un gruppo ristretto di sviluppatori esperti, misura l'accordo, tara il lavoro su qualche esempio e solo dopo allarga.
Imposta un tempo minimo per istanza. Le tracce si fanno complicate. Un limite di 30 secondi evita che gli annotatori corrano senza leggere davvero le modifiche. Regolalo sulla lunghezza media delle tue tracce.
Prepara esempi di taratura. Prima di partire con l'annotazione vera, fai etichettare a tutti le stesse 10-20 tracce e discutete dove non eravate d'accordo. Sulla coerenza cambia parecchio.