verl One Step Off Policy：异步并行生成与训练的离策略 RL 训练方案详解

verl One Step Off Policy：异步并行生成与训练的离策略 RL 训练方案详解

【免费下载链接】verlverl/HybridFlow: A Flexible and Efficient RL Post-Training Framework项目地址: https://gitcode.com/GitHub_Trending/ve/verl

导读

本文深入剖析 verl（HybridFlow）中位于 verl/experimental/one_step_off_policy 的One Step Off Policy Async Trainer配方（Recipe）：它通过将样本生成（Rollout）与模型训练并行化，用上一步生成的样本训练当前模型，从而消除长尾生成场景下 GPU 的空闲等待，显著提升 RL 后训练吞吐。读完本文，你将掌握其异步流水线设计、基于 NCCL 的参数同步机制、PPO 正确性保障，以及 FSDP2 / Megatron 两种引擎下的资源划分配置方法。

背景：同步 RL 训练的效率瓶颈

verl 默认的强化学习训练流程是同步的，遵循 PPO、GRPO、DAPO 等既有算法的经典工作流：每一步用最新模型生成训练样本，训练完成后再更新模型。这种"边生成、边训练"的对齐方式符合离策略（off-policy）强化学习范式，也能稳定 RL 训练，但存在严重的效率问题：

• 模型更新必须等待生成阶段最长的那条输出完成；
• 在长尾样本（long-tail）生成期间，GPU 处于空闲状态，利用率显著不足；
• 样本生成的长尾问题越严重，整体训练效率越低。

以 DAPO 32B 训练为例，Rollout 阶段约占总体时间的70%，且单纯增加资源并不能缩短 Rollout 时长。这是因为生成阶段的耗时由最慢样本决定，而非由算力决定。

解决方案：One Step Off Policy 异步训练

为缓解上述问题，verl 实现了One Step Off Async Trainer，核心思想是把生成与训练两个阶段并行化：

1. 并行生成与训练（Parallel Generation and Training）：在训练当前 batch 的同时，异步生成下一 batch 的样本；
2. 资源隔离（Resource Isolation）：与hybrid_engine不同，本方案要求为 rollout 显式分配资源，剩余资源自动分配给训练；
3. NCCL 参数同步（NCCL Parameter Synchronization）：使用 NCCL 通信原语在生成与训练模块之间无缝传递模型参数。

整个过程中，生成和训练使用的模型参数始终保持"一步滞后"（one-step off）的策略：当前训练使用上一步生成的样本，而当前生成则基于训练前的模型权重。

该设计参考了异步 RL 领域的相关工作：AReaL（Large-Scale Asynchronous Reinforcement Learning System for Language Reasoning）与 Asynchronous RLHF（Faster and More Efficient Off-Policy RL for Language Models）。

实验效果

原文档给出的实测数据基于以下配置：

• 机器配置：2 节点，每节点 16 张 H20 GPU
  • 生成（Generation）：4 张 GPU
  • 训练（Training）：12 张 GPU
• 模型：Qwen2.5-Math-7B
• 最大响应长度：FSDP2 为 20,480 tokens；Megatron 为 8,192 tokens
• 算法：DAPO
• Rollout 引擎：vLLM

• colocate sync 模式下：step ≈ gen + old_log_prob + update_actor
• one-step-overlap async 模式下：step ≈ wait_prev_gen + old_log_prob + update_actor

可以看到，异步模式下gen（同步生成耗时）被完全隐藏在训练阶段背后，只残留少量wait_prev_gen（等待上一轮生成收尾的时间），FSDP2 与 Megatron 两种引擎分别获得了约 23% 与 40% 的端到端加速，且 acc 指标略有提升或持平。

实现：One Step Off Policy 异步流水线

核心训练循环

One Step Off 异步流水线以极低成本无缝嵌入既有训练逻辑，无需额外的样本存储管理。核心机制是用async_gen_next_batch驱动异步 rollout 生成，并通过create_continuous_iterator在 epoch 切换期间保持数据流连续。

在 ray_trainer.py 中，OneStepOffRayTrainer继承自分离式SeparateRayPPOTrainer，其构造阶段断言hybrid_engine必须关闭，并从role_worker_mapping中移除Role.Rollout，交由 AgentLoop 动态创建 rollout 资源。

数据迭代器跨 epoch 连续产出训练样本：

def _create_continuous_iterator(self): """ Create a continuous data iterator across epoch """ for epoch in range(self.config.trainer.total_epochs): iterator = iter(self.train_dataloader) for batch_dict in iterator: yield epoch, batch_dict

异步生成下一 batch 样本（先读数据、再同步参数、后异步生成）：

