自定义算法与 reward

10.1 三种 reward 来源

verl 把 reward 抽象成三种"风格"，通过 reward_model.style 选择：

verl 的 verl/workers/reward_manager/ 下针对每种风格都有对应实现。

10.2 rule reward：可验证奖励

R1 / DAPO 风格训练的关键。verl 自带几种内置 rule reward，通过 data_source 字段路由：

怎么自定义 rule reward

verl 的 reward 在 verl/utils/reward_score/ 下，每种 data_source 一个 Python 函数：

# verl/utils/reward_score/my_task.py
def compute_score(solution_str: str, ground_truth: str) -> float:
    """
    Return a float reward in [0, 1].
    """
    if my_validator(solution_str, ground_truth):
        return 1.0
    return 0.0

然后在 verl/utils/reward_score/__init__.py 里注册：

SCORE_FN_REGISTRY = {
    "openai/gsm8k":      gsm8k.compute_score,
    "lighteval/MATH":    math.compute_score,
    "codecontests":      codecontests.compute_score,
    "my-org/my-task":    my_task.compute_score,        # ★ 新加
}

数据集预处理时把 data_source = "my-org/my-task" 写进 parquet，训练时就会路由到你写的函数。

10.3 多 reward 加权组合

常见诉求：

$$ r = w_1 \cdot r_\text{correctness} + w_2 \cdot r_\text{format} + w_3 \cdot r_\text{length penalty} $$

在 compute_score 里直接组合：

def compute_score(solution_str, ground_truth):
    r_correct = 1.0 if check_answer(solution_str, ground_truth) else 0.0
    r_format  = 0.1 if has_proper_format(solution_str) else -0.1
    r_length  = -0.001 * max(0, len(solution_str) - 2048)
    return r_correct + r_format + r_length

10.4 model reward：神经 RM

用神经 RM 时，配置长这样：

reward_model.style=model
reward_model.model.path=/path/to/reward_model     # 训好的 RM 路径
reward_model.micro_batch_size_per_gpu=4
reward_model.use_remove_padding=true

verl 会起一组 RewardModelWorker（参见第 4 章），每 step 把 rollouts 送进去拿 scalar reward。

注意：

• RM 必须是带 value head 的 LM，不是普通 chat 模型；
• RM 的 tokenizer / template 必须和 actor 一致；
• 多机训练时 RM 也要分配 ResourcePool。

10.5 api reward：外部服务

把 reward 计算放在 verl 外部，verl 用 HTTP 调：

reward_model.style=api
reward_model.api.url=http://reward-server:8000/score
reward_model.api.timeout=30

这种模式特别适合：

• 用 GPT-4 / Claude 当 judge；
• reward 计算依赖外部资源（编译器、数据库）；
• 团队内已有的内部 LLM judge。

10.6 RewardManager：把所有 reward 串起来

verl/workers/reward_manager/naive.py · NaiveRewardManager 是 reward 处理的"枢纽"。它的工作：

1. 从每条 rollout 里读 data_source；
2. 按 data_source 路由到对应 compute_score；
3. 把 scalar reward 写回 DataProto；
4. 可选：处理 long sequence 的 token-level reward / process reward。

想换 reward 处理逻辑，新写一个 RewardManager 子类，比如同时跑规则 reward + 神经 RM 加权：

class HybridRewardManager(NaiveRewardManager):
    def __init__(self, rule_weight=0.7, rm_weight=0.3, **kw):
        super().__init__(**kw)
        self.rule_weight = rule_weight
        self.rm_weight = rm_weight

    def __call__(self, data):
        rule_scores = self._compute_rule(data)
        rm_scores   = self._compute_rm(data)
        data.batch["token_level_rewards"] = (
            self.rule_weight * rule_scores +
            self.rm_weight * rm_scores
        )
        return data

10.7 process reward / step reward

对于多步推理任务，可以给中间 step 单独评分（PRM / Process Reward Model）。verl 通过token-level reward 机制支持：

