verl 的 RL 数据流：一个 batch 怎样从 rollout 走到 PPO 更新

读完这节你应该知道：为什么 DataProto 要分三层，以及「reward 放在哪个 token 上」为什么会直接影响 loss 和 metrics。

先看一步训练里 batch 需要装什么。模型前向和 loss 需要 per-token 的 responses、attention_mask、response_mask、 old_log_probs、ref_log_prob、values、advantages、returns； reward 和验证需要 per-sample 的 uid、data_source、reward_model；调度和日志还需要 per-step 的 temperature、global_steps、global_token_num。这三类数据的形状和生命周期都不一样。

最关键的约定是 reward 的落点。默认路径把一条序列的 outcome reward 压成一个 float，写在最后一个有效 response token 上（仅当 L_i>0）： 其余 token 为 0。L_i=0 时索引 valid_response_length - 1 = -1 会写到末列（padding 位置），这是源码的边界行为。 这样 token_level_scores.sum(-1) 仍能还原成每条序列的标量 reward （verl/trainer/ppo/metric_utils.py:453）。

这个落点不是固定的 R − 1，而是 Li − 1。因为 batch 里的 response 是右填充的，不同序列的有效长度不同； 如果把 reward 写在固定末列，短序列的 reward 会落进 padding。batch 路径的 assemble_rm_scores 用 valid_response_length - 1 做 scatter（base.py:79）；单条轨迹路径在 as_dict 里直接写 rm_scores[-1]， 因为那条轨迹还没有被 batch padding（verl/experimental/agent_loop/agent_loop.py:147-148）。两条路径的索引写法不同，但在 L_i>0 时都指向「最后一个有效 token」； L_i=0 时 batch 路径的索引 −1 会写到末列（padding 位置），这是源码的边界行为。

这也解释了 metrics 为什么容易读错。同一个 batch，response_length/mean 会把 abort 的 0 长度样本算进去， response_length_non_aborted/mean 则把它们排除 （verl/trainer/ppo/metric_utils.py:573、:581）。分母不同，两个数就不是同一个东西。 后面的 03 讲 DataProto 怎么保证三层对齐，04 讲 token-mean 的分母，05 讲一步的调用顺序，06 讲每个指标的分母。

读完这节你应该知道：逐样本打分和批量 reward model 各自适合什么场景，以及 trainer 在什么条件下会跳过其中一条。

路线 A 是逐样本打分。reward loop 的 worker 在 compute_score_batch 里为每个样本创建一个 asyncio task， 调用 self.compute_score（verl/experimental/reward_loop/reward_loop.py:138-155）：默认规则 reward 或自定义 reward function 走 run_single（:148、:155），启用 reward model 时走 compute_score_disrm（:153）。 默认的 NaiveRewardManager 只解码 response（prompt 不参与打分），读取 data_source 和 ground_truth，调用 compute_score，返回一个 {"reward_score": ..., "reward_extra_info": ...}（verl/experimental/reward_loop/reward_manager/naive.py:34-99）。 更早的 legacy 版本在 verl/workers/reward_manager/naive.py:90-176 里逐条 for i in range(len(data))， 它会同时解码 prompt 与 response；它的 docstring 明确说：一次慢的 compute_score 会阻塞整个训练循环，所以加了 SIGALRM 超时保护 （naive.py:32-40）。

路线 B 是批量 reward model。trainer 侧的 compute_rm_score 先把 batch 补到 worker 数的整数倍， 再 chunk(num_workers) 分给各个 worker，收集后按原始长度截断，最后把 scores 装配成 (B, R) 的 rm_scores （verl/experimental/reward_loop/reward_loop.py:343-363）。装配同样是「每行一个写入位置」： rm_scores[arange(B), valid_response_length - 1] = scores（base.py:76-81）。

trainer 只在一种情况下会走路线 B：use_rm 为真且 batch 里还没有 rm_scores （verl/trainer/ppo/ray_trainer.py:1563-1565）。如果 agent loop 已经把 rm_scores 放进 rollout 输出， trainer 直接调用 extract_reward 读出来（:1568；verl/trainer/ppo/reward.py:160-167）。 两条路线最终都落到同一个字段，DataProto 不需要知道 reward 是谁算的。

表 1 · reward 的两条落点（页面整理；出处见各单元格）