async def _async_gen_next_batch(self, continuous_iterator): try: epoch, batch_dict = next(continuous_iterator) except StopIteration: return None ... batch = DataProto.from_single_dict(batch_dict) batch.non_tensor_batch["uid"] = np.array([str(uuid.uuid4()) for _ in range(len(batch.batch))], dtype=object) gen_batch = self._get_gen_batch(batch) gen_batch.meta_info["global_steps"] = self.global_steps gen_batch_output = gen_batch.repeat(repeat_times=self.config.actor_rollout_ref.rollout.n, interleave=True) # async generation with marked_timer("generate_async", timing_raw, color="purple"): gen_batch_output = await self.async_rollout_manager.generate_sequences(gen_batch_output) ... return metrics, timing_raw, epoch, batch, future_reward

训练主循环（fit）在进入迭代前先启动第一轮异步生成，实现一步滞后：

# across epoch iterator continuous_iterator = self._create_continuous_iterator() # Start the first asynchronous generation task. batch_data_future = asyncio.create_task(self._async_gen_next_batch(continuous_iterator)) while batch_data_future is not None: batch_data_future = await self.fit_step(batch_data_future, continuous_iterator) if self.is_last_step: return

每一步（fit_step）中，先等待上一轮生成结果，随即启动下一轮异步生成，再继续执行 reward / log_prob / advantage / critic / actor 更新等训练阶段；各阶段之间穿插await asyncio.sleep(0)，确保异步任务能及时得到事件循环调度：

async def _fit_generate(self, batch_data_future, continuous_iterator): with marked_timer("gen", timing_raw, color="red"): _metrics, _timing_raw, epoch, batch, future_reward = await batch_data_future ... # sync weights from actor to rollout with marked_timer("sync_rollout_weights", timing_raw, color="purple"): self._fit_update_weights() # async next generation if not self.is_last_step: batch_data_future = asyncio.create_task(self._async_gen_next_batch(continuous_iterator)) await asyncio.sleep(0) else: batch_data_future = None return batch, batch_data_future

参数同步：基于 NCCL 的高效权重同步

One Step Off 方案最亮眼之处在于基于 NCCL 的 rollout 权重同步性能：绝大多数情况下延迟低于 300ms，对 RLHF 训练流程而言几乎可以忽略。

同步的建立分两步走：

1. 交换参数元信息：actor worker 通过get_actor_weights_info暴露参数（key、shape、dtype）元信息，rollout worker 通过set_actor_weights_info接收；随后驱动进程在 actor 与 rollout 的 worker 集合上创建名为actor_rollout的 NCCL 通信组：

# rollout obtains the meta-info of model parameters from the actor for parameter sync weights_info = self.actor_wg.get_actor_weights_info()[0] self.rollout_wg.set_actor_weights_info(weights_info) # Create an actor-rollout communication group for parameter sync actor_rollout_workers = self.actor_wg.workers + self.rollout_wg.workers collective.create_collective_group( actor_rollout_workers, len(actor_rollout_workers), list(range(0, len(actor_rollout_workers))), backend="nccl", group_name="actor_rollout" )

2. 逐张量广播：驱动进程分别触发 actor 与 rollout 的sync_rollout_weights，actor 侧准备参数（若为分片参数则取full_tensor()），rollout 侧定位 vLLM 推理引擎内部的模型对象，随后按元信息逐张量执行 NCCLbroadcast（src_rank=0）并写入推理模型：

# drive process call the actor and rollout respectively to sync parameters by nccl def sync_rollout_weights(self): self.actor_wg.sync_rollout_weights() ray.get(self.rollout_wg.sync_rollout_weights()) # fsdp model parameter sync @register(dispatch_mode=Dispatch.ONE_TO_ALL, blocking=False) def sync_rollout_weights(self): params = self._get_actor_params() if self._is_actor else None if self._is_rollout: inference_model = ( self.rollout.inference_engine.llm_engine.model_executor.driver_worker.worker.model_runner.model ) from verl.utils.vllm.patch import patch_vllm_moe_model_weight_loader patch_vllm_moe_model_weight_loader(inference_model) # Model parameters are broadcast tensor-by-tensor from actor to rollout for key, shape, dtype in self._weights_info: tensor = torch.empty(shape, dtype=dtype, device=get_torch_device().current_device()) if self._is_actor: assert key in params origin_data = params[key] if hasattr(origin_data, "full_tensor"): origin_data = origin_data.full_tensor() if torch.distributed.get_rank() == 0: tensor.copy_(origin_data) from ray.util.collective import collective collective.broadcast(tensor, src_rank=0, group_name="actor_rollout") if self._is_rollout: inference_model.load_weights([(key, tensor)])

