Como coletar dados de recompensa de processo para treinar agentes de codificação melhores

O que são modelos de recompensa de processo?

Duas formas de rotular recompensas de processo

Modelos de recompensa de resultado olham apenas para o fim da trajetória de um agente de codificação: o código compilou, os testes passaram, o problema foi resolvido? Modelos de recompensa de processo, por sua vez, pontuam cada etapa intermediária. Com um sinal em cada etapa, o treinamento consegue identificar onde o agente errou.

Trabalhos recentes avançaram nessa direção. AgentPRM redefine recompensas de processo para tarefas de agente, pontuando cada ação pelo progresso que ela faz em direção ao objetivo em vez de pela correção, e relata eficiência computacional mais de 8× melhor que a das baselines com que se compara. ToolRM constatou que modelos de recompensa treinados com saídas em linguagem natural julgam mal chamadas de ferramentas, e construiu modelos de recompensa específicos para ferramentas, além do FC-RewardBench para avaliá-los. Em contraste, DeepSWE treina um agente de codificação apenas com uma recompensa de resultado esparsa, se os testes passam, e chega a 42,2% de Pass@1 e 59% com escalonamento em tempo de teste no SWE-bench Verified. Essa é a configuração baseada apenas em resultado que a supervisão de processo tenta melhorar.

O que todos eles precisam é de anotação humana no nível da etapa, e esse costuma ser o gargalo. Para a referência do esquema, consulte a documentação de recompensa de processo.

Dois modos de anotação

Modo primeiro-erro

O anotador lê a trajetória de cima para baixo e clica na primeira etapa em que o agente erra. Todas as etapas anteriores são registradas como corretas, e essa etapa e todas as seguintes como incorretas. É rápido porque há uma única decisão por trace, e se encaixa no caso comum em que um agente que sai do rumo não se recupera.

yaml

annotation_schemes:
  - annotation_type: process_reward
    name: step_rewards
    description: "Click the first step where the agent makes a mistake. Steps after it are flagged automatically."
    mode: first_error
    steps_key: structured_turns
    step_text_key: content

Modo por etapa

Cada etapa recebe o próprio rótulo, o que captura o que o modo primeiro-erro não consegue: recuperação parcial, um desvio inofensivo ou uma etapa que está certa isoladamente e errada no contexto. allow_neutral adiciona a terceira opção no estilo do PRM800K.

yaml

annotation_schemes:
  - annotation_type: process_reward
    name: step_rewards
    description: "Rate each step correct, neutral, or incorrect"
    mode: per_step
    steps_key: structured_turns
    step_text_key: content
    allow_neutral: true
    inline_with_trace: true

inline_with_trace prende os controles a cada etapa na exibição, o que faz diferença em um trace longo, em que um painel separado faz o anotador perder o ponto onde estava.

Configurando o projeto

Etapa 1: prepare os traces

Cada linha é uma trajetória. steps_key indica a lista de etapas e step_text_key indica o campo de cada etapa a ser exibido:

json

{
  "id": "trace_001",
  "issue_description": "Fix the TypeError in parse_config() when the config file is empty",
  "structured_turns": [
    {"step_idx": 0, "type": "thinking", "content": "I need to find where the TypeError occurs. Let me read the file."},
    {"step_idx": 1, "type": "file_read", "content": "def parse_config(path):\n    with open(path) as f:\n        data = json.load(f)\n    return data['settings']"},
    {"step_idx": 2, "type": "thinking", "content": "When the file is empty json.load returns None, so None['settings'] raises. Add a check."},
    {"step_idx": 3, "type": "file_edit", "content": "+    if data is None:\n+        return {}"},
    {"step_idx": 4, "type": "bash_command", "content": "python -m pytest tests/test_config.py -v  ->  2 passed"}
  ]
}

Converter a partir de outro formato de agente é um passo de CLI:

bash

python -m potato.trace_converter \
  --input claude_code_sessions.json \
  --input-format claude_code \
  --output data/traces.jsonl

--list-formats mostra os formatos suportados, entre eles claude_code, aider, swe_agent_trajectory, swebench, react, langchain e mais nove. Use --auto-detect quando não tiver certeza de qual formato você tem.

Etapa 2: escreva a configuração

yaml

annotation_task_name: "PRM Data Collection"
task_dir: "."
 
data_files:
  - "data/traces.jsonl"
 
item_properties:
  id_key: id
  text_key: issue_description
 
instance_display:
  fields:
    - key: issue_description
      type: text
      label: "Issue"
 
    - key: structured_turns
      type: coding_trace
      label: "Agent trajectory"
      display_options:
        diff_view: unified
        show_file_tree: true
        show_reasoning: true
        terminal_theme: dark
        collapse_long_outputs: true
        max_output_lines: 50
 
annotation_schemes:
  - annotation_type: process_reward
    name: step_rewards
    description: "Click the first step where the agent makes an error"
    mode: first_error
    steps_key: structured_turns
    step_text_key: content
 
  - annotation_type: radio
    name: outcome
    description: "Did the agent resolve the issue?"
    labels:
      - name: resolved
        text: "Fully resolved"
      - name: partial
        text: "Partially resolved"
      - name: not_resolved
        text: "Not resolved"
 
  - annotation_type: text
    name: error_description
    description: "If the trajectory went wrong, describe the error briefly"
    rows: 3
    placeholder: "e.g. edited the wrong file"
    label_requirement:
      required: false
 
num_annotators_per_item: 2
 
attention_checks:
  enabled: true
  items_file: "data/attention_checks.json"
  frequency: 20
  min_response_time: 20.0
  failure_handling:
    warn_threshold: 2
    block_threshold: 5
 