两条路线之外，还存在一条「复用」路径：如果 batch 里已经有 rm_scores，trainer 不会重新算 （ray_trainer.py:1563）。这也是为什么 reward model 和 rule-based reward 可以在同一个 trainer 里切换： 切换的是谁写 rm_scores，不是下游 loss 和 metrics 的读法。

读完这节你应该知道：DataProto 的哪个字段会被 chunk/split 切开，哪个字段在每个分片里保持同一份，以及 union 为什么不会静默覆盖。

DataProto 的定义只有三个字段：batch 是 TensorDict，non_tensor_batch 是 dict[str, np.ndarray]， meta_info 是普通 dict（verl/protocol.py:317-328）。 check_consistency 强制 tensor 的 batch 维只有一维；batch is not None 且 non_tensor_batch 非空时， non_tensor 的每个数组必须是 numpy.ndarray，且 shape[0] 等于 batch size（verl/protocol.py:451-474）。

chunk：前两层按行切，meta_info 原样传给每个分片

chunk(chunks) 先在 dim0 上切 TensorDict，再按同样的 chunk 边界用 np.array_split 切每个 non_tensor 数组， 最后把同一个 meta_info 传给每个输出（verl/protocol.py:861-900）。如果没开 auto padding， 它还要求 len(self) % chunks == 0，否则直接断言失败（:870-872）。

这种设计让 temperature、global_steps 这类 per-step 标量不会在切分时被复制成 B 份 （ray_trainer.py:1479、:1489），也让 uid、data_source、reward_model 这些 per-sample 对象跟着行一起走。切分语义由字段决定，不需要每个调用点自己记。

其它算子：select / reorder / repeat 都同时作用于两层

select_idxs 用一个索引数组同时索引 TensorDict 和每个 non_tensor 数组（verl/protocol.py:632-670）； reorder 在原地重排两层（:960-966）；repeat 在 interleave=True 时用 repeat_interleave/np.repeat，interleave=False 时 tensor 走 expand+reshape、non_tensor 走 np.tile， 两条分支都保持行对应（:968-1010，np.tile 在 :1001-1004）。split(split_size) 只是按切片反复调用 __getitem__（:902-911）。这些算子都不碰 meta_info，或者只把同一个 dict 传下去。

union：同名 key 必须相等，否则报错而不是覆盖

一步 RL 会多次调用 batch.union(new_tensor)：rollout 输出、old_log_probs、ref_log_prob、values 都是这样并进 batch 的。 union 分别合并三层（verl/protocol.py:778-796），每一层都有自己的相等性检查：

• tensor：两个 TensorDict 的 batch size 必须相同；同名 key 必须 equal，否则断言失败（verl/protocol.py:109-122）。
• non_tensor：同名数组做深比较，NaN 和 object dtype 也覆盖（verl/protocol.py:188-199）。
• meta_info：同名 key 必须 ==，否则断言失败（verl/utils/py_functional.py:151-168）。

这带来一个容易忽略的行为：union 是原地修改左操作数并返回 self （verl/protocol.py:792-795）。batch = batch.union(x) 看起来像纯函数，实际会改掉原来的 batch； 如果别处还持有同一个引用，会一起看到新 key。

padding：重复样本凑整除，collect 后再把尾部切掉

pad_dataproto_to_divisor 在长度不能被整除时，从 batch 头部取样本重复，拼到尾部，返回 pad_size （verl/protocol.py:85-99）。dispatch 层把 pad_size 记进 kwargs（verl/single_controller/base/decorator.py:91-117）， worker 返回后 collect 层用 output.select_idxs(indices[:-padding_count]) 把尾部切掉 （verl/single_controller/ray/base.py:53-63）。所以对调用方来说 padding 是透明的； 但对 worker 内部计算的 per-sample 指标，重复样本仍然参与过统计（这一点在 07 展开）。

读完这节你应该知道：response_mask 怎么从 attention_mask 来、全局 token 分母怎么在 DP 之间对齐、reward 打分为什么先补齐再分片。

loss 的 token 权重来自 response_mask。trainer 先取 response 张量的宽度 R，再切 attention_mask[:, -R:]（verl/trainer/ppo/ray_trainer.py:132-135）。 response 张量是右填充的，模型输出回到 padding 布局时也在右侧补零（verl/workers/utils/padding.py:141-144）， 所以最后 R 列正好是 response 段。到了 engine 里，loss_mask 直接等于 response_mask （verl/workers/utils/padding.py:77）。