说明：上述代码片段摘自原文档，用于展示参数同步的核心思路。实际仓库中该逻辑经由 checkpoint 引擎（checkpoint_engine.backend: "nccl"）与_fit_update_weights统一调度，具体见 ray_trainer.py 与分离式训练器 separation/ray_trainer.py。

PPO 正确性：rollout log_probs 与重要性采样

为保证异步训练下 PPO 算法的正确性，本方案使用rollout 阶段的 log_probs进行 PPO 重要性采样（importance sampling），即用行为策略（生成样本时的旧模型）的 log_prob 修正策略比率，算法细节可参考 verl 文档中的 rollout_corr_math（Rollout Correction 数学推导）。

默认开启bypass_ppo_clip（即algorithm.rollout_correction.bypass_mode=True）模式，也可以探索其他修正策略。对应配置位于 one_step_off_ppo_trainer.yaml：

# Only then will the use of log probs be correct. # And it can be used in conjunction with other rollout_correction algorithms. algorithm: rollout_correction: bypass_mode: True

Rollout Correction 的默认行为定义在 rollout_correction.yaml 中，其中bypass_mode指定使用compute_policy_loss_bypass_mode()并配合loss_type选择损失函数类型（详见 algorithm.py）。

AgentLoop：多轮工具调用支持

当前实现不再提供 SPMD 模型 rollout 模式，而是切换为AgentLoop 模式，该模式同时支持多轮工具调用（multi-turn tool calling）。在 ray_trainer.py 的_init_async_rollout_manager中：

• 要求config.actor_rollout_ref.rollout.mode == "async"；
• 支持通过agent_loop_manager_class配置自定义 AgentLoopManager，默认使用verl.experimental.agent_loop中的AgentLoopManager；
• 通过LLMServerManager创建并持有 LLM 服务客户端。

使用方式

入口

One Step Off 训练入口为 main_ppo.py，它通过@hydra.main加载one_step_off_ppo_trainer配置，在main()中将顶层config.rollout的nnodes/n_gpus_per_node同步到config.actor_rollout_ref.rollout，随后调用run_ppo(config, task_runner_class=OneStepTaskRunner)启动训练。OneStepTaskRunner负责创建资源池、数据集、tokenizer 与OneStepOffRayTrainer，并asyncio.run(trainer.fit())。

FSDP2 配置示例

python3 -m verl.experimental.one_step_off_policy.async_main_ppo \ --config-path=config \ --config-name='one_step_off_ppo_trainer.yaml' \ actor_rollout_ref.actor.strategy=fsdp2 \ # actor and rollout are placed separately actor_rollout_ref.hybrid_engine=False \ # actor and rollout resource trainer.nnodes=1 \ trainer.n_gpus_per_node=6 \ rollout.nnodes=1 \ rollout.n_gpus_per_node=2

注意：仓库中实际的入口模块名为verl.experimental.one_step_off_policy.main_ppo（见 shell 脚本与 main_ppo.py），FSDP2 与 Megatron 配置示例分别见 one_step_off_ppo_trainer.yaml 与 one_step_off_ppo_megatron_trainer.yaml。训练侧 GPU 数量通过trainer.n_gpus_per_node指定，rollout 侧通过rollout.n_gpus_per_node指定，二者之和不超过物理 GPU 总数。

Megatron 配置示例

python3 -m verl.experimental.one_step_off_policy.async_main_ppo \ --config-path=config \ --config-name='one_step_off_ppo_megatron_trainer.yaml' \ actor_rollout_ref.actor.strategy=megatron \ # actor and rollout are placed separately actor_rollout_ref.hybrid_engine=False \ # actor and rollout resource trainer.nnodes=1 \ trainer.n_gpus_per_node=6 \ rollout.nnodes=1 \ rollout.n_gpus_per_node=2

关键配置项说明

两份 YAML 配置（FSDP2 与 Megatron 版内容一致）中有三个必须遵守的约束（详见 one_step_off_ppo_trainer.yaml）：

rollout: nnodes: 1 # rollout 使用的节点数 n_gpus_per_node: 8 # 每节点 GPU 数 actor_rollout_ref: rollout: # 必须关闭！否则无法进行参数同步。 free_cache_engine: False # 必须开启！否则无法计算 log_probs。 calculate_log_probs: True checkpoint_engine: backend: "nccl" algorithm: rollout_correction: bypass_mode: True

配置指南：卡数与资源调优

卡数关系

为保证训练样本能在训练 GPU 上均匀分布，需满足以下任一关系：

• actor_rollout_ref.rollout.n应为trainer.n_gpus_per_node * trainer.nnodes的整数约数；
• 或actor_rollout_ref.rollout.n * data.train_batch_size应能被trainer.n_gpus_per_node * trainer.nnodes整除。

