执行摘要

• 一句话：默认使用标准 pre-KL 奖励（rm_scores）进行 DAPO 动态分组过滤
• 推荐动作：此 PR 是一个有价值且实现精良的功能增强。它通过提供开箱即用的标准奖励过滤路径，显著简化了 DAPO 动态采样的配置，并提升了其与核心奖励计算流程的集成度。实现方案清晰，通过分支逻辑兼容了新旧两种指标读取方式，并附带了针对性强的单元测试。考虑到默认值的变更可能对现有用户产生影响，建议在发布说明中明确提示此行为变化。整体而言，值得精读其核心逻辑分支和测试用例的设计，以理解如何优雅地扩展现有功能并保持兼容性。

功能与动机

PR body 明确指出，此变更旨在允许 V1 DAPO 分组过滤器使用存储在 rm_scores 中的标准 pre-KL 奖励。动机是简化配置：当 algorithm.filter_groups.metric=reward 时，ReplayBuffer 直接比较每条轨迹的 rm_scores 总和。其他指标名（如 acc）则继续从 reward_extra_info 读取。默认过滤指标被更改为 reward，使得用户可以通过单条命令 algorithm.filter_groups.enable=True 即可启用动态采样。

实现拆解

1. 核心逻辑扩展：在 verl/trainer/ppo/v1/replay_buffer.py 中，_dapo_filtered_keys 方法被重构。新增了 FILTER_GROUPS_REWARD_METRIC = "reward" 常量。方法内部通过 use_canonical_reward = self.filter_groups_metric == FILTER_GROUPS_REWARD_METRIC 判断是否使用标准奖励路径。若是，则从 TransferQueue 批量获取 rm_scores 字段并对其求和作为指标值；否则，沿用旧逻辑从 extra_fields 的 reward_extra_info 中读取。
2. 配置与契约调整：
   • verl/trainer/config/algorithm.py 中的 FilterGroupsConfig 文档字符串被更新，明确说明 "reward" 指标会从 rm_scores 读取。
   • verl/trainer/config/ppo_trainer.yaml 中 algorithm.filter_groups.metric 的默认值从 null 更改为 reward。
   • 所有 _generated_ppo_*_trainer.yaml 配置文件（megatron, torch titan, veomni 等）同步更新了此默认值。
3. 测试覆盖：在 tests/trainer/ppo/v1/test_replay_buffer_on_cpu.py 中，新增了 test_dapo_reward_metric_uses_canonical_rm_scores 测试用例。该测试构造了两组轨迹，一组在 rm_scores 上奖励一致但 reward_extra_info.acc 不一致，另一组则相反。测试验证了当 filter_groups_metric="reward" 时，只有前者被正确过滤，从而确保新逻辑的正确性。同时，PromptSpec 数据类和 RolloutProducer 被扩展以支持 canonical_rewards 参数。

关键文件：

• verl/trainer/ppo/v1/replay_buffer.py（模块 回放缓冲区；类别 source；类型 core-logic）: 包含 DAPO 分组过滤的核心逻辑变更，是本次功能的实现主体。
• tests/trainer/ppo/v1/test_replay_buffer_on_cpu.py（模块 回放缓冲区测试；类别 test；类型 test-coverage；符号 test_dapo_reward_metric_uses_canonical_rm_scores）: 包含针对新功能的单元测试，验证了使用标准奖励进行过滤的正确性。
• verl/trainer/config/ppo_trainer.yaml（模块 训练器配置；类别 config；类型 configuration）: 主配置文件，将 DAPO 过滤的默认指标从 null 更改为 reward，这是行为变更的核心。

关键符号：_dapo_filtered_keys

关键源码片段

verl/trainer/ppo/v1/replay_buffer.py

包含 DAPO 分组过滤的核心逻辑变更，是本次功能的实现主体。

# verl/trainer/ppo/v1/replay_buffer.py
# 标准奖励指标的魔术字符串
FILTER_GROUPS_REWARD_METRIC = "reward"
​
class ReplayBuffer:
    # ...
    def _dapo_filtered_keys(self, partition_id: str) -> tuple[set[str], Counter]:
        """Finished groups whose configured DAPO metric is identical across all trajectories.
        Returns the filtered uids and a ``{shared_metric_value: group_count}`` breakdown.
        """
        if partition_id == "val" or self.filter_groups_metric is None:
            return set(), Counter()
​
        finished_uids = self.finished_keys[partition_id]
        classification_cache = self._dapo_classification_cache[partition_id]
        # ... ( 缓存清理逻辑 ) ...
​
        new_finished_uids = finished_uids - classification_cache.keys()
        trajectory_keys = [key for key in self.partitions[partition_id] if key.split("_")[0] in new_finished_uids]