agg_loss 的 token-mean 分支做的是「masked sum ÷ 有效 token 数」： 如果 batch_num_tokens 没给，它就用本地 loss_mask.sum()；当 dp_size > 1 时，它要求调用方传入全局 batch_num_tokens，否则直接抛错（verl/trainer/ppo/core_algos.py:1170-1175）。

这个 global_N 不是 trainer 手算的。FSDP engine 在每个 rank 上先算 batch_num_tokens = data["loss_mask"].sum()，再对 DP 组做一次 ReduceOp.SUM，最后写回 TensorDict 的 non-tensor 区 （verl/workers/engine/fsdp/transformer_impl.py:672-676）。 推断（由 core_algos.py:1175 的 * dp_size 与 DDP 的梯度平均语义推出）： 每个 rank 的 loss 先乘 dp_size，DDP 再对梯度取平均，两步合起来等价于用全局 token 数做一次平均。

分片还有一层：_balance_batch 按每条序列的 workload 把所有序列重新排到各个 DP rank，使每个 rank 的总负载接近 （verl/trainer/ppo/ray_trainer.py:1182-1250）。workload 不是原始 token 数，而是 24576 * seqlen + seqlen²（verl/utils/seqlen_balancing.py:27-46），用 attention 的线性项与二次项近似 FLOPs。 它只改顺序，不改 token 总数；代码注释也说明这会改变 mini-batch 的组成，但 advantage 按 uid 计算，不受顺序影响 （:1546-1548）。

reward 打分的 worker 池：先补齐，再分片，最后按原长度截断

默认配置开 8 个 reward worker（verl/trainer/config/reward/reward.yaml:5）。 compute_rm_score 把 batch 补到 worker 数的整数倍，chunk 成等长分片，每个 worker 并发处理， 收集后把 padding 部分截掉（verl/experimental/reward_loop/reward_loop.py:343-358）。 这和 03 节的 DataProto padding 是同一套思路：先凑整除，再把尾部丢掉。

在默认的规则 reward 路径上，agent loop 会从可用的 reward worker handle 里随机选一个，远程调用 compute_score， 把返回的 reward_score 写进轨迹输出（verl/experimental/agent_loop/agent_loop.py:1113-1115）。 handle 只在「RM 关闭」或「RM 使用独立 resource pool」时返回（reward_loop.py:320-322）； 否则 reward 由 trainer 在 rollout 之后统一计算（ray_trainer.py:1563-1565）。

表 2 · 本 commit 的默认开关（页面整理；出处见各行）

下面把这个 batch 的 token 账本做成可调参数的模拟。B 是 prompt 数，n 是 rollout.n，P 是 prompt 长度，R 是 response 长度。 它展示五个量：序列数 B×n、有效 response token 数、rm_scores 写入位置数、总 token 数，以及 token-mean 的每 token 归一化权重。

读完这节你应该知道：每个阶段往 DataProto 里加了哪些 key，哪些阶段是可选的，以及 metrics 在什么时候被收集。

1. 采样：trainer 从 dataloader 取一个 prompt batch，加 uid，按 rollout.n 重复后交给 rollout （ray_trainer.py:1478-1491）。rollout 输出 responses、response_mask、rollout_log_probs； 默认路径下 agent loop 还会把 reward 写进 rm_scores（agent_loop.py:122-149）。
2. 对齐：原始 prompt batch 也按 n 重复，然后与 rollout 输出 union （ray_trainer.py:1540-1541）。如果 batch 里没有 response_mask，就从 attention_mask 末尾切一段补上 （:1543-1544）。
3. 打分：若 use_rm 为真且没有 rm_scores，先调用 reward model；随后 extract_reward 取出 reward_tensor 和额外信息（ray_trainer.py:1561-1568）。
4. 重算 old_log_probs：actor 对当前 batch 前向，返回 old_log_probs 和 entropys； entropy 指标先算，再从 batch 里 pop 掉（ray_trainer.py:1585-1601）。
5. 参考 log prob：只有 use_reference_policy 为真时才计算。它由 use_kl_in_reward 或 use_kl_loss 触发（verl/trainer/ppo/utils.py:75-79），结果 union 进 batch （ray_trainer.py:1618-1622）。
6. values：只有 use_critic 为真时，critic 前向返回 values 并 union 进 batch （ray_trainer.py:1625-1628）。默认 adv_estimator=gae，所以 critic 默认开启 （verl/trainer/ppo/utils.py:96-101）。
7. 优势：先把 reward_tensor 写成 token_level_scores，把 reward 额外信息写进 non_tensor （ray_trainer.py:1633-1636）；如果开了 KL in reward，就减 β·KL(old‖ref)，否则 token_level_rewards = token_level_scores（:1639-1645）。最后调用 compute_advantage （:1667-1675）。
8. 更新与结算：critic 更新（:1677-1681）、actor 更新（:1689-1690）， 然后在尾部收集 data / timing / throughput / variance / spec 指标并 logger.log（:1764-1802）。

