1
# 结果监督（Outcome Supervision / ORM）
2
class OutcomeSupervision:
3
    """
4
    只看最终答案是否正确
5
    不关心推理过程
6
    """
7
    def reward(self, solution, final_answer, ground_truth):
8
        return 1.0 if final_answer == ground_truth else 0.0
9

10
# 过程监督（Process Supervision / PRM）
11
class ProcessSupervision:
12
    """
13
    评估每个推理步骤是否正确
14
    提供细粒度反馈
15
    """
16
    def reward_per_step(self, steps, ground_truth):
17
        rewards = []
18
        for i, step in enumerate(steps):
19
            # 判断当前步骤是否正确
20
            is_correct = verify_step(step, steps[:i], ground_truth)
21
            rewards.append(1.0 if is_correct else 0.0)
22
        return rewards

1
# OpenAI 的对比实验
2

3
results = {
4
    "ORM": {
5
        "solve_rate": "70%",  # 能解决多少问题
6
        "step_accuracy": "85%"  # 步骤准确率
7
    },
8
    "PRM": {
9
        "solve_rate": "78%",  # 显著提升
10
        "step_accuracy": "95%"  # 步骤更可靠
11
    }
12
}

1
class PRM800KDataset:
2
    """
3
    PRM800K：80 万步骤级人类反馈标注
4
    用于训练过程监督奖励模型
5
    """
6

7
    def __init__(self):
8
        self.annotations = []
9

10
    def add_annotation(self, problem, solution, step_feedback):
11
        """
12
        添加一个标注样本
13
        problem: 数学问题
14
        solution: 逐步解答
15
        step_feedback: 每一步是否正确的标注
16
        """
17
        self.annotations.append({
18
            "problem": problem,
19
            "solution": solution,
20
            "step_feedback": step_feedback
21
        })

1
class ActiveLearning:
2
    """主动学习减少标注成本"""
3

4
    def select_samples(self, model, unlabeled_problems):
5
        """
6
        选择模型最不确定的样本进行标注
7
        """
8
        uncertainties = []
9

10
        for problem in unlabeled_problems:
11
            # 让模型生成解答
12
            solutions = model.generate_solutions(problem, n=10)
13

14
            # 计算答案的一致性
15
            answers = [sol.final_answer for sol in solutions]
16
            agreement = max(set(answers), key=answers.count) / len(answers)
17

18
            # 一致性低 = 不确定 = 值得标注
19
            uncertainties.append((problem, 1 - agreement))
20

21
        # 选择最不确定的样本
22
        uncertainties.sort(key=lambda x: x[1], reverse=True)
23
        return [p for p, _ in uncertainties[:self.batch_size]]

1
class ProcessRewardModel:
2
    """过程监督奖励模型"""
3

4
    def __init__(self, base_model):
5
        self.model = base_model
6
        self.reward_head = nn.Linear(hidden_size, 1)
7

8
    def forward(self, problem, solution_steps):
9
        """
10
        输出每一步的奖励分数
11
        """
12
        rewards = []
13
        hidden = self.model.get_input_embedding(problem)
14

15
        for step in solution_steps:
16
            step_embedding = self.model.encode_step(step, hidden)
17
            reward = self.reward_head(step_embedding)
18
            rewards.append(reward)
19
            hidden = self.model.update_hidden(hidden, step_embedding)
20

21
        return rewards
22

23
    def compute_loss(self, problems, solutions, step_labels):
24
        """
25
        计算过程监督损失
26
        step_labels: 每一步的人类标注（正确/错误）
27
        """
28
        rewards = self.forward(problems, solutions)
29

30
        # 每一步的交叉熵损失
31
        loss = 0
32
        for reward, label in zip(rewards, step_labels):
33
            loss += F.binary_cross_entropy_with_logits(
34
                reward,
35
                torch.tensor(label)
36
            )
37
        return loss

1
# 过程监督对不同规模模型的效果
2

3
model_sizes = ["1B", "7B", "13B", "175B"]
4

5
results = {
6
    "ORM": {
7
        "1B": 45,
8
        "7B": 62,
9
        "13B": 71,
10
        "175B": 78
11
    },
12
    "PRM": {
13
        "1B": 52,
14
        "7B": 73,
15
        "13B": 82,
16
        "175B": 85
17
    }
18
}
19

20
# PRM 在所有规模上都优于 ORM