Guides6 min read
如何收集过程奖励数据，训练更好的编码智能体
用 Potato 收集逐步奖励信号、训练 PRM 的分步指南，涵盖首错模式、逐步标注，以及导出到训练流水线。
Potato Team
什么是过程奖励模型
标注过程奖励的两种方式
结果奖励模型（ORM）只看编码智能体轨迹的结尾：代码编译过了吗，测试跑通了吗，问题解决了吗。过程奖励模型（PRM）则给每一个中间步骤打分。每一步都有奖励信号，训练方法就能定位智能体是从哪里开始走错的，这往往让学习更省样本，也有助于泛化。
近期的工作正朝这个方向推进。AgentPRM 为智能体任务重新定义了过程奖励：每个动作按它把任务推进到离目标多近来打分，而不是按对错，论文报告的算力效率比其对比的基线高出 8 倍以上。ToolRM 发现，在自然语言输出上训练的奖励模型判断工具调用的能力很差，于是构建了面向工具的奖励模型，以及用来评测它们的 FC-RewardBench。作为对照，DeepSWE 只用一个稀疏的结果奖励（测试是否通过）来训练编码智能体，在 SWE-bench Verified 上达到 Pass@1 42.2%，配合测试时扩展达到 59%。过程监督想要改进的，正是这种只看结果的设置。
这些方法共同需要的是高质量的步骤级人工标注，而这通常就是瓶颈。Potato 的过程奖励方案就是为了让这类数据收集得更快。底层方案见轨迹评估文档，trace 输入格式的细节见智能体 trace 文档。
两种标注模式
Potato 提供两种 PRM 标注模式，在速度和粒度之间做取舍。按你的数据预算和目标挑一种。
首错模式
在首错模式下，标注者从上到下读完轨迹，点击智能体第一次出错的那一步。Potato 随后把这一步之前的所有步骤标为正确，把这一步及之后的所有步骤标为错误。
这种方式快，因为标注者只需要找一个决策点。当错误会连锁传播时它很好用，也就是说智能体一旦走偏就很少能自己回来，而这在实际中是常态。
yaml
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.首错标注的流程是这样的：
- 标注者打开一条 trace，看到所有步骤由 CodingTraceDisplay 组件渲染出来。
- 他们按顺序读完这些步骤，查看 diff、终端输出和推理过程。
- 找到第一个错误步骤时，点击它旁边的错误标记。
- 第 0 步到第 N-1 步变绿（正确），第 N 步到结尾变红（错误）。
- 标注者检查自动打上的标签，点击 “Submit” 确认。
如果整条 trace 都正确（智能体完美解决了任务），标注者点 “All Correct”。如果第一步就已经错了，就点第 0 步或者用 “All Incorrect”。
下面是 PRM 标注界面的实际样子：
在首错模式下，点击第一个错误步骤，后续步骤会自动标记
逐步模式
在逐步模式下，每一步都有自己的标签。这样得到的数据更丰富，因为它能记录智能体从错误中部分恢复、绕了一段无害但没必要的路，或者某一步本身没问题但放在上下文里是错的这类情况。
yaml
annotation_schemes:
- annotation_type: process_reward
name: prm_per_step
mode: "per_step"搭建一个 PRM 标注项目
第 1 步：准备 trace 数据
输入数据应该是一个 JSONL 文件，每一行是一个包含智能体轨迹的 JSON 对象。关键字段是 id、issue_description 和 structured_turns：
json
{
"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
}
]
}如果你要从已有的智能体格式转换过来，用 trace 转换工具：
bash
# 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 渲染编码智能体 trace 时会正确高亮 diff：
代码 diff、终端输出和文件读取都带语法高亮渲染
第 2 步：写配置文件
下面是一份用首错模式做 PRM 标注的完整项目配置：
yaml
# 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"第 3 步：启动标注服务
bash
# 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 &打开 http://localhost:8000，用配置里的任一标注者账号登录，就可以开始审阅 trace 了。
第 4 步：跟踪进度
标注进行期间，随时查看进度和一致性：
bash
# Check annotation progress
potato status config.yaml
# View inter-annotator agreement
potato agreement config.yaml --metric krippendorff_alpha导出到训练格式
标注完成后，按训练流水线需要的格式导出数据。
用于奖励模型训练的 PRM 格式
PRM 导出格式为每条 trace 生成一个带步骤级标签的 JSON 对象：
bash
potato export \
--format prm \
--project ./output/ \
--output ./training_data/prm_labels.jsonl输出长这样：
json
{
"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}
]
}DPO/RLHF 偏好对
当同一个问题有多条 trace 时（比如来自不同智能体或不同次运行），Potato 可以基于 PRM 标签生成偏好对：
bash
potato export \
--format preference_pairs \
--project ./output/ \
--output ./training_data/preferences.jsonl \
--pair_by "issue_id"偏好对导出会比较尝试同一任务的多条 trace，按步骤级标签选出更好的那一条：
json
{
"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
}SWE-bench 兼容结果
导出成 SWE-bench 格式用于基准测试：
bash
potato export \
--format swe_bench \
--project ./output/ \
--output ./training_data/swe_bench_results.json分析示例
收集完标注之后，可以用下面这些 Python 片段分析数据、找出规律。
按步骤类型统计步骤级准确率
python
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']})")找出常见的失败点
python
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}")计算 PRM 标签的标注者间一致性
python
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}")高效收集 PRM 数据的几点建议
要快就用首错模式。如果你训练 PRM 是为了引导搜索（MCTS、best-of-N 采样），首错模式给的信号已经够用，而标注速度是逐步模式的 2 到 3 倍。反正大多数智能体的失败都是连锁式的：一步错，后面一串都跟着错。
需要细节时用逐步模式。如果你在意部分恢复、无害的绕路，或者你要训练一个标签多于两类的步骤级奖励模型，逐步模式多花的时间是值得的。
把 PRM 和成对比较结合。先用 PRM 单独标注每条 trace，再对尝试同一个问题的 trace 做一次成对比较。一轮标注同时拿到步骤级奖励和偏好对。
先找有经验的标注者。PRM 标注要读代码、diff 和终端输出。先从一小组有经验的开发者开始，量一下一致性，用样例校准过，再扩大规模。
设一个每条数据的最短时间。trace 会变得很复杂。30 秒的下限能防止标注者不看改动就一路点下去。这个值按你的平均 trace 长度调。
准备校准样例。正式标注之前，让所有人标同样的 10 到 20 条 trace，然后一起过一遍分歧在哪。这对一致性影响很大。