user_config:
  allow_all_users: false
  users:
    - reviewer1@example.com
    - reviewer2@example.com
    - reviewer3@example.com
 
output_annotation_dir: "annotation_output/"
export_annotation_format: jsonl

Um objeto de rótulo é identificado por name. O campo text é o que o anotador lê, então o valor armazenado continua sendo um identificador estável enquanto o texto pode mudar.

min_response_time: 20.0 é o controle útil aqui. Ninguém lê uma trajetória de várias etapas com diffs em vinte segundos, então ele pega a pressa que um limite de tempo por trace não pegaria.

Etapa 3: inicie o servidor

bash

potato start config.yaml -p 8000

Exportando para treinamento

O exportador coding_eval grava diretamente os formatos de treinamento, então a preparação já é feita para você:

bash

python -m potato.export -c config.yaml -f coding_eval -o ./export/

Ele grava prm_training_data.jsonl com os rótulos por etapa, preference_pairs.jsonl com traces escolhidos e rejeitados no estilo DPO montados a partir das anotações, swebench_results.jsonl e code_reviews.jsonl. Os quatro são gravados por padrão. Para restringir, passe a opção types do exportador: --option types=prm,preference. Quando um tipo solicitado não encontra dados no estudo, ele avisa em vez de gravar um arquivo vazio.

A anotação é armazenada sob o nome do esquema como uma lista steps mais o modo. reward é 1 para correto, -1 para incorreto, 0 para neutro e null para uma etapa que ninguém julgou:

json

{
  "step_rewards": {
    "mode": "first_error",
    "steps": [
      {"reward": 1, "source": "human", "verified": true},
      {"reward": 1, "source": "human", "verified": true},
      {"reward": -1, "source": "human", "verified": true},
      {"reward": -1, "source": "human", "verified": true}
    ]
  }
}

Recorra a um script apenas quando precisar de um formato que o exportador não grava. O motivo usual é tratar você mesmo as etapas não julgadas, já que uma etapa que ninguém julgou é null, e não 0:

python

import json
from pathlib import Path
 
SCHEME = "step_rewards"
 
examples = []
for f in Path("export/").rglob("*.jsonl"):
    with open(f) as fh:
        for line in fh:
            rec = json.loads(line)
            payload = (rec.get("labels", rec) or {}).get(SCHEME)
            if not payload:
                continue
 
            steps = payload.get("steps", [])
            # Drop traces with unjudged steps rather than guessing at them.
            if any(s.get("reward") is None for s in steps):
                continue
 
            item = items_by_id[rec["instance_id"]]      # your own data file
            examples.append({
                "trace_id": rec["instance_id"],
                "prompt": item["issue_description"],
                "steps": [
                    {"content": src["content"], "label": s["reward"]}
                    for src, s in zip(item["structured_turns"], steps)
                ],
            })
 
with open("prm_labels.jsonl", "w") as out:
    for ex in examples:
        out.write(json.dumps(ex) + "\n")
 
print(f"Wrote {len(examples)} traces")

Os pares de preferência vêm das mesmas anotações. Quando dois traces tentaram resolver o mesmo problema, o exportador os pareia e trata como trace escolhido aquele cujo primeiro erro aparece mais tarde, ou que não tem nenhum.

Análise

Onde cai o primeiro erro

python

import json
from pathlib import Path
 
SCHEME = "step_rewards"
 
traces = []
for f in Path("export/").rglob("*.jsonl"):
    with open(f) as fh:
        for line in fh:
            payload = (json.loads(line).get("labels", {}) or {}).get(SCHEME)
            if payload:
                traces.append(payload)
 
positions = []
all_correct = 0
for t in traces:
    steps = t.get("steps", [])
    first = next((i for i, s in enumerate(steps) if s.get("reward") == -1), None)
    if first is None:
        all_correct += 1
    elif len(steps) > 1:
        positions.append(first / (len(steps) - 1))
 
print(f"Traces annotated: {len(traces)}")
print(f"No errors: {all_correct} ({all_correct / len(traces):.1%})")
if positions:
    print(f"Average first-error position: {sum(positions) / len(positions):.2f} (0=start, 1=end)")

Concordância sobre o primeiro erro

Dois anotadores raramente escolhem a mesma etapa, e a concordância exata subestima o quanto eles concordam. Agrupar a posição em faixas antes é a comparação mais informativa:

python

 
def bin_position(steps):
    first = next((i for i, s in enumerate(steps) if s.get("reward") == -1), None)
    if first is None:
        return "all_correct"
    return "early_error" if first < len(steps) / 2 else "late_error"

Compare essas faixas entre os anotadores que compartilham itens, usando a sobreposição que num_annotators_per_item oferece.

Dicas

Use o modo primeiro-erro para ganhar velocidade. Para treinar um PRM que guie a busca, ele dá sinal suficiente por uma fração do custo de anotação, e a maioria dos agentes falha em cascata de qualquer forma.

Use o modo por etapa quando precisar do detalhe. Recuperação parcial e desvios inofensivos só existem como rótulos no modo por etapa.

Comece com anotadores experientes. O trabalho envolve ler código, diffs e saída de terminal. Comece com um grupo pequeno, meça a concordância, calibre com exemplos compartilhados e depois amplie.

Defina um tempo mínimo de resposta. As trajetórias ficam longas, e um piso impede que os anotadores passem correndo pelos diffs.

Faça uma rodada de calibração. Peça que todos rotulem os mesmos dez a vinte traces e discutam as divergências antes de começar a anotação de produção.