​
        # 核心分支：决定从 rm_scores 还是 extra_fields 读取指标
        use_canonical_reward = self.filter_groups_metric == FILTER_GROUPS_REWARD_METRIC
        metrics_by_uid: dict[str, list[float]] = defaultdict(list)
        missing_metric_uids = new_finished_uids - {key.split("_")[0] for key in trajectory_keys}
​
        if trajectory_keys:
            # 根据路径选择不同的字段
            data = tq.kv_batch_get(
                keys=trajectory_keys,
                partition_id=partition_id,
                select_fields=["rm_scores"] if use_canonical_reward else ["extra_fields"],
            )
            metric_data = list(data["rm_scores"] if use_canonical_reward else data["extra_fields"])
        else:
            metric_data = []
​
        for key, value in zip(trajectory_keys, metric_data, strict=True):
            uid = key.split("_")[0]
            if use_canonical_reward:
                # 标准奖励路径：对 rm_scores 求和作为组内指标值
                metrics_by_uid[uid].append(float(value.sum().item()))
            else:
                # 旧路径：从 reward_extra_info 字典中读取指定键
                extra_fields = getattr(value, "data", value)
                reward_extra_info = extra_fields.get("reward_extra_info", {}) if isinstance(extra_fields, dict) else {}
                if self.filter_groups_metric not in reward_extra_info:
                    missing_metric_uids.add(uid)
                else:
                    metrics_by_uid[uid].append(float(reward_extra_info[self.filter_groups_metric]))
​
        # ... ( 后续指标缺失检查与缓存更新逻辑 ) ...

tests/trainer/ppo/v1/test_replay_buffer_on_cpu.py

包含针对新功能的单元测试，验证了使用标准奖励进行过滤的正确性。

# tests/trainer/ppo/v1/test_replay_buffer_on_cpu.py
def test_dapo_reward_metric_uses_canonical_rm_scores(tq_init, partition_id):
    """测试当 filter_groups_metric 设为 "reward" 时，DAPO 过滤使用标准的 rm_scores。"""
    # 构造一组轨迹：rm_scores 总和相同 (1.0)，但 reward_extra_info.acc 不同
    same_reward = PromptSpec(
        uid=_uid(),
        status="finished",
        sessions=2,
        rewards=[0.0, 1.0], # 用于 extra_fields.acc
        canonical_rewards=[[0.25, 0.75], [1.0]], # 用于 rm_scores，总和均为 1.0
    )
    # 构造另一组轨迹：rm_scores 总和不同 (1.0)，但 reward_extra_info.acc 相同
    different_reward = PromptSpec(
        uid=_uid(),
        status="finished",
        sessions=2,
        rewards=[1.0, 1.0],
        canonical_rewards=[0.0, 1.0], # 总和为 1.0
    )
    _produce(partition_id, [same_reward, different_reward]).join_and_check()
​
    # 创建 ReplayBuffer，明确指定 filter_groups_metric="reward"
    rb = _make_rb(filter_groups_metric="reward", refill_fn=lambda _n: None)
    try:
        rb._sync_metadata_from_transfer_queue()
        filtered_uids, filtered_counts = rb._dapo_filtered_keys(partition_id)
​
        # 预期：只有 same_reward 组被过滤（因为其 rm_scores 标准奖励总和一致）
        assert filtered_uids == {same_reward.uid}
        assert dict(filtered_counts) == {1.0: 1}
    finally:
        _clear_partition(partition_id)

评论区精华

本 PR 的审核过程非常简洁，仅包含一条批准评论（reviewer: wuxibin89），无任何讨论或争议。这表明变更的意图和实现方案在团队内部得到了清晰共识，且测试覆盖充分，未引发额外设计或技术风险讨论。

• 暂无高价值评论线程

风险与影响

• 风险：

  • 影响：影响范围：直接影响所有使用 V1 训练器并启用 DAPO 动态采样（algorithm.filter_groups.enable=True）的训练流程。
    影响程度：这是一个行为变更，因为默认的过滤指标从 null（即不启用过滤）变更为 reward。这意味着，对于未显式设置 metric 的现有用户，在升级后，DAPO 动态采样将自动启用。这可能导致训练行为的改变（如样本筛选策略变化），需要用户在升级时注意。对于已经显式设置 metric 的用户，行为保持不变。
    系统影响：变更集中在训练器数据流的关键路径（ReplayBuffer），但通过清晰的配置和向后兼容的代码分支（use_canonical_reward）降低了风险。配套的配置文件更新确保了各后端（FSDP, Megatron, TorchTitan, VeOmni）的一致性。

• 风险标记：默认配置变更可能影响现有用户行为, 核心数据路径变更

关联脉络

• PR #7724 [veomni] feat: add async_activation_offload config: 同为 V1 训练器的功能增强 PR，且都涉及核心训练流程或配置的调整，属于同一功能演进方向上的并行工作。