执行摘要

• 一句话：支持 replay buffer 陈旧性控制策略 drop/wait
• 推荐动作：值得精读 replay_buffer.py 中的 _has_enough_samples 和 _drop_max_off_policy_samples 方法，理解两种策略的设计权衡。注意 review 中关于 parameter_sync_step 断言和 global_steps 安全访问的建议，建议在后续 PR 中跟进修复。同时关注配置迁移对现有训练任务的影响。

功能与动机

Follow up PR #6710（统一同步/异步 trainer 抽象），需要支持对 off-policy 轨迹的陈旧性控制，以适配异步训练场景中不同版本模型产生的轨迹质量差异，避免陈旧轨迹损害训练效果。

实现拆解

1. 修改 replay_buffer.py：重写 __init__，新增 trainer_mode、trainer_config、max_off_policy_threshold、max_off_policy_strategy 和 sampler_kwargs 参数，并添加参数校验断言。新增 _has_enough_samples 方法根据策略判断采样条件：'wait' 策略会检查 in-flight 提示的陈旧性是否超过阈值，若超过则阻塞；'drop' 策略忽略 in-flight 但采样后丢弃超标轨迹。新增 _drop_max_off_policy_samples 方法实现丢弃逻辑。修改 sample 方法接受 global_steps 参数，返回 (KVBatchMeta, metrics) 元组。
2. 修改 trainer_base.py：新增 _build_replay_buffer 方法，从配置加载 sampler 配置，并支持通过 custom_sampler 字段动态加载自定义 ReplayBuffer 子类。在 __init__ 中调用该方法替换硬编码的 ReplayBuffer()。
3. 更新配置：在 ppo_trainer.yaml 及三个生成的配置文件中添加 sampler 区块，包含 max_off_policy_threshold（默认 8）、max_off_policy_strategy（默认 drop）、custom_sampler 和 sampler_kwargs。
4. 更新单元测试：在 test_replay_buffer_on_cpu.py 中新增辅助函数 _make_rb、_sample、_set_prompt_status，并添加测试用例覆盖参数校验、'drop' 策略下忽略 in-flight 采样、'wait' 策略下阻塞陈旧 in-flight 等场景。

关键文件：

• verl/trainer/ppo/v1/replay_buffer.py（模块 训练器；类别 source；类型 core-logic；符号 init, sample, _has_enough_samples, _drop_max_off_policy_samples）: 核心变更文件，实现了陈旧性控制的全部逻辑：新增参数、校验、采样策略方法 _has_enough_samples 和 _drop_max_off_policy_samples，修改 sample 返回指标。
• tests/trainer/ppo/v2/test_replay_buffer_on_cpu.py（模块 测试；类别 test；类型 test-coverage；符号 _make_rb, _sample, _set_prompt_status, init）: 配套测试覆盖了参数校验、drop 策略忽略 in-flight、wait 策略阻塞陈旧 in-flight 等关键场景，验证新逻辑正确性。
• verl/trainer/ppo/v1/trainer_base.py（模块 训练器；类别 source；类型 core-logic；符号 _build_replay_buffer）: 新增 _build_replay_buffer 方法将配置传入 ReplayBuffer，支持自定义 sampler，是集成到训练流程的关键入口。

关键符号：init, sample, _has_enough_samples, _drop_max_off_policy_samples, _build_replay_buffer, _make_rb, _sample, _set_prompt_status

关键源码片段

verl/trainer/ppo/v1/replay_buffer.py

核心变更文件，实现了陈旧性控制的全部逻辑：新增参数、校验、采样策略方法 _has_enough_samples 和 _drop_max_off_policy_samples，修改 sample 返回指标。

def __init__(
    self,
    trainer_mode: str,
    trainer_config: DictConfig,
    max_off_policy_threshold: int,
    max_off_policy_strategy: str,
    sampler_kwargs: DictConfig,
    poll_interval: float = 2.0,
):
    # store off-policy control parameters
    self.trainer_mode = trainer_mode
    self.trainer_config = trainer_config
    self.max_off_policy_threshold = max_off_policy_threshold
    self.max_off_policy_strategy = max_off_policy_strategy
    self.sampler_kwargs = sampler_kwargs
    self.poll_interval = poll_interval
    # extract from trainer_config to calculate staleness
    self.parameter_sync_step = trainer_config.get("parameter_sync_step", 1)
​
    # validate threshold and strategy
    assert isinstance(self.max_off_policy_threshold, int) and self.max_off_policy_threshold > 0, (
        f"Invalid max off policy threshold: {self.max_off_policy_threshold}, must be an integer greater than 0"
    )
    assert self.max_off_policy_strategy in ["drop", "wait"], (
        f"Invalid max off policy strategy: {self.max_off_policy_strategy}, must be one of ['drop', 'wait']"
    )
​
    # existing metadata containers (unchanged)
    self.partitions: dict[str, dict[str, dict]] = defaultdict(dict)
    self.pending_keys: dict[str, set] = defaultdict(set)
    self.running_keys: dict[str, set] = defaultdict(set)
    self.finished_keys: dict[str, set] = defaultdict(set)
    self.failure_keys: dict[str, set] = defaultdict(set)
    self.prompt_global_steps: dict[str, dict[str, int]] = defaultdict(dict)

评论区精华

gemini-code-assist[bot] 在 review 中提出三个建议：

1) 在 replay_buffer.py 中对 parameter_sync_step 添加正值断言，防止零除错误；
2) 使用 tag.get("global_steps", 0) 代替直接索引 tag["global_steps"]，避免 KeyError；
3) 在 trainer_base.py 中添加同样的 parameter_sync_step 断言。这些建议未在 PR 中明确回复或修正，但 PR 已被合并。

• parameter_sync_step 未添加正值断言 (correctness): 建议未被采纳或关闭，PR 已合并但断言未被添加。
• 使用 tag.get 代替直接索引 global_steps (correctness): 建议未被采纳或关闭，PR 已合并但直接索引保留。

风险与影响

• 风险：
  1. KeyError 风险：_sync_metadata_from_transfer_queue 中将 tag.get("global_steps", 0) 改为直接索引 tag["global_steps"]，若旧版本或自定义集成未设置该字段，将引发 KeyError。
  2. 零除风险：parameter_sync_step 用于计算轨迹陈旧性，缺失正值断言可能因配置错误导致 ZeroDivisionError。
  3. 默认配置兼容性：新增 sampler 配置区块有默认值，但若用户配置覆盖不完整可能导致缺失键错误。
  4. 测试覆盖：测试主要覆盖 CPU 场景，未覆盖 GPU 或真实分布式环境。
     - 影响：影响所有使用 PPO v1 trainer 的同步和异步训练流程。默认配置下 old-policy 阈值为 8，策略为 'drop'，行为与之前基本一致（当 global_steps 较小时无影响）。异步训练（colocate_async/separate_async）可通过配置策略控制采样阻塞或丢弃，提升训练稳定性。用户需更新配置文件（若自定义配置不含 sampler 区块则会使用默认值，向后兼容）。
     - 风险标记：配置健壮性, 向后兼容性, 未修复的 Review 建议

关联脉络

• PR #6710 [trainer] feat: add unify trainer abstraction for sync and async training: 本 PR 明确声明是 6710 的 follow-up，在该 PR 的统一抽象基础上增加了陈旧性控制策略。
• PR #6736 [trainer] feat: add off_policy metrics: 6736 也涉及 off-policy 相关指标计算，本 PR 在 sample 返回中增加了 off_policy_metrics，两条线可能共享部分设计。