异步 GRPO 训练器

概述

AsyncGRPOTrainer 实现了相同的 GRPO 算法，但将 rollout 生成与训练解耦。后台工作进程持续从 vLLM 服务器流式获取补全结果，同时训练循环消费这些结果，从而使生成与梯度更新重叠进行，而非交替执行。

此训练器由 Amine Dirhoussi 和 Quentin Gallouédec 贡献。

快速入门

本示例演示了如何使用异步 GRPO 方法训练模型。我们使用 DeepMath-103K 数据集 中的提示词来训练 Qwen3.5 2B 模型。您可以在这里查看该数据集的数据：

以下是用于训练模型的脚本。Trainer 需要 vLLM 和 kernels，而 accuracy_reward 需要 math-verify，您可以通过 pip install trl[vllm,kernels,math_verify] 进行安装。

# train_async_grpo.py
from datasets import load_dataset
from trl import AsyncGRPOTrainer
from trl.rewards import accuracy_reward

if __name__ == "__main__":  # the rollout worker is a spawned process that re-imports this script
    dataset = load_dataset("trl-lib/DeepMath-103K", split="train")

    trainer = AsyncGRPOTrainer(
        model="Qwen/Qwen3.5-2B",
        reward_funcs=accuracy_reward,
        train_dataset=dataset,
    )
    trainer.train()

生成任务在 vLLM 服务器上运行，使用的 GPU 与训练器不同。请在 GPU 0 上启动它：

CUDA_VISIBLE_DEVICES=0 VLLM_SERVER_DEV_MODE=1 vllm serve Qwen/Qwen3.5-2B \
    --weight-transfer-config '{"backend": "nccl"}' \
    --logprobs-mode processed_logprobs \
    --api-server-count 8 \
    --max-model-len 4096

然后使用以下命令在 GPU 1 上执行该脚本：

CUDA_VISIBLE_DEVICES=1 python train_async_grpo.py

兼容性与约束

当前 AsyncGRPOTrainer 实现中，部分参数组合被有意限制：

• 生成始终在独立的 vLLM 服务器上运行，该服务器通过 vllm serve 启动。不存在共置模式。
• 数据集中的提示必须采用对话格式。如果使用environment_factory，数据集可以没有prompt列，由reset()提供该列。
• 奖励函数必须是 Python 可调用对象。不支持以模型 ID 或 PreTrainedModel 形式提供的奖励模型，请参阅 使用自定义奖励函数。
• Trainer 需要 transformers>=5.2.0。
• 分片训练使用 FSDP2。不支持 DeepSpeed，也不支持上下文并行和序列并行（parallelism_config 中的 cp_size > 1 或 sp_size > 1）。
• use_liger_kernel=True 不受支持。
• 针对 lm_head 的 PEFT 适配器不受支持。
• 不支持带有注意力汇聚（attention sinks）的模型。唯一支持的非注意力层是混合模型（Qwen3.5、Qwen3.6、Qwen3-Next 等）中的门控 DeltaNet 层，这需要 transformers>=5.15.0。
• 不支持图像，请参阅 训练视觉语言模型。

训练会通过 kernels 库从 Hub 加载其注意力内核：在 AMD GPU（ROCm）上使用 kernels-community/aiter-flash-attn，在 NVIDIA Blackwell GPU 上使用 kernels-community/flash-attn4，在其他 NVIDIA GPU 上使用 kernels-community/flash-attn3。对注意力 logits 设置上限（attn_logit_softcapping）的模型（例如 Gemma 2）不受 AMD GPU 或 NVIDIA Blackwell GPU 支持。

深入探讨异步 GRPO 方法

在 GRPOTrainer 中，生成和训练是顺序执行的：生成一个批次，计算损失，更新权重，然后重复。即使在 colocate 模式 下，一个阶段也必须完成后另一个阶段才能开始。在 AsyncGRPOTrainer 中，rollout 工作进程在后台进程中生成并评估补全，同时训练循环计算损失并更新模型。要理解其工作原理，可以将其分解为三个主要步骤：生成补全、计算优势 和 计算损失。由于没有参考模型，因此没有 KL 散度项（GRPO 的默认值为 beta=0.0）。

生成补全

Rollout 工作进程最多同时向 vLLM 服务器发送 max_concurrent_rollouts 个请求。对于每个提示，它会生成一组 $G$ 个补全（num_generations) ），待整组完成后对其进行评分并计算优势值。生成的训练样本会被放入队列，供训练循环从中拉取。仅当生成进度领先训练 max_staleness 个步骤时，生成才会等待（参见 并发), ），而训练步骤仅在队列为空时才会等待。

每隔 weight_sync_steps 个训练步骤，trainer 会通过 NCCL 将其更新后的权重发送到 vLLM 服务器，因此后续的 completion 将来自最新的 policy。因此，某个 completion 可能来自比当前正在训练的 policy 旧几个步骤的版本：当 trainer 拉取某个样本时，如果该样本落后于当前权重更新超过 max_staleness 次，则会被丢弃。

指标和本页其余部分使用以下术语来描述补全所经历的各个阶段：

• a 组 是一个提示及其 $G$ 个 rollout。优势值在其中计算。
• rollout 是为某个提示词生成的一次完整对话，包含所有工具调用。它由一个或多个 turn 组成，每个 turn 对应一次 vLLM 请求。
• sample 是用于计算损失的单元：即一次 rollout 的 token 及其对应的 advantage。多轮 rollout 可能会产生多个 sample（参见 Multi-turn rollouts）。
• 行是指一个进程在一个微批次中前向传播的内容：多个样本打包在一起（参见树打包）。一个微批次包含每个进程的一行，gradient_accumulation_steps个微批次构成一个步，即优化器步。

Token 数量分为三种：generated token 是模型生成的 token，forwarded token 是样本中的所有 token（包括提示词和工具结果），trained token 是用于计算损失的 token。

计算优势

对于每个 $G$ 组中的补全，奖励是各个奖励函数奖励的总和。优势值在组内进行归一化，与 GRPO 相同： $$\hat{A}_{i,t} = \frac{r_i - \text{mean}(\mathbf{r})}{\text{std}(\mathbf{r})}$$

没有 scale_rewards 选项：优势值始终由该组的标准差进行缩放。对于所有奖励函数均返回 None 的补全，将不计入该组的均值和标准差，其优势值设为零。

奖励全部相等的组会被丢弃：在默认优势函数下，其值处处为零，因此对其进行训练只会产生一次前向和反向传播，却不会带来任何参数更新。此类组所占的比例会记录为 frac_reward_zero_std。训练步骤由保留下来的组填充，因此被丢弃的组越多，每个步骤所需的生成量就越大。被丢弃的组仍会计入 epoch，而 epoch 始终表示对提示词的一次完整遍历。

计算损失

vLLM 服务器的对数概率值永远不会与训练器完全一致：不同的内核、不同的精度，以及滞后几个步骤的策略（参见 处理训练-推理不匹配问题). 训练器将采样器的对数概率值作为旧策略，因此单个重要性比率同时涵盖了不匹配和滞后问题： $$r_{i,t}(\theta) = \frac{\pi_\theta(o_{i,t} \mid q, o_{i,< t})}{\pi_{\text{sampler}}(o_{i,t} \mid q, o_{i,< t})}.$$

比例超出 [vllm_importance_sampling_clip_min, vllm_importance_sampling_clip_max] 范围的 Token 将从损失中剔除。有关更多信息，请参阅 Masked Importance Sampling。

默认情况下（score_centering=True), 训练器使用 Score Centering. 仅靠重要性权重无法消除这种不匹配：采样 token 的得分 $\nabla_\theta \log \pi_\theta$ 在采样器 $\mu$, 下并不平均为零，因此每次更新都会将策略拉向采样器。Score centering 在每个位置减去该期望值： $$\nabla_\theta \mathcal{L} = - \mathbb{E}_{o \sim \mu} \left[ \hat{A} \left( f(r_t) \nabla_\theta \log \pi_\theta(o_t) - \mathbb{E}_{v \sim \mu} \left[ f(r_v) \nabla_\theta \log \pi_\theta(v) \right] \right) \right], \quad r_v = \frac{\pi_\theta(v)}{\mu(v)},$$