理由：使用部分资源做生成时，确保训练样本可以均匀切分到各训练 GPU，避免负载不均。

动态资源调优

根据各阶段耗时调整trainer.nnodes、trainer.n_gpus_per_node、rollout.nnodes、rollout.n_gpus_per_node：

• 理想状态：Rollout 与训练阶段耗时相当；
• 诊断指标：
  • 监控wait_prev_gen（等待上一轮 rollout 结束、未被完全重叠的时间）耗时；
  • 分析sequence_length（序列长度）分布；
• 调整策略：
  • wait_prev_gen高 + 序列长度均匀 →增加 rollout 资源；
  • wait_prev_gen高 + 长尾序列 → 优化停止准则（增加资源无济于事）。

资源受限场景：通过调整 GPU 分配比例来优化资源利用率，保持节点数相等，让训练与 rollout 共享节点：

• 配置trainer.nnodes = rollout.nnodes，且trainer.n_gpus_per_node + rollout.n_gpus_per_node = physical_gpus_per_node，通过调整n_gpus_per_node控制 rollout 资源分配。

资源充足场景：通过调整节点数优化性能，保持每节点 GPU 数相等，使训练与 rollout 并行度可独立扩展：

• 配置trainer.n_gpus_per_node = rollout.n_gpus_per_node，通过调整trainer.nnodes与rollout.nnodes控制 rollout 资源分配。

注意：系统实际所需节点总数并非简单的trainer.nnodes + rollout.nnodes，需按 GPU 容量计算：

• 当trainer.n_gpus_per_node + rollout.n_gpus_per_node <= physical_gpus_per_node时，所需节点数为max(trainer.nnodes, rollout.nnodes)；
• 当trainer.n_gpus_per_node + rollout.n_gpus_per_node > physical_gpus_per_node时，所需节点数为trainer.nnodes + rollout.nnodes。

现成脚本示例

仓库 shell 目录提供了多组开箱即用的启动脚本，涵盖不同引擎与资源配置：

• FSDP2 + vLLM：dapo_7b_math_fsdp2_4_12.sh（4 卡 rollout / 12 卡训练）、dapo_7b_math_fsdp2_64_64.sh、dapo_7b_math_fsdp2_colocate.sh（同步基线对比）
• FSDP2 + SGLang：dapo_7b_math_fsdp2_sglang_4_12.sh、dapo_7b_math_fsdp2_sglang_colocate.sh
• Megatron + vLLM：dapo_7b_math_megatron_4_12.sh、dapo_7b_math_megatron_colocate.sh
• 轻量入门（GRPO + GSM8K）：grpo_0.6b_gsm8k_fsdp2_2_6.sh、grpo_0.6b_gsm8k_fsdp2_sglang_2_6.sh（含 delta 分片变体 grpo_0.6b_gsm8k_fsdp2_sglang_delta_sharded_2_6.sh）
• NPU：grpo_qwen3_8b_gsm8k_fsdp2_8_8_npu.sh

以 dapo_7b_math_fsdp2_4_12.sh 为例，脚本通过NNODES/NGPUS_PER_NODE环境变量控制总体规模，并计算n_gpus_rollout与n_gpus_training后分别传给rollout.*与trainer.*；算法侧为 DAPO 风格配置（clip_ratio_low=0.2、clip_ratio_high=0.28、overlong buffer 惩罚等）。Megatron 版脚本额外指定了train_tp=2、train_pp=2的模型并行切分。需要留意脚本中提示：Qwen2.5-Math-7B 需将config.json中的max_position_embeddings修改为 32768 以配合 32K 上下文训练。

功能支持矩阵

小结

One Step Off Policy Async Trainer 是 verl 生态中解决同步 RL 训练长尾等待问题的关键配方：通过一步滞后的异步流水线把生成与训练重叠起来，用显式资源隔离替代 hybrid engine 的隐式调度，并以 NCCL 广播实现毫秒级参数同步。实测在 DAPO + Qwen2.5-Math-7B 场景下带来约 23%（FSDP2）至 40%（Megatron）的端到端加速。若你在训练中观测到wait_prev_gen长期偏高且序列长度分布均匀，优先考虑为 rollout 增配资源；若瓶颈源于长尾序列本身，则应转向停止准则等采样侧优化。相关实现、配置与脚本均可在 verl/experimental/one_step_off_policy 目录下进一步查阅。

【免费下载链接】verlverl/HybridFlow: A Flexible and Efficient RL Post-Training Framework项目地址: https://gitcode.com/GitHub_Trending/ve/verl