# PR #7792 完整报告

- 仓库：`verl-project/verl`
- 标题：[trainer, cfg] feat: filter V1 DAPO groups by canonical reward by default
- 合并时间：2026-09-08 20:43
- 原文链接：http://prhub.com.cn/verl-project/verl/pull/7792

---

# 执行摘要

- 一句话：默认使用标准 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 分组过滤的核心逻辑变更，是本次功能的实现主体。

```python
# 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`

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

```python
# 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，且都涉及核心训练流程或配置的调整，属于同一功能演进方向上的并行工作。