顺序上有两个容易看错的地方。第一，reward 的计算发生在 old_log_prob 之前，但 token_level_scores 是在优势阶段才写进 batch （ray_trainer.py:1561-1568 对 :1633）。第二，entropy 是 old_log_prob 阶段算完就 pop 掉的临时字段 （:1601），后续的 actor 更新不会从 batch 里再读到它。

KL 的位置也要分清。KL in reward 用的是 old_log_probs 和 ref_log_prob （ray_trainer.py:100-102），在优势之前执行；KL in loss 则是在 actor 的 mini-batch 内用当前 log_prob 和 ref_log_prob 计算（verl/workers/utils/losses.py:172-182）。两者默认都关闭 （ppo_trainer.yaml:102、actor.yaml:103）。

读完这节你应该知道：reward、response_length、clipfrac、KL、grad_norm、MFU、throughput 分别怎么算，以及 validation 的 @N 指标为什么不能和训练指标混着读。

指标分两批进来。第一批是各阶段就地写入 metrics dict 的标量，例如 old_log_prob 阶段的 actor/entropy（ray_trainer.py:1596-1600）、KL 阶段的 actor/reward_kl_penalty（:1643）、critic 和 actor 的返回值（:1680、:1718）。 第二批是尾部对整个 batch 的统计：compute_data_metrics（:1772）、 compute_timing_metrics（:1783）、compute_throughout_metrics（:1786）。

表 3 · 常用指标 key 与分母（页面整理；出处见各行）

下面把其中三个 key 缩成两条序列的模拟：response_length/mean、response_length/clip_ratio 和 perf/throughput。P 是 prompt 长度，L 是最大 response 长度，r1 / r2 是两条序列的有效 response 长度， t 是 step 时间，g 是 GPU 数。总 token 数按 2×P + r1 + r2 计算。

同名 std：GRPO 用样本标准差，validation 用总体标准差

GRPO 的 advantage 归一化在 core_algos.py 里用 torch.std(scores_tensor) （verl/trainer/ppo/core_algos.py:321）。PyTorch 的 torch.std 默认做 Bessel 校正，分母是 N − 1；当组内只有一个样本时，代码直接把 std 设为 1.0（:315-317）。

validation 的 std@N 走的是另一条路：process_validation_metrics 里 np_std = np.std （verl/trainer/ppo/metric_utils.py:974），对每个 uid 的 N 个样本，仅在 N>1 时才计算 float(np_std(var_vals))（:1019-1020）。numpy.std 默认不做 Bessel 校正，分母是 N。 GRPO 归一化用样本标准差（torch.std，ddof=1）；validation 的 std@N（仅 N>1 时计算；np.std，ddof=0）不是同一个估计量，直接比较会得到不可解释的差。

validation 按 uid 聚合，再按 data_source 分组

_validate 先把每个验证样本按 val_kwargs.n 重复（ray_trainer.py:617-619）， 收集 uid（:670）和 data_source（:691）。 process_validation_metrics 按 (data_source, uid) 分组（metric_utils.py:948-953）， 对每个 uid 的 N 个样本先算 mean@N（:1017）；仅当 N>1 时再算 std@N（:1019-1020）。 N>1 时，用 bootstrap 算 best@N / worst@N（:1027-1040）；如果样本里有 pred， 还算 maj@N（:1042-1057）。bootstrap 次数 1000（:976），只在这个 N>1 分支执行； 默认 val_kwargs.n=1（rollout.yaml:175）时，只有 mean@1，没有 std / best / worst / maj 这类 @N 指标。 最后对 uid 取平均（:1062-1073）。

_val_metrics_update 把 acc 优先当作 core variable，没有 acc 才用 reward （ray_trainer.py:724-725）；只有 mean/maj/best 的 @n_max 结果会进 val-core， 其余进 val-aux（:729-738）。这决定了日志里哪些 validation 指标会被高亮。

reduce_metrics：先按 Metric 的显式 aggregation，再回退到 key 子串