reward_model.reward_kwargs.process_reward=true
reward_model.reward_kwargs.step_score_path=/path/to/prm

在 verl/utils/reward_score/ 里实现 step 拆分 + 评分逻辑。

10.8 自定义算法：从 GRPO 改出新算法

verl 算法核心都在 verl/trainer/ppo/core_algos.py。新增算法的最小路径：

步骤 1：写自己的 advantage 估计

# verl/trainer/ppo/core_algos.py
def compute_my_advantage(token_level_rewards, response_mask, ...):
    """
    Args:
        token_level_rewards: [B, T] tensor
        response_mask: [B, T]
    Returns:
        advantages: [B, T]
        returns: [B, T]
    """
    # 你的算法
    return advantages, returns

步骤 2：在 dispatch 处加 case

# verl/trainer/ppo/core_algos.py
def compute_advantage(data, adv_estimator, gamma, lam, **kw):
    if adv_estimator == "gae":
        ...
    elif adv_estimator == "grpo":
        ...
    elif adv_estimator == "my_algo":             # ★ 新加
        return compute_my_advantage(...)

步骤 3：可选改 loss

如果只动 advantage，PPO loss 不用改。如果要改 loss，到 verl/workers/actor/dp_actor.py 改 compute_loss。

步骤 4：在 Hydra 里启用

algorithm.adv_estimator=my_algo

这就完了。新增一个算法可以只动一份代码 + 一行配置，这是 hybrid controller 设计的回报。

10.9 case 学习：DAPO 是怎么从 GRPO 改出来的

大致 4 个文件改动就能落地一个新算法。verl-recipe 仓库里的 dapo/ 就是这种实战例子。

10.10 把 reward 调试好的一组技巧

1. 先小批样本 dry run：data.train_batch_size=8、trainer.test_freq=1，看 reward 分布；
2. 打印 score 分布：每 step 看 reward 直方图，trainer.logger=[console] + 加 print；
3. 用 sft 模型当 baseline：rule reward 在 sft 后模型上应该有 baseline 分（GSM8K 大约 0.5+），如果 0.0 说明数据 / 模板有问题；
4. reward hacking 自检：训练后期 reward 飙升但生成质量降，多是 reward 设计漏洞，要补 format / length 惩罚；
5. kl 失控自检：kl 飙升说明策略走偏，调大 kl_loss_coef 或降 lr。

10.12 verl 的三套注册表：advantage / policy_loss / reward_manager

10.8 那段"步骤 1-4 加新算法"其实就是用 verl 的注册表机制。一份完整的"自定义"工作流要碰三个注册表，一一对照：

典型的"自定义新算法"工作：

# my_algos.py
from verl.trainer.ppo.core_algos import register_adv_est, register_policy_loss

@register_adv_est("my_grpo_variant")
def compute_my_advantage(token_level_rewards, response_mask, index, **kwargs):
    # 计算并返回 (advantages, returns)
    ...
    return advantages, returns

@register_policy_loss("my_loss")
def compute_my_loss(old_log_prob, log_prob, advantages, response_mask, ...):
    # 返回 (loss, metrics_dict)
    ...
    return loss, metrics

然后 YAML：

algorithm:
  adv_estimator: my_grpo_variant
actor_rollout_ref:
  actor:
    policy_loss:
      mode: my_loss

仅此而已——不需要改 ray_trainer.py、不需要碰 worker 代码。前提是把 my_algos.py 在训练入口前 import 一次（让装饰器执行）。

10.13 自定义 reward manager：完整脚手架

规则 reward 写到顶之后，下一步通常是写"组合 reward manager"——能在不同样本上用不同 reward 函数。verl/workers/reward_manager/abstract.py 的 AbstractRewardManager 定义了接口：

# my_reward_manager.py
from verl.workers.reward_manager.abstract import AbstractRewardManager
from verl.workers.reward_manager.registry import register