其中 $f(r) = r$ 在 $[C_{\min}, C_{\max}]$ 内，且在 $0$ 外。内部期望是对采样器在每个位置最可能的 num_sampler_logprobs 个 token 取的，这些 token 由 rollout worker 向服务器请求。

当 score_centering=False, 时，训练器改用 GRPO 的截断代理目标，并使用相同的掩码比率： $$\mathcal{L}(\theta) = - \frac{1}{\sum_{i} |o_i|} \sum_{i} \sum_{t=1}^{|o_i|} \mathbb{1}\left[ C_{\min} \le r_{i,t}(\theta) \le C_{\max} \right] \min \left( r_{i,t}(\theta) \hat{A}_{i,t}, \, \text{clip}\left( r_{i,t}(\theta), 1 - \epsilon_{\text{low}}, 1 + \epsilon_{\text{high}} \right) \hat{A}_{i,t} \right),$$

其中 $\epsilon_{\text{low}}$ 是 epsilon ， $\epsilon_{\text{high}}$ 是 epsilon_high.

在这两种情况下，损失值都会根据所有进程中微批次内已训练 token 的数量进行归一化，并在一个步骤的 gradient_accumulation_steps 个微批次上取平均值。

记录的指标

在训练过程中，我们记录以下指标：

• num_tokens：迄今为止转发的 token 总数，包括 prompt 和 completion。在使用 tree packing 时，多个样本共享的前缀只计算一次。
• completions/mean_length：生成补全的平均长度。使用工具时，仅统计非工具 token。
• completions/min_length：生成补全的最小长度。使用工具时，仅统计非工具 token。
• completions/max_length：生成补全的最大长度。使用工具时，仅计算非工具 token。
• completions/mean_terminated_length：以 EOS 终止的生成补全的平均长度。使用工具时，仅统计非工具 token。
• completions/min_terminated_length：以 EOS 终止的生成补全的最小长度。使用工具时，仅统计非工具 token。
• completions/max_terminated_length：以 EOS 终止的生成补全的最大长度。使用工具时，仅统计非工具 token。
• completions/clipped_ratio：截断（clipped）补全的比例，即最后一轮未以 EOS 标记结尾的补全所占的比例。
• completions/duplicate_frac：组内逐 token 重复之前 rollout 的 rollout 所占比例。参见 饱和与重复。
• tools/call_frequency：所有 rollout（包括随后被丢弃的组）中，每次补全的平均工具调用次数。仅在训练器配置了工具时记录。
• tools/failure_frequency：工具调用失败的比例（工具未找到或抛出异常），分母为所有工具调用，包括随后被丢弃的组。在没有工具调用的日志窗口中，该值为 NaN，而 GRPO 会记录 0.0。仅在训练器具有工具时记录。
• rewards/{reward_func_name}/mean：特定奖励函数在所有其评分的完成项上的平均奖励，包括随后被丢弃的组。当环境通过 get_reward 拥有奖励时，{reward_func_name} 为该环境的类名。
• rewards/{reward_func_name}/group_std：特定奖励函数的奖励标准差，在每个组内计算，并对训练样本取平均值。
• reward：在跨函数求和奖励后，对所有已评分的补全（包括随后被丢弃的组）计算得出的总体平均奖励。这是策略的奖励，因此是需要跟踪的指标：reward_kept 会排除奖励值全部相等的组。
• reward_group_std：各函数求和奖励的标准差，在每组内计算，并对训练样本取平均。优势值会除以该值，这与 GRPO 的 reward_std 不同，后者是在整个批次上计算的。
• reward_kept：训练样本的平均奖励，不包括被丢弃的组。
• reward_dropped：被丢弃的组的平均奖励，即它们停滞的奖励值。请参阅 饱和与重复，结合 frac_reward_zero_std 和 reward 来解读该指标。
• frac_reward_zero_std：奖励标准差为零（所有回答均正确或均错误）或可评分补全少于两个的组所占的比例。这些组将被丢弃。
• sampling/sampling_logp_difference/mean：采样器（vLLM）返回的对数概率与训练模型重新计算的对数概率之间，在已训练 token 上的平均绝对差值。
• sampling/sampling_logp_difference/max：此类绝对差值的最大值。
• sampling/importance_sampling_ratio/min: 最小的重要性采样比率 $r_{i,t}(\theta)$ ，即模型与采样器之间在训练 token 上的比率，且在丢弃 [vllm_importance_sampling_clip_min, vllm_importance_sampling_clip_max] 之外的 token 之前。
• sampling/importance_sampling_ratio/mean：此类重要性采样比率的平均值。
• sampling/importance_sampling_ratio/max：最大的重要性采样比率。
• sampling/masked_mean：由于重要性采样比率落在 [vllm_importance_sampling_clip_min, vllm_importance_sampling_clip_max] 之外而从损失中丢弃的已训练 token 的比例。
• sampling/sampler_mass：采样器最可能的 num_sampler_logprobs 个 token 所覆盖的平均概率质量。仅在 score_centering=True 时记录。
• entropy：训练 token 上 token 预测的平均熵。
• aux_loss：混合专家（Mixture-of-Experts）模型的负载均衡辅助损失，在乘以 router_aux_loss_coef 并加到总损失之前。仅当模型为 MoE 模型且 router_aux_loss_coef 非零时记录。
• clip_ratio/region_mean: GRPO 目标被裁剪以保持在信任区域内的 token 概率比率： $\text{clip}\left( r_{i,t}(\theta), 1 - \epsilon_\mathrm{low}, 1 + \epsilon_\mathrm{high} \right)$. 当 score_centering=True 时，损失不会进行裁剪，因此 clip_ratio/* 指标统计的是信任区域之外的 token 数量。
• clip_ratio/low_mean: 信任区域下界被截断的 token 概率的平均比率： $r_{i,t}(\theta) < 1 - \epsilon_\mathrm{low}$.
• clip_ratio/low_min: 信任区域下界处被裁剪的每个 completion 的最小 token 比例： $r_{i,t}(\theta) < 1 - \epsilon_\mathrm{low}$.
• clip_ratio/high_mean: 信任区域上界被裁剪的 token 概率的平均比率： $r_{i,t}(\theta) > 1 + \epsilon_\mathrm{high}$.
• clip_ratio/high_max: 信任区域上界处被裁剪的每个 completion 中 token 的最大比例： $r_{i,t}(\theta) > 1 + \epsilon_\mathrm{high}$.

其他指标用于描述异步流水线。

指标在每个日志窗口记录一次：记录自上次日志以来的所有内容，涵盖最近logging_steps个优化器步骤。在该窗口内：

• 像 reward、rollout/turns_mean 和 tools/latency_s 这样按 rollout、样本或调用测量的均值，会在窗口内的所有 rollout、样本或调用上进行平均。像 rollout/generated_tokens_per_s 这样的速率，则是总 token 数除以总秒数。
• 否则，名称最后一部分包含单词 count 的指标会被求和：sample/dropped_stale_count 表示自上次日志以来被丢弃的样本数量，而非自训练开始以来的数量。
• 名称的最后一部分包含单词 max 或 min（例如 sample/staleness_max 或 completions/max_length）时，将保留最大值或最小值。
• 其他任何指标均为窗口内记录值的平均值。

单位：

• _s：秒。
• tokens_per_s：每秒生成的 token 数。
• _frac：0 到 1 之间的分数。
• _per_step：单个优化器步长中的数量，在窗口内的步长上取平均值。

Rollout 指标

• rollout/duration_s：单次 rollout 的挂钟时间，从分发开始到 rollout 循环收集完成为止。
• rollout/generated_tokens_per_s：自上次报告以来，以每秒生成的 token 数衡量的生成吞吐量。
• rollout/concurrency：发送到 vLLM 服务器的并发 rollout 数量。
• rollout/concurrency_cap：当前正在进行的 rollout 数量上限，该值会随服务器负载变化。请参阅 并发。
• rollout/turns_mean、rollout/turns_max：每次 rollout 的平均轮数和最大轮数。
• rollout/samples_per_rollout：每次 rollout 的训练样本数量。高于 1.0 时，部分 rollout 被分叉；低于 1.0 时，部分 rollout 未产生任何训练 token。参见 多轮 rollout。
• rollout/fork_frac：提示词未以保留 token 开头的轮次转换所占的比例，即该轮次开启了一个新的样本。
• rollout/drift_tokens_mean、rollout/drift_tokens_max：分叉重新渲染的保留 token 数量，按轮次转换取平均值（未发生分叉的轮次计为 0）。新样本共享分叉点之前的 token，因此该值表示每个分叉为前向传播新增的 token 数。
• rollout/score_queue_size：等待评分的已完成组数。
• rollout/queue_size：在 trainer 拉取样本的位置，等待打分的 rollout 队列中的样本数量。参见 rollout 队列。
• rollout/score_s：对一组进行评分所需的时间。
• rollout/score_wait_s：已完成的组在评分前等待的时间。
• rollout/score_block_s：由于评分队列已满，生成被阻塞所花费的时间。
• rollout/backpressure_s：由于生成比训练超前了 max_staleness 步，生成在开始 rollout 前等待的时间，取日志窗口内各次等待的平均值。参见 并发。
• rollout/vllm_retry_count：重试的 vLLM 请求数量。否则，性能下降的服务器会表现为无法解释的延迟。
• rollout/vllm_kv_cache_usage_frac：vLLM KV 缓存使用率（取各引擎的最大值），在日志窗口内的多次轮询中取平均值。
• rollout/vllm_running_requests：服务器引擎正在解码的请求数量（即 vLLM 的 num_requests_running，对所有引擎求和）。它与 rollout/concurrency 之间的差值，是位于 API 服务器进程中、正在构建响应的部分。
• rollout/vllm_queued_requests：服务器已接受但尚未调度的请求数量（即 vLLM 的 num_requests_waiting，对所有引擎求和）。
• rollout/vllm_preemption_count：服务器抢占的请求数量（即 vLLM 的 num_preemptions_total，对所有引擎求和）。权重同步会重置服务器缓存，从而抢占所有正在运行的请求，因此该指标并非压力信号。

工具指标

• tools/latency_s、tools/{tool_name}/latency_s：执行工具调用的耗时，包括所有工具的总耗时以及每个工具的单独耗时。
• tools/{tool_name}/call_count、tools/{tool_name}/failure_count：每个工具的调用次数和失败调用次数。
• tools/unknown_name_count：调用不存在工具的次数。与运行后抛出异常的工具不同，这属于策略错误。
• tools/parallel_calls_mean：调用工具的助手消息中工具调用次数的平均值。
• tools/loop_exhausted_frac：在 max_tool_calling_iterations 处被截断且仍在调用工具的 rollout 占比，分母为所有 rollout，包括未调用工具的 rollout 以及随后被丢弃的组。仅在训练器配置了工具时记录。

HarnessAsyncGRPOTrainer 会以相同的名称记录其中大部分内容。有关差异，请参阅 在 harness 上进行训练。

环境指标

• env/reset_s：在 rollout 开始前，环境 reset() 所花费的时间。仅在提供了 environment_factory 时记录。
• env/reward_s：rollout 结束后，在环境的 get_reward() 中花费的时间。仅当环境定义了 get_reward 时才会记录。

示例指标

• sample/forwarded_tokens_mean、sample/forwarded_tokens_max：每个样本中转发 token 的平均数量和最大数量。
• sample/trained_tokens_mean：每个样本的平均训练 token 数。
• sample/time_in_queue_s：样本在被训练前在 rollout 队列中等待的时间。
• sample/staleness_mean、sample/staleness_max：生成样本的策略与训练该样本的策略之间经历的权重更新次数。sampling/importance_sampling_ratio/* 和 sampling/sampling_logp_difference/* 指标将此陈旧度与训练-推理不匹配相结合：在陈旧度为零时，比率偏离 1.0 或对数概率差偏离 0.0 表明存在不匹配，例如与服务器的数据类型（dtype）差异。
• sample/dropped_stale_count：因超过 max_staleness 而被丢弃的样本数量。

批量指标

• batch/forwarded_tokens_per_step、batch/trained_tokens_per_step：所有进程上每个步骤的前向和训练 token 数量。
• batch/samples_per_step、batch/groups_per_step：每个步骤中的样本数和不同组数。它们之间的差异等于每组中的样本数。
• batch/microbatches_per_step：每个步骤中的 micro-batch 数量。
• batch/masked_token_frac：未参与训练的转发 token 所占的比例。
• batch/samples_per_row、batch/row_tokens_mean、batch/row_tokens_max：每行的样本数，以及每行转发 token 的平均数和最大值。
• batch/row_fill_frac：各行使用的 token_budget 的比例。
• batch/row_imbalance：最昂贵行的注意力成本与平均成本之比。拥有最昂贵行的进程会在梯度 all-reduce 时阻塞其他进程；1.0 表示完全均衡。
• batch/packing_ratio：每个前向传播 token 在打包前的 token 数量：打包使前向传播缩减的倍数，当没有共享时为 1.0。
• batch/pad_frac：为使各行长度一致而添加的填充比例。它仅产生传输开销，在前向传播前会被移除。
• batch/dropped_oversize_count：因超出 token_budget 而被丢弃的样本数量。

性能指标

• perf/step_s：一个 step 的墙钟时间，包括优化器、权重同步以及等待样本的时间。
• perf/fwd_bwd_s、perf/fwd_s：分别表示一个步骤中前向和反向传播所花费的时间，以及仅前向传播所花费的时间。
• perf/optimizer_s：优化器步骤所花费的时间。
• perf/rollout_wait_s：训练因 rollout 队列为空而等待的时间。参见 rollout 队列。
• perf/weight_sync_s：权重同步的耗时，分为 perf/weight_sync_pause_s（等待 vLLM 暂停）、perf/weight_sync_barrier_s（等待其他进程）和 perf/weight_sync_transfer_s（发送权重）。
• perf/fwd_bwd_forwarded_tokens_per_s、perf/wall_clock_forwarded_tokens_per_s：每秒转发的 token 数量，分别对应前向与反向传播，或整个训练步骤。
• perf/wall_clock_trained_tokens_per_s：整个步骤中每秒训练的 token 数量。
• perf/fwd_bwd_mfu、perf/wall_clock_mfu：训练设备在正向和反向传播期间，或在整个训练步骤中的模型 FLOPs 利用率。它们不包含 vLLM 服务器。仅当已知训练设备在训练精度下的峰值计算能力，且 FLOPs 估算支持该模型的架构（例如 Llama、Qwen2、Qwen3、Qwen3-MoE 或 Qwen3-VL）时，才会记录这些指标。

自定义

配置 vLLM 服务器

Trainer 通过 vLLM 服务器进行生成，该服务器位于 vllm_server_base_url（默认为 http://localhost:8000）：

VLLM_SERVER_DEV_MODE=1 vllm serve <model_name> \
    --weight-transfer-config '{"backend": "nccl"}' \
    --logprobs-mode processed_logprobs \
    --api-server-count 8 \
    --max-model-len 4096

• VLLM_SERVER_DEV_MODE=1 和 --weight-transfer-config 会暴露训练器用于发送权重的端点。
• --logprobs-mode processed_logprobs 使服务器返回在应用温度和 top-k/top-p 之后所采样的分布的 log-probabilities，这正是重要性比率所比较的对象。
• --api-server-count 8 会保留对评分中心请求而言服务成本较低的 top log-probabilities（参见下方的警告）。
• --max-model-len 是您预期的最大总序列长度（包括 prompt 和 completion）。较低的取值会降低服务器上的 GPU 内存占用，从而为 KV cache 释放更多内存并提高吞吐量。一个好的起点是 prompt 长度加上配置中的 max_completion_length。在使用 tools 时，每一轮都会增加最多 max_completion_length 个 token 以及工具结果，因此请将其设置为预期的最长对话长度。如果某一轮的 prompt 加上 max_completion_length 超过了 --max-model-len，服务器将拒绝该请求，运行会因服务器错误而停止。若使用了 top_k 或 top_p，服务器会在该轮达到 --max-model-len 时停止该轮，而仅拒绝长度本身已达到该值的 prompt。它同时也是 token_budget 的默认值。

请确保服务器使用的 GPU 与训练器不同，否则可能会遇到 NCCL 错误。您可以通过 CUDA_VISIBLE_DEVICES 环境变量来指定要使用的 GPU。

在 vLLM 的 API 服务器中，序列化 top log-probabilities 属于 CPU 工作，而默认的单个 API 服务器在 num_sampler_logprobs=8 设置下会使生成吞吐量减半。在 vLLM 0.32 或更高版本中，请使用 --api-server-count 8 启动服务器：在单张 H100 上对 1.5B 至 7B 模型进行测量，top-8 log-probabilities 的开销占生成吞吐量的比例低于 3%。在 vLLM 0.31 中，相同的标志仍会导致 20% 至 40% 的性能损失，该问题已通过上游的 vllm-project/vllm#58181 修复。

并发

三个限制条件决定了同时运行的 rollout 数量。前两个会自动调整，第三个由您自行设置。

陈旧度。 max_staleness 设置生成过程可以领先训练器多少个训练步骤。由于每个步骤中的样本数量会随着补全变长或变短而变化，工作进程以 token 为单位跟踪此限制。在每个步骤之后，训练器统计其使用的 token 数量，并允许工作进程保留最多 max(max_staleness, 1) 倍该数量的 token 处于排队或进行中状态。在第一个步骤之前，它假设每一行都是满的。工作进程仅在该限制之下才会启动新一组 rollout。在最初阶段，它尚不知道一个 rollout 会产生多少 token，因此每次只启动一组，直到第一组完成为止。如果一个样本最终仍然比 max_staleness 个步骤更陈旧，则会被丢弃。

vLLM 负载。 如果同时向 vLLM 发送过多请求，其速度会变慢：KV cache 会被填满，vLLM 不得不暂停并重做部分请求。因此，worker 还会对同时运行的 rollout 数量保持一个实时上限，记录为 rollout/concurrency_cap，并每 5 秒检查一次服务器的 /metrics 来调整该上限。该上限的初始值为 KV cache 在最大长度下能容纳的请求数（KV cache tokens / max_model_len），以避免服务器在启动时过载。只要 KV cache 使用率保持在 60% 以下且没有等待的请求，上限就会增加 25%，但每完成一个上限数量的 rollout 最多只增加一次。当使用率超过 80% 时，上限会降低 30%：正在运行的请求继续执行，新请求则等待。如果服务器没有 /metrics 端点，则上限为 max_concurrent_rollouts。

资源。 max_concurrent_rollouts（默认值为 512）是同时运行的 rollout 数量上限。这是唯一需要您自行设置的限制，因为 trainer 无法获知您的环境和工具能够承受多大的负载。每个正在运行的 rollout 都会占用一个环境，可以在一个大小为此值的线程池上执行同步工具调用，并且会保持与 vLLM 服务器的连接处于打开状态。请将其设置为这些组件能够承受的值，例如您的环境服务器可接受的会话数：

from trl import AsyncGRPOConfig

training_args = AsyncGRPOConfig(
    ...,
    max_concurrent_rollouts=64,
)

如果设置过低，生成速度会变慢。Rollout 在运行工具或等待轮次之间时也会保留其槽位，而不仅仅是在 vLLM 生成期间，因此调用大量工具的 rollout 需要更多槽位才能保持 vLLM 处于繁忙状态。如果设置远高于默认值，vLLM 服务器可能会开始丢弃连接。

记录的指标显示了哪个限制正在阻碍生成：

• rollout/backpressure_s 大于零：表示陈旧度限制。生成进度领先于训练，当推理速度比训练快时，这种情况是预期内的。
• rollout/concurrency 达到 rollout/concurrency_cap，低于 max_concurrent_rollouts：vLLM 服务器的负载。
• rollout/concurrency 达到 max_concurrent_rollouts 的上限，同时 rollout/vllm_kv_cache_usage_frac 较低且 perf/rollout_wait_s 较高：这表明已触及瓶颈。如果环境和工具允许，请提高 max_concurrent_rollouts。

Rollout 队列

Worker 将已评分的样本推入队列，Trainer 从中拉取。四个指标用于描述该队列：

• rollout/queue_size：等待的样本数量。
• sample/time_in_queue_s：单个样本在被训练前等待了多长时间，即其陈旧度（staleness）中的时间部分。
• perf/rollout_wait_s：由于队列为空，训练等待了多长时间。
• rollout/backpressure_s：由于生成比训练超前了 max_staleness 步，生成所等待的时长。

最后两个是镜像关系。结合队列大小一起阅读，它们可以告诉你哪一侧是瓶颈：

• 队列接近空，perf/rollout_wait_s 较高：生成受限。训练器处于饥饿状态；请查看 rollout/generated_tokens_per_s、rollout/concurrency 和 rollout/score_queue_size。
• 队列中大约持有 max_staleness 个步骤的 token，rollout/backpressure_s 较高：受训练器限制。生成过程被阻塞，sample/staleness_mean 接近其上限。
• 两者均偏高：max_staleness 限制本身过于严格。正在进行的 rollout 在其样本到达队列之前就已填满该限制，导致训练器等待，而生成过程被阻塞。请调高 max_staleness。
• 两者均接近零：平衡。

饱和与重复

策略从一组 rollout 之间的差异中学习：每个优势值都是相对于该组的平均奖励计算的。当一组中的所有 rollout 都获得相同的奖励时，它们的优势值均为零，因此该组会被丢弃（参见 记录的指标），并且不会为策略提供任何学习信号。随着运行趋于饱和，越来越多的组会呈现这种状态，原因可能是所有 rollout 都解决了提示，或者没有一个 rollout 能解决提示。有四个指标可以区分这些情况，每个指标都回答其自身的问题，因此您可以直接从它们的图表中读取：

将前三项放在一起阅读，并对照你的奖励范围：

completions/duplicate_frac 表示在所有轮次中，与同组内先前某次 rollout 逐 token 完全重复的 rollout 所占的比例。由于相同的 rollout 会获得相同的奖励，因此重复项也会推高 frac_reward_zero_std。

这些指标是在 worker 对某个 group 进行评分时记录的，而不是在其样本被用于训练时记录的，因此它们会略微领先于训练过程，并且包含那些因超过 max_staleness 而导致样本随后被丢弃的 group。如果一个 group 中的所有 rollout 都无法被评分（因为每个奖励函数都返回了 None），该 group 将被丢弃并计入 frac_reward_zero_std，但由于它没有奖励，reward_dropped 不会将其包含在内。

树打包

微批次的样本被打包到每个进程的一行中，无需填充。一行是一个前缀树，而非简单的拼接：多个样本共享的 token 只会被前向传播一次，而注意力机制仍确保每个样本只能看到其自身的历史。一个组的所有 rollout 都会重复其提示词，多轮 rollout 的每一轮都会重复迄今为止的对话，因此前向传播的计算量会按打包比率（batch/packing_ratio）缩小。这仅仅是布局上的改变：每个用于训练的 token 都保留其自身的目标、优势值和旧对数概率，且其对数概率与单独前向传播该样本时的结果一致。

在针对同一提示词 p 的两次生成中，其中 a 和 b 为两个补全结果：

1. Rollouts         r0:  p1 p2 p3 a1 a2                10 tokens
                    r1:  p1 p2 p3 b1 b2

2. Trie             S0 [p1 p2] ─┬─ S1 [p3 a1 a2]       p3 is not shared: a position predicts
                                └─ S2 [p3 b1 b2]       one token, and p3 predicts a1 or b1

3. Packed row       index         0  1 |  2  3  4 |  5  6  7
   (depth-first)    input_ids    p1 p2 | p3 a1 a2 | p3 b1 b2    8 tokens: everything but attention
                    position_ids  0  1 |  2  3  4 |  2  3  4    runs on 8 instead of 10
                    segment      S0    | S1       | S2

4. Attention        segment   queries    keys and values: the segment's path in the trie
   (one call)       S0        p1 p2      p1 p2              index 0 1
                    S1        p3 a1 a2   p1 p2 p3 a1 a2     index 0 1 2 3 4
                    S2        p3 b1 b2   p1 p2 p3 b1 b2     index 0 1 5 6 7

5. Loss             r0: a1 a2 predicted at index 2 3   each trained token is predicted at the index
                    r1: b1 b2 predicted at index 5 6   of the token before it in its rollout

一些推论：

• 任何公共前缀都会被共享：同一提示词的不同 rollout，以及跨提示词的公共系统提示词或工具 schema。样本会被分配到已经持有其所属组的行中，因为该行的提示词已经存在。当该行已满时，如果某个进程仍有空行，或者分配不会增加负载最高的行（该行决定了 micro-batch 的耗时），样本就会被分配到另一行；否则，该 micro-batch 将关闭。
• 未共享任何内容即为普通打包：此时每个样本都是其自身的分段，即普通的无填充打包。
• 仅复制 K/V：每个后代片段只收集一次共享前缀，其大小与未打包训练所持有的量相当，在使用分组查询注意力时则更小。
• 支持 Gated DeltaNet 混合模型（Qwen3.5、Qwen3.6、Qwen3-Next 等）：其卷积操作读取每个 token 自身的前驱，而分块 delta 规则则从父段的最终状态开始处理每个分段。

token_budget 限制了每行的 token 数量，共享的 token 只计算一次。默认情况下，它等于服务器的 max_model_len，因此任何样本都不会超过该上限。在没有共享内容时，树打包的运行速度与普通的无填充打包大致相当，且加速比随打包率的增长而提升（在 Qwen3-0.6B 和 1.7B 上，当打包率为 3.4 时，在 H100 上约为 2.8 到 3 倍）。打包率等于共享前缀长度乘以每行的样本数，因此 batch/samples_per_row 是首要指标：当其接近 1 时，几乎没有内容可以共享，此时调节手段是 token_budget。当样本较长时，batch/row_fill_frac 变得重要：一个 10k token 的样本对 32k 预算的填充效果很差（三个能放下，四个永远放不下，因此规划器通常只能放入两个，行的填充率约为 77%），而 1k token 的样本则能几乎完美地填充该预算。

根据吞吐量来评估树打包，即 batch/trained_tokens_per_step / perf/fwd_bwd_s。打包后的微批次包含更多样本，因此 perf/fwd_bwd_s 会上升，而除以前向传播 token 数的 perf/fwd_bwd_mfu 会下降：这两者均不属于性能回退。每行包含更多样本也意味着每步会采样更多 rollout，因此受生成限制的运行会变得更加受限。

批量指标由以下部分组成：一个 step 包含 gradient_accumulation_steps × num_processes 行，因此 batch/samples_per_step 大约是 batch/samples_per_row 的相应倍数，token 同理。由于每步指标是求和，而每行指标是均值，两侧仅在 step 内各 micro-batch 之间的差异范围内保持一致。

任何使行数变长或变少的操作都会破坏共享。一个从不分叉的多轮对话就是一个长样本，rollout/samples_per_rollout 表示一个对话会产生多少个样本。

Top-k 和 top-p 采样

Top-k 和 top-p 是在受限的候选集上，而非整个词表上，对采样分布进行重新归一化。服务器的对数概率反映了这一点（--logprobs-mode processed_logprobs），因此训练器必须在相同的集合上进行归一化，否则每个重要性比率都会产生偏差：某个 token 在整个词表下的概率低于其在被采样自的候选集下的概率（DeepSeek-V3.2，第 3.1 节）。当 top_p < 1 或 top_k > 0 时，worker 会向服务器请求每个采样 token 的候选集，训练器则在该集合上对其对数概率进行重新归一化。这需要额外的两个服务器标志，并且必须设置 top_k，因为 vLLM 使用它来限制候选集的大小：

VLLM_SERVER_DEV_MODE=1 vllm serve <model_name> \
    --weight-transfer-config '{"backend": "nccl"}' \
    --logprobs-mode processed_logprobs \
    --api-server-count 8 \
    --max-model-len 4096 \
    --return-sampling-mask \
    --enable-scale-out

from trl import AsyncGRPOConfig

training_args = AsyncGRPOConfig(
    ...,
    top_p=0.95,
    top_k=512,
)

--return-sampling-mask 使 vLLM 记录候选集（vLLM 0.28 或更高版本，在其 V2 模型运行器上：如果服务器拒绝该标志，则设置 VLLM_USE_V2_MODEL_RUNNER=1），而 --enable-scale-out 暴露了返回该集合的 token-in-token-out 端点。工作进程在启动时会在服务器上检查这两项。旧版 vLLM 无法返回该集合，因此训练器会继续在整个词汇表上进行归一化，并针对有偏的比率发出警告。

使用 PEFT 训练适配器

我们支持与 🤗 PEFT 库的紧密集成，让您能够训练适配器并将其共享到 Hub，而无需训练整个模型。

from datasets import load_dataset
from peft import LoraConfig
from trl import AsyncGRPOTrainer
from trl.rewards import accuracy_reward

if __name__ == "__main__":
    dataset = load_dataset("trl-lib/DeepMath-103K", split="train")

    trainer = AsyncGRPOTrainer(
        model="Qwen/Qwen3.5-2B",
        reward_funcs=accuracy_reward,
        train_dataset=dataset,
        peft_config=LoraConfig(r=32, lora_alpha=64, target_modules="all-linear"),
    )
    trainer.train()

每次权重同步都需要将更新后的策略发送到服务器。默认情况下，训练器会将适配器合并到基础模型中，并发送完整的合并权重，对于 7B 模型，每次同步大约需要 15 GB。使用 --enable-lora 启动 vLLM 服务器后，它只会发送适配器，在 r=32 时约为 160 MB：

VLLM_SERVER_DEV_MODE=1 VLLM_ALLOW_RUNTIME_LORA_UPDATING=1 vllm serve <model_name> \
    --weight-transfer-config '{"backend": "nccl"}' \
    --logprobs-mode processed_logprobs \
    --api-server-count 8 \
    --max-model-len 4096 \
    --enable-lora \
    --max-lora-rank 32 \
    --max-loras 6

VLLM_ALLOW_RUNTIME_LORA_UPDATING=1 会暴露训练器用于提交每个新适配器的端点。即使使用了 --enable-lora，也要保留 --weight-transfer-config：训练器只在启动时选择同步模式，而那时服务器已经启动，合并同步是回退方案。

--max-lora-rank 必须是 1, 8, 16, 32, 64, 128, 256, 320, 512 之一。它设置的是服务器能够支持的最高秩，而非实际服务的秩，因此 r=4 的适配器在 8 下可以正常工作。--max-loras 必须至少为 max_staleness + 2：训练器会注册 max_staleness + 1 个适配器版本，以便在旧策略下开始的 rollout 能够在该策略下完成，而不是在生成中途切换策略；并且每次同步都会在卸载最旧版本之前加载下一个版本。训练器在启动时会检查这两个值，并告知你应使用什么参数重启服务器。

vLLM 从磁盘而非网络加载适配器，因此训练器会将每个版本写入 <output_dir>/.vllm_lora/ 下，并将该路径提供给服务器。如果服务器运行在另一台机器上，该目录必须位于两台机器都能访问的文件系统上。

<output_dir>/.vllm_lora/ 是一个服务缓存，并非存放训练后适配器的位置。它仅保留最近版本的适配器，且每个版本在超出过期窗口后不久即被删除。检查点才是保存训练适配器的关键，因此请勿将其关闭：若设置 save_strategy="no"，数小时的训练任务结束后将不会留下任何产物。output_dir 必须可被服务器读取，且其生命周期需长于任务本身，因此节点本地临时存储不可行，尽管它能避免在缓慢的共享文件系统上进行同步。

Adapter 同步并不总是更快。与提供合并后的权重相比，vLLM 在提供 LoRA adapter 时，每个 token 的生成速度大约慢 1.33 到 1.39 倍，而在强化学习中，生成通常主导了实际耗时。作为回报，每次同步时的暂停时间会大幅缩短，且同步带宽占用约为 1%。如果生成吞吐量是你的瓶颈，合并同步通常仍是更好的选择。

使用自定义奖励函数

AsyncGRPOTrainer 支持使用自定义奖励函数。为确保兼容性，您的奖励函数必须满足以下要求：

奖励函数可以是同步的 Python 可调用对象，也可以是异步的 async def 协程。当您提供多个奖励函数时，它们会并发执行：异步函数会被一起等待，而同步函数则在线程中运行，因此它们的延迟会相互重叠。

1. 输入参数：

   • 该函数必须接受以下关键字参数：

     • prompts（包含提示词），
     • completions（包含生成的补全内容），
     • completion_ids（包含分词后的补全内容），
     • 数据集可能包含的所有列名（prompt 除外）。例如，如果数据集包含名为 ground_truth 的列，则函数将以 ground_truth 作为关键字参数被调用。

     满足此要求的最简单方法是在函数签名中使用 **kwargs。

   • 数据集采用对话格式，因此prompts和completions是消息字典的列表。

2. 返回值：该函数必须返回一个浮点数列表。每个浮点数代表对应单个补全的奖励。

奖励函数在 rollout worker（一个独立进程）中运行，因此传入其中的所有内容都会被 pickled。每个奖励函数、工具以及 environment_factory（以及它们所闭包捕获的任何内容）都必须可 pickle：请使用模块级函数、functools.partial 或 可调用类实例。Lambda 函数和闭包会在 trainer.train() 时引发 TypeError。这与 GRPOTrainer 不同，在后者中奖励函数在进程内调用，闭包可以正常工作。

Rollout 过程同样在 CUDA_VISIBLE_DEVICES="" 下运行，因此无法使用 GPU。基于 GPU 的奖励模型（例如 AutoModelForSequenceClassification 评分器）加载时不会报错，但会静默回退到 CPU。请保持奖励函数轻量（例如 accuracy_reward 等验证器，或格式、长度检查）。如果需要 GPU 奖励模型，请将其部署在独立的 GPU 上，并通过独立的推理引擎（vLLM、TGI 等）提供服务，然后让一个可 pickle 的奖励函数通过 HTTP 调用它。

示例 1：奖励更长的补全

以下是一个奖励较长补全的奖励函数示例：

def reward_func(completion_ids, **kwargs):
    """Reward function that assigns higher scores to longer completions (in terms of token count)."""
    return [float(len(ids)) for ids in completion_ids]

您可以按以下方式测试：

>>> prompts = [[{"role": "user", "content": "The sky is"}], [{"role": "user", "content": "The sun is"}]]  # not used in the reward function, but the trainer will pass it
>>> completions = [[{"role": "assistant", "content": " blue."}], [{"role": "assistant", "content": " in the sky."}]]  # not used in the reward function, but the trainer will pass it
>>> completion_ids = [[6303, 13], [304, 279, 12884, 13]]
>>> reward_func(prompts=prompts, completions=completions, completion_ids=completion_ids)
[2.0, 4.0]

示例 2：奖励具有特定格式的补全

以下是一个奖励函数的示例，用于检查补全内容是否具有特定格式。该示例受论文 DeepSeek-R1: Incentivizing Reasoning Capability in LLMs via Reinforcement Learning 中使用的 format reward 函数启发。 它专为对话格式设计，其中提示和补全内容均由结构化消息组成。

import re

def format_reward_func(completions, **kwargs):
    """Reward function that checks if the completion has a specific format."""
    pattern = r"^<think>.*?</think><answer>.*?</answer>$"
    completion_contents = [completion[0]["content"] for completion in completions]
    matches = [re.match(pattern, content) for content in completion_contents]
    return [1.0 if match else 0.0 for match in matches]

您可以按以下方式测试此函数：

>>> prompts = [
...     [{"role": "user", "content": "What is the result of (1 + 2) * 4?"}],
...     [{"role": "user", "content": "What is the result of (3 + 1) * 2?"}],
... ]
>>> completions = [
...     [{"role": "assistant", "content": "<think>The sum of 1 and 2 is 3, which we multiply by 4 to get 12.</think><answer>(1 + 2) * 4 = 12</answer>"}],
...     [{"role": "assistant", "content": "The sum of 3 and 1 is 4, which we multiply by 2 to get 8. So (3 + 1) * 2 = 8."}],
... ]
>>> format_reward_func(prompts=prompts, completions=completions)
[1.0, 0.0]

示例 3：基于参考的奖励完成

下面是一个奖励函数的示例，用于检查补全结果是否正确。该示例受论文 DeepSeek-R1: Incentivizing Reasoning Capability in LLMs via Reinforcement Learning 中使用的 accuracy reward 函数启发。数据集中包含一个名为 ground_truth 的列。

import re

def reward_func(completions, ground_truth, **kwargs):
    # Regular expression to capture content inside \boxed{}
    matches = [re.search(r"\\boxed\{(.*?)\}", completion[0]["content"]) for completion in completions]
    contents = [match.group(1) if match else "" for match in matches]
    # Reward 1 if the content is the same as the ground truth, 0 otherwise
    return [1.0 if c == gt else 0.0 for c, gt in zip(contents, ground_truth)]

您可以按以下方式测试此函数：

>>> prompts = [
...     [{"role": "user", "content": "Solve the equation $2x + 3 = 7$."}],
...     [{"role": "user", "content": "Solve the equation $3x - 5 = 10$."}],
... ]
>>> completions = [
...     [{"role": "assistant", "content": r"The solution is \boxed{2}."}],
...     [{"role": "assistant", "content": r"The solution is \boxed{6}."}],
... ]
>>> ground_truth = ["2", "5"]
>>> reward_func(prompts=prompts, completions=completions, ground_truth=ground_truth)
[1.0, 0.0]

示例 4：多任务奖励函数

以下是在 AsyncGRPOTrainer 中使用多个奖励函数的示例。在此示例中，我们定义了两个针对特定任务的奖励函数：math_reward_func 和 coding_reward_func。math_reward_func 根据数学问题的正确性给予奖励，而 coding_reward_func 则根据代码解决方案是否可运行来给予奖励。

from datasets import Dataset
from trl import AsyncGRPOTrainer

# Define a dataset that contains both math and coding problems
dataset = Dataset.from_list(
    [
        {"prompt": [{"role": "user", "content": "What is 2+2?"}], "task": "math"},
        {"prompt": [{"role": "user", "content": "Write a function that returns the sum of two numbers."}], "task": "code"},
        {"prompt": [{"role": "user", "content": "What is 3*4?"}], "task": "math"},
        {"prompt": [{"role": "user", "content": "Write a function that returns the product of two numbers."}], "task": "code"},
    ]
)

# Math-specific reward function
def math_reward_func(prompts, completions, task, **kwargs):
    rewards = []
    for prompt, completion, t in zip(prompts, completions, task):
        if t == "math":
            # Calculate math-specific reward
            correct = check_math_solution(prompt, completion)
            reward = 1.0 if correct else -1.0
            rewards.append(reward)
        else:
            # Return None for non-math tasks
            rewards.append(None)
    return rewards

# Coding-specific reward function
def coding_reward_func(prompts, completions, task, **kwargs):
    rewards = []
    for prompt, completion, t in zip(prompts, completions, task):
        if t == "code":
            # Calculate coding-specific reward
            works = test_code_solution(prompt, completion)
            reward = 1.0 if works else -1.0
            rewards.append(reward)
        else:
            # Return None for non-coding tasks
            rewards.append(None)
    return rewards

# Use both task-specific reward functions
if __name__ == "__main__":
    trainer = AsyncGRPOTrainer(
        model="Qwen/Qwen2.5-0.5B-Instruct",
        reward_funcs=[math_reward_func, coding_reward_func],
        train_dataset=dataset,
    )

    trainer.train()

在此示例中，math_reward_func 和 coding_reward_func 旨在配合包含数学和编程问题的混合数据集使用。数据集中的 task 列用于确定对每个问题应用哪个奖励函数。如果数据集中的某个样本没有相关的奖励函数，该奖励函数将返回 None，AsyncGRPOTrainer 将继续使用有效的函数和任务。这使得 AsyncGRPOTrainer 能够处理具有不同适用性的多个奖励函数。

请注意，AsyncGRPOTrainer 会忽略奖励函数返回的 None 奖励，仅考虑相关函数返回的奖励。这确保了模型在相关任务上进行训练，并忽略没有相关奖励函数的任务。

示例 5：异步奖励函数

自定义奖励函数也可以定义为 async def 协程。如果您的奖励依赖于缓慢的 I/O（例如，调用远程服务），这将非常有用。当您传入多个异步奖励函数时，AsyncGRPOTrainer 会并发执行它们，从而使它们的延迟相互重叠。

以下是一个异步奖励函数的最小示例，用于模拟 I/O 密集型操作：

import asyncio

async def async_reward_func(prompts, completions, **kwargs):
    # Simulate an I/O-bound call (e.g., HTTP request, database lookup)
    await asyncio.sleep(0.01)
    # Simple toy reward: 1.0 if the completion is non-empty, else 0.0
    return [1.0 if completion else 0.0 for completion in completions]

将奖励函数传递给训练器

要使用自定义奖励函数，请将其传递给 AsyncGRPOTrainer，如下所示：

from trl import AsyncGRPOTrainer

trainer = AsyncGRPOTrainer(
    reward_funcs=reward_func,
    ...,
)

您可以将多个奖励函数作为列表传入；该列表可以同时包含同步函数和异步函数：

from trl import AsyncGRPOTrainer

trainer = AsyncGRPOTrainer(
    reward_funcs=[reward_func, async_reward_func1, async_reward_func2],
    ...,
)

奖励将计算为每个函数奖励的总和。

自定义 rollout 循环

生成、奖励、优势计算和过滤均在 AsyncRolloutLoop 中的 rollout 流程里执行。若要修改其中某一步，请子类化该循环，重写对应步骤的方法，并将该子类设置为 trainer 子类的 loop_cls。trainer 的构造函数保持不变。

例如，一种课程策略会停止采样策略已经解决的提示，其阈值位于配置子类中：

from dataclasses import dataclass

import numpy as np

from trl import AsyncGRPOConfig, AsyncGRPOTrainer
from trl.trainer.async_grpo_trainer import AsyncRolloutLoop


@dataclass
class CurriculumConfig(AsyncGRPOConfig):
    pass_rate_cap: float = 0.9


class CurriculumRolloutLoop(AsyncRolloutLoop):
    def filter_prompt(self, row_index, row, epoch):
        return self.pass_rates.get(row_index, 0.0) < self.config.pass_rate_cap

    async def setup(self):
        self.pass_rates = {}

    def compute_advantages(self, group):
        scored = group.rewards[~np.isnan(group.rewards)]
        if len(scored):
            self.pass_rates[group.row_index] = float(np.mean(scored > 0))
        return super().compute_advantages(group)


class CurriculumTrainer(AsyncGRPOTrainer):
    loop_cls = CurriculumRolloutLoop

奖励是对 reward_funcs、定义了 get_reward 的环境，以及自定义 generate_one 在其 Rollout 上设置的 reward 进行求和：若要通过裁判模型评分，请传入异步奖励函数，或在 generate_one 内部进行评分。compute_advantages 会为每个组运行，包括那些因所有已评分奖励相等而被丢弃的组；丢弃操作检查的是奖励，而非优势值。该循环在派生的进程中构建，因此请在模块级别定义子类，并将训练代码放在 if __name__ == "__main__": 之下。请在 setup 中创建客户端，因为它运行在循环的事件循环上，而不是重写 __init__。被 filter_prompt 拒绝的提示词在该 epoch 中会被跳过，并计入该 epoch 的计数，类似于被丢弃的组。如果它拒绝了某个 epoch 中的所有提示词，循环将抛出异常。

从检查点恢复

基础的 Trainer 通过跳过并重放其已训练过的批次来恢复训练，但这不适用于实时 rollout 队列，因此 ignore_data_skip 始终为 True。取而代之的是，尚未训练的第一个 prompt 的索引会随每个 checkpoint 一起保存到 rollout_state.json 中，并在恢复时还原，从而让 worker 快速前进到该 prompt，而无需重放样本。这是 trained 位置，而非生成器的位置：worker 会超前于训练 rollout 队列深度的量，这些缓冲的样本在运行结束时会被丢弃，因此从生成器的位置恢复会跳过那些已生成但从未训练过的 prompt。流式数据集（IterableDataset）无法重新定位，因此其 worker 在恢复时会从第一个 prompt 重新开始。

Agent 训练

异步 GRPO 支持智能体训练：模型在生成过程中调用工具，并根据结果进行学习。

• 工具是暴露给模型的普通 Python 函数（同步或异步）。对于无状态调用（如计算器、网络搜索），请使用 tools。
• 环境是更通用的形式：一个在每次 rollout 时全新构建的有状态对象，其公开方法作为工具暴露，并包含一个 reset 生命周期钩子以及一个可选的 get_reward，使其能够自行管理奖励。当您需要每次 rollout 的独立状态、重置钩子或由环境管理的奖励时，请使用 environment_factory。

它们可以组合使用——你可以在 environment_factory 旁边传入独立的 tools。

工具

tools 参数需要一个 Python 函数（同步或异步）列表，用于定义智能体可用的工具：

from trl import AsyncGRPOTrainer

trainer = AsyncGRPOTrainer(
    tools=[tool1, tool2],
    ...,
)

每个工具必须是一个标准的 Python 函数，具有带类型提示的参数和返回类型，并附带一个Google 风格的 docstring，用于描述其用途、参数和返回值。 有关更多详细信息，请参阅传递工具指南。

模型通过名称调用工具，因此每个工具都以其 __name__ 进行注册，并且必须是一个 模块级函数，这也有助于保持其可被 pickle 序列化，以便供 rollout worker 使用。同步工具运行在大小为 max_concurrent_rollouts 的线程池中，因此一个缓慢的工具不会阻塞其他正在进行的 rollout。在单个 assistant 回合内，工具调用按照模型发出的顺序执行。因此，最多可以有 max_concurrent_rollouts 个同步工具同时运行，所以任何读取或写入共享状态的工具都必须是线程安全的。

工具调用循环要求聊天模板是前缀保持的（追加一条工具消息时，不得改变之前消息的渲染方式）。对于已知的模型系列（例如 Qwen3、DeepSeek-V3），当启用工具时，TRL 会自动替换为经过修补的训练模板。完整的列表请参阅聊天模板。

在 AsyncGRPOConfig 中使用 max_tool_calling_iterations 来限制工具调用轮次的数量。默认情况下没有上限，当模型生成一个不包含工具调用的响应轮次时，生成即停止。

示例：

from trl import AsyncGRPOTrainer

def multiply(a: int, b: int) -> int:
    """
    Multiplies two integers.

    Args:
        a: The first integer.
        b: The second integer.

    Returns:
        The product of the two integers.
    """
    return a * b

async def async_add(a: int, b: int) -> int:
    """
    Asynchronously adds two integers.

    Args:
        a: The first integer.
        b: The second integer.

    Returns:
        The sum of the two integers.
    """
    return a + b

trainer = AsyncGRPOTrainer(
    tools=[multiply, async_add],
    ...,
)

环境

你还可以通过 environment_factory 提供工具。在此模式下，AsyncGRPOTrainer 会为每个 rollout 创建一个环境实例，并将该环境的公开方法作为工具暴露出来。

使用 environment_factory 时，环境拥有数据：不存在外部的 train_dataset。在每次 rollout 中，reset() 会生成任务——从环境持有的语料库中自采样，或以程序化方式生成状态——并返回提示词。max_steps 设置训练长度。以下是一个最小示例，展示了一个自采样目标并暴露 increment 方法（作为工具暴露）的环境。

import random

from trl import AsyncGRPOConfig, AsyncGRPOTrainer

class IncrementEnv:
    # Reserved methods
    def reset(self, **kwargs) -> str | None:  # required; called at the start of each rollout
        self.counter = 0
        self.target = random.randint(1, 6)  # self-sample the task
        return f"Increment the counter by {self.target}."  # returned string becomes the prompt

    def get_reward(self) -> float:  # optional: the environment scores itself from its own state
        return float(self.counter == self.target)

    # Public methods (exposed as tools)
    def increment(self, step: int) -> int:
        """
        Increment the internal counter.

        Args:
            step: Value to add to the counter.

        Returns:
            The updated counter value.
        """
        self.counter += step
        return self.counter

if __name__ == "__main__":
    trainer = AsyncGRPOTrainer(
        model="Qwen/Qwen3-0.6B",
        args=AsyncGRPOConfig(max_steps=1000, chat_template_kwargs={"enable_thinking": False}),
        environment_factory=IncrementEnv,
    )
    trainer.train()

环境类有两个保留方法：reset 和 get_reward。与其他任何公共方法不同，这些方法不会作为工具暴露给模型。

• reset（必需）在 rollout 开始时被调用。它可以返回 None 或一个字符串。如果返回字符串，该字符串即为用户提示词。
• get_reward（可选，同步或异步）不接收任何参数，并返回一个 float：环境根据其自身的内部状态对刚刚运行的回合进行评分（游戏是否以胜利结束？单词是否被猜中？）。它在每个完成的 rollout 中被调用一次，并作为奖励来源。

两者都在工作进程的线程池中运行，因此较慢的环境只会延迟其自身的 rollout，而不会影响其他正在进行的 rollout。它们的耗时分别记录为 env/reset_s 和 env/reward_s。

由于环境在每次 reset() 时进行自采样，GRPO 组中的 G 个成员可能不共享相同的初始状态，这使得组基线略微更具噪声。对于组内相同的状态，请让 reset() 从共享的键中确定性地派生状态。

提供外部数据集（可选）

与其让环境自行采样，您可以提供一个外部的 train_dataset。例如，当您的任务已经存在于数据集中，或者需要在 多个环境 之间进行路由时。在每次 rollout 中，训练器会采样一行，将其 "prompt" 列用作提示，并将该行的各列作为关键字参数传递给 reset()（以便环境从数据集中读取任务）。

from datasets import Dataset
from trl import AsyncGRPOTrainer

# Each row carries the prompt and the task data (here, the counter target to reach).
dataset = Dataset.from_dict(
    {
        "prompt": [[{"role": "user", "content": f"Increment the counter to {i}."}] for i in range(1, 7)],
        "target": list(range(1, 7)),
    }
)

class IncrementEnv:
    def reset(self, target, **kwargs) -> None:  # the row's `target` column arrives as a keyword argument
        self.counter = 0
        self.target = target

    def get_reward(self) -> float:
        return float(self.counter == self.target)

    def increment(self, step: int) -> int:  # exposed as a tool (see the full docstring above)
        self.counter += step
        return self.counter

if __name__ == "__main__":
    trainer = AsyncGRPOTrainer(
        model="Qwen/Qwen3-0.6B",
        train_dataset=dataset,  # the dataset owns the task; the "target" column reaches reset(target=...)
        environment_factory=IncrementEnv,
    )
    trainer.train()

奖励

环境拥有的奖励和训练器拥有的奖励并非互斥。对于从补全结果中自然计算得出的奖励（例如对文本进行格式检查），请传入 reward_funcs，并让环境通过 get_reward 返回其状态相关的奖励。所有来源的奖励均以权重 1 进行求和。

def format_reward(completions, **kwargs):  # trainer-owned: scores the completion text
    ...

trainer = AsyncGRPOTrainer(
    model="Qwen/Qwen3-0.6B",
    args=AsyncGRPOConfig(max_steps=1000),
    reward_funcs=format_reward,          # trainer-owned: scores the completion
    environment_factory=IncrementEnv,    # env-owned: scores its internal state via `get_reward`
)

多个环境

要在单次运行中混合训练多种任务（例如编码任务和游戏任务），请传入一个将环境名称映射到工厂的字典，而不是单个可调用对象。每次 rollout 都会通过数据集中的 environment 字段选择其环境，并且 仅暴露该环境的工具。这可以避免泄露无关工具（例如将游戏的 move 工具暴露给编码示例，反之亦然）。每个环境仍然拥有自己的数据，并从 reset() 返回提示，因此数据集只需要 environment 列来在它们之间进行路由。

import random

from datasets import Dataset
from trl import AsyncGRPOConfig, AsyncGRPOTrainer


class CodingEnv:
    def reset(self, **kwargs) -> str:
        n = random.randint(1, 20)
        return f"Compute the {n}th Fibonacci number."

    def run_code(self, code: str) -> str:  # exposed as a tool only for "coding" examples
        """Run the given Python code and return its stdout.

        Args:
            code: The Python source to execute.

        Returns:
            The captured standard output.
        """
        ...

class GameEnv:
    def reset(self, **kwargs) -> str:
        return "Reach the goal tile."

    def move(self, action: str) -> str:  # exposed as a tool only for "game" examples
        """Apply an action to the game and return the new observation.

        Args:
            action: One of "up", "down", "left", "right".

        Returns:
            The observation after the move.
        """
        ...

# The dataset only routes: each row picks the environment for that rollout
if __name__ == "__main__":
    dataset = Dataset.from_dict({"environment": ["coding", "game"] * 500})

    trainer = AsyncGRPOTrainer(
        model="Qwen/Qwen3-0.6B",
        args=AsyncGRPOConfig(chat_template_kwargs={"enable_thinking": False}),
        train_dataset=dataset,
        reward_funcs=reward_func,
        environment_factory={"coding": CodingEnv, "game": GameEnv},
    )
    trainer.train()

环境实例会在多个步骤之间复用（在回合之间重置），因此昂贵的 __init__ 只需支付一次，而不是每个步骤都支付。environment 字段由训练器使用，并且 不会 转发给 reset。

多轮展开

每一轮都会通过聊天模板渲染整个对话并从中生成。聊天模板并不总是以相同的方式渲染不断增长的对话（例如，某些模板会丢弃之前轮次的推理内容），因此某一轮的提示词 token 可能与当前已持有的 token 不同。在 rollout 结束时，worker 会将各轮次整合为训练样本：如果某一轮的提示词以已持有的 token 开头，则将其追加到同一个样本中；否则，该轮次将开启一个新样本。每个生成的 token 恰好在一个样本中参与训练，因此分支（fork）只会产生前向传播的 token 开销，而不会带来额外的训练信号。

Rollout 中的每个样本都会获得该 rollout 的优势值。rollout/fork_frac 和 rollout/drift_tokens_* 展示了各轮次是如何进行协调的。

环境集成

要训练在自身运行环境中执行的智能体（例如在 Harbor 沙箱中运行的编码智能体），请参阅 OpenEnv 指南中的 在运行环境中训练。

训练视觉语言模型

视觉语言模型（Qwen3.5、Qwen3.6、Qwen3-VL 等）可以在纯文本数据集上进行训练。照常传入模型 ID 即可，其他部分无需更改。

视觉塔已被冻结。纯文本数据集永远不会生成图像标记，因此前向传播过程不会调用该塔。文本模型之外的所有组件（视觉塔、多模态投影器）均设置为 requires_grad=False：它们不会接收梯度，也不占用优化器状态，权重同步过程会跳过它们，因此服务器会保留从检查点加载的原始值。由于该塔仍会被加载，其权重会占用 GPU 内存。

目前尚不支持图像。包含图像的提示词不会被传递到 vLLM 服务器或训练前向传播中。多模态训练还需要无填充打包路径来从图像网格构建 3D M-RoPE 位置，但此处尚未实现。

AsyncGRPOTrainer

>>> from trl import AsyncGRPOTrainer
>>> from trl.rewards import accuracy_reward
>>> from datasets import load_dataset

>>> dataset = load_dataset("trl-lib/DeepMath-103K", split="train")

>>> trainer = AsyncGRPOTrainer(
...     model="Qwen/Qwen2.5-0.5B-Instruct",
...     reward_funcs=accuracy_reward,
...     train_dataset=dataset,
... )
>>> trainer.train()

>>> class CurriculumTrainer(AsyncGRPOTrainer):
...     loop_cls = CurriculumRolloutLoop

AsyncGRPOConfig

AsyncRolloutLoop

Rollout

TurnRecord

RolloutGroup