actor 和 critic 的 worker 会返回一组 micro-batch 指标列表。worker 先对包装成 Metric 的列表调用 Metric.aggregate_dp，按对象自带的显式 aggregation（MEAN / SUM / MIN / MAX）在 DP 之间聚合 （verl/workers/engine_workers.py:358-368；aggregate_dp 定义在 verl/utils/metric/utils.py:139-156）。trainer 侧的 reduce_metrics 先判断值是不是 Metric： 是就调用 val.aggregate()，按显式 aggregation 聚合（verl/utils/metric/utils.py:49-51、:123-136）； 其余未包装成 Metric 的值才按 key 子串回退——含 max 用 np.max，含 min 用 np.min， 否则用 np.mean（:52-57）。所以给指标改名时，要先区分它是否已经是 Metric；纯列表形式的指标才依赖名字里的子串。

读完这节你应该知道：哪些行为是 DataProto 的保证，哪些是默认路径的假设，以及换 reward / 换 trainer 时最先要检查什么。

• 全量 batch 在 driver 上汇合：ref_log_prob 和 old_log_prob 的 worker 输出在 engine 侧 .cpu() 后返回（verl/workers/engine_workers.py:755、:763），values 则在 driver 侧 由 no_padding_2_padding 重建（ray_trainer.py:1252-1262）；各阶段结果逐次 union 进同一个 DataProto（ray_trainer.py:1541、:1610、:1622、:1628）。 序列越长、batch 越大，driver 侧的 CPU 内存和 union 次数越值得关注；本页不讨论流式分页。
• padding 的重复样本只保证形状：pad_dataproto_to_divisor 用重复样本来凑整除 （verl/protocol.py:85-99），collect 后才把尾部切掉（verl/single_controller/ray/base.py:58-63）。 worker 内部如果先算 per-sample 指标，重复样本会参与统计，再把结果平均到 driver。 推断（依据：重复发生在 dispatch 前，截断发生在 collect 后）：padding 比例高时，worker 侧指标会偏向被重复的样本。
• response_mask 依赖 padding 约定：trainer 直接取 attention_mask[:, -R:] （ray_trainer.py:132-135）。推断（依据：rollout 输出按左 prompt、右 response 组织， 且 response 右填充，见 verl/workers/utils/padding.py:141-144）：这个切片的前提是 prompt 在左、response 在右、response 右填充； 如果自定义 collate 改了这个约定，mask 不会报错，只会静默错位，loss 和所有 response_length 指标会一起偏。
• union 不覆盖冲突 key：同名 key 值不等时直接断言失败（verl/protocol.py:109-122、:188-199）。 这是保护，也是限制：想用同名 key 覆盖旧值，必须先 pop 再并，不能指望 union 做覆盖写。 另外 union 是原地修改（:792-795），持有旧引用的代码会看到新字段。
• 默认 reward 是 outcome 级：一条序列只有一个标量，L_i>0 时写在最后一个有效 token 上（base.py:76-81）； L_i=0 时索引 -1 会写进末列（padding），不是最后一个有效 token。如果要做 token-level reward、process reward 或逐 turn reward， 需要自定义 reward manager / reward loop：rm_scores 是 (B, R) 张量，形状本身可承载多个非零位置，但默认 reward manager 只写一个位置（base.py:79）。
• 默认 KL 关闭：use_kl_in_reward=False、use_kl_loss=false （ppo_trainer.yaml:102、actor.yaml:103），所以默认不会启用 reference policy （verl/trainer/ppo/utils.py:75-79），也不会产生 actor/reward_kl_penalty 或 kl_loss。 如果日志里找不到这些 key，先检查这两个开关，而不是怀疑指标收集。
• 本页对应的是 legacy trainer：verl/trainer/ppo/ray_trainer.py 在本 commit 标了 @deprecated("... Please use `trainer.use_v1=True` instead.")（:285）。 v1 trainer 的字段名、reward loop 和 metrics 路径可能不同；本页的源码锚点都指向这个 commit 的 legacy 文件。
• 指标名是接口：reduce_metrics 先按 Metric 的显式 aggregation 聚合，未包装成 Metric 的值才按 key 子串选择 max/min/mean（verl/utils/metric/utils.py:49-57）， validation 的 @N 后缀决定它是 val-core 还是 val-aux （ray_trainer.py:729-738）。改 key 名会同时改聚合方式和日志分组，不能当成纯改名。