@register("hybrid_v2")
class HybridV2RewardManager(AbstractRewardManager):
    def __init__(self, tokenizer, **kwargs):
        self.tokenizer = tokenizer
        # 这里建你的 reward 函数列表 / 规则映射

    def __call__(self, data: DataProto, return_dict: bool = False):
        # 1. 遍历 data.batch / data.non_tensor_batch
        # 2. 给每条样本算一个标量 reward 或多维 reward
        # 3. 返回带 reward 字段的 DataProto，或 (reward_tensor, info_dict) 二元组
        ...

YAML 启用：

reward_model:
  manager: hybrid_v2          # 走你刚注册的名字

仓库自带的几个参考实现就在 workers/reward_manager/：

• naive.py——单一 reward 函数，按 data_source 字段 dispatch；
• batch.py——批量 reward（一次算一批样本，对模型类 reward 友好）；
• dapo.py——按 DAPO 论文做 group-level overshoot 惩罚；
• prime.py——PrIME（基于 prompt 重要性）reward 重加权。

10.14 BaseTool：写一个工具给 RL agent 调用

R1 类训练越来越多用到"工具增强 reasoning"——agent 在 CoT 里调用 calculator / code interpreter / web search。verl 的工具框架在 verl/tools/：

# verl/tools/base_tool.py:24
class BaseTool:
    @property
    def get_openai_tool_schema(self) -> dict:
        """返回 OpenAI 兼容的 tool schema，模型 prompt 里会看到"""

    async def create(self, instance_id=None, **kwargs):
        """工具实例化（多轮对话里需要持久化状态时用）"""

    async def execute(self, instance_id, parameters: dict) -> (ToolResponse, float, dict):
        """真正调用，返回 (输出文本, reward 分数, metrics)"""

    async def calc_reward(self, instance_id) -> float:
        """收尾时再要一次 reward 分数"""

    async def release(self, instance_id):
        """释放实例资源"""

注意所有方法都是 async——sandbox 调用、网络请求都不能阻塞 worker。execute() 返回的二元 reward (分数 + metrics) 会被聚合到 compute_reward 路径。

工具的 YAML 装配

verl/tools/tool_registry.py 提供两条加载路径：

典型 tools_config.yaml：

tools:
  - class_name: verl.tools.python_executor.PythonExecutorTool
    config:
      timeout_sec: 5
      sandbox_url: http://localhost:8080
  - class_name: my_tools.calculator.CalculatorTool
    config:
      precision: 10

启用：训练 YAML 的 actor_rollout_ref.rollout.multi_turn.tools_config_file=tools_config.yaml 这一段。Rollout 时 vLLM/SGLang 会把这些工具 schema 注入 system prompt，agent 生成 tool_call → 客户端解析 → 调你写的工具 → 把工具输出回灌为下一轮 user message。

10.15 Sandbox tool：让 RL 学会写代码

verl 自带几个 sandbox 集成（verl/tools/），是 R1 / DAPO 类训练的核心组件：

这些工具不只是"加 reward"——它们是 agent 在 inference 时真正调用的。一个 R1 风格训练里，模型在 CoT 中写 <tool_call name="python">...</tool_call>，verl 拿到这段就调 PythonExecutorTool.execute()，把 stdout 回灌成下一轮上下文。最终 reward 看模型最后给出的答案对不对。

10.16 整套自定义工作流的"四个 grep"

第一次给 verl 加东西时常迷失方向。下面这 4 个 grep 命令是最有效的"哪里改？"问题答案：

新加东西时先 grep 一遍看有没有重复——verl 的注册表里同名会覆盖，不会报错。

10.17 这章你需要带走什么

• 知道 reward 有 rule / model / api 三种，业务任务首选 rule；
• 会注册新 data_source 到 compute_score 映射；
• 会写 HybridRewardManager 组合多个 reward 源；
• 会用 core_algos.py + 4 个文件改动落地新算法；
• 知道 reward hacking / kl 失控这两类常见 bug 怎么自检；
• 记得 verl 的三套注册表（advantage / policy_loss / reward_manager）和 BaseTool 接口；
• 新功能开发都以"装饰器注册 → YAML 启用"为标准 PR 形态。