verl 与 verl-agent 自定义 Gym 环境接入指南

iiiiii11

227人浏览 · 2026-11-05 11:05:11

iiiiii11 · 2026-11-05 11:05:11 发布

核心结论

如果只需要快速把自定义 Gym 风格环境接入 verl，可以继承 AgentLoopBase，在 run() 中实现“LLM 生成动作 → 环境 step() → 读取新观测”的循环，并返回 prompt_ids、response_ids、response_mask。
如果要做完整的多轮 Agent RL 训练，verl-agent 在 verl 之上增加了 agent_system/ 扩展层。自定义环境通常需要按 Worker → Envs 包装器 → EnvironmentManager 三层实现，并在 env_manager.py 的 make_envs 中注册路由。
默认配置下，verl-agent 训练使用的是轨迹级奖励。EpisodeRewardManager 会把整条轨迹的累计奖励写成一个标量，并放在该轨迹最后一个有效响应 token 的位置。
如果使用 GiGPO，则存在一条独立的 step-level 奖励与优势计算通道。step_rewards 会被单独记录，由 step_norm_reward 做分组归一化，再广播为 step-level 优势。EpisodeRewardManager 不负责这部分。

verl & verl-agent

verl 是一个由字节跳动 Seed 团队发起、由 verl 社区维护的大语言模型强化学习训练库。它以 HybridFlow 为学术基础，定位是灵活、高效、可用于生产环境的 RL 训练框架，支持 GRPO、PPO、DPO、REINFORCE++ 等多种算法，并通过 Ray 分布式架构实现 actor 与 rollout 的资源分离，可扩展至 70B 模型和数百 GPU 规模。verl 的核心设计包括可扩展的混合控制器编程模型、与 FSDP/Megatron-LM/vLLM/SGLang 等现有基础设施的模块化集成，以及面向不同集群规模的灵活设备映射。GitHub 地址为：https://github.com/verl-project/verl

verl-agent 是构建在 verl 之上的扩展，专门用于通过强化学习训练 LLM/VLM Agent，同时也是 GiGPO 算法论文的官方代码实现。它的核心设计是 step-independent 多轮 rollout：每一步独立构造输入，而不是把完整交互历史拼接成一个不断增长的 prompt，因此可以支撑长达 50 步甚至更长程的任务，例如 ALFWorld 中的多步操作。verl-agent 在 verl 的 PPO 训练主循环之上，增加了并行的 Gym 风格环境、可插拔记忆模块、轨迹级奖励管理，以及 GiGPO 的 step-level 优势计算。GitHub 地址为：https://github.com/langfengQ/verl-agent。

1. 基础知识：Agent RL、rollout、token mask 与奖励

1.1 Agent RL 中的一次 rollout

在 Agent RL 中，一次 rollout 通常指：环境重置后，模型与环境进行多轮交互，直到环境终止或达到最大步数，产生一条完整轨迹。形式化地，一条轨迹可以写成：

$\tau = (o_0, a_1, r_1, o_1, a_2, r_2, o_2, \ldots, a_T, r_T, o_T)$

其中：

$o_t$ ：第 $t$ 步环境观测；
$a_t$ ：第 $t$ 步模型动作；
$r_t$ ：第 $t$ 步环境奖励；
$T$ ：轨迹步数。

在 verl-agent 中，轨迹采集由 TrajectoryCollector 编排。每一步通常执行：

EnvironmentManager 结合 memory.fetch() 构建当前观测文本；
调用 LLM 生成动作文本；
动作经过投影函数解析为环境合法动作；
调用环境 step()；
调用 memory.store() 更新记忆。

1.2 Token-in-token-out 与 response_mask

verl 的 AgentLoopOutput 需要返回：

prompt_ids：输入 prompt 的 token id；
response_ids：模型生成及可选环境观测的 token id；
response_mask：标记哪些 token 参与训练。

通常：

response_mask = 1 表示该 token 由 LLM 生成，参与策略梯度；
response_mask = 0 表示该 token 来自环境观测等非模型生成内容，不参与或按特定方式处理。

文档强调必须遵循 token-in-token-out（TITO） 原则：不要对已经解码的文本重新编码，否则 token 序列可能与模型实际生成序列不一致。

1.3 轨迹级奖励与单步奖励

轨迹级奖励是对整条轨迹给一个标量，例如：

$R_i = \sum_{t=1}^{T_i} r_{i,t}$

其中 $R_i$ 是第 $i$ 条轨迹的累计奖励， $r_{i,t}$ 是第 $i$ 条轨迹第 $t$ 步奖励。

单步奖励则保留每个 $r_{i,t}$ ，用于判断每一步动作的相对好坏。默认 verl-agent 使用轨迹级奖励；GiGPO 额外使用 step-level 奖励通道。

2. 使用 verl 原生 `AgentLoopBase` 快速接入自定义 Gym 环境

如果只想快速把自定义 Gym 环境接入 verl，最小路径是实现一个 AgentLoopBase 子类。核心骨架如下：

from verl.agent_loop.base import AgentLoopBase, AgentLoopOutput

class MyGymAgentLoop(AgentLoopBase):
    async def run(self, sampling_params: dict, **kwargs) -> AgentLoopOutput:
        env = kwargs.get("env")
        obs, info = env.reset()

        all_prompt_ids = []
        all_response_ids = []
        all_response_mask = []

        done = False
        while not done:
            prompt_messages = self._format_obs_to_prompt(obs, info)
            llm_output = await self.llm_client.generate(
                prompt_messages,
                sampling_params=sampling_params
            )

            action_text = llm_output.text
            action = self._parse_action(action_text)

            next_obs, reward, terminated, truncated, step_info = env.step(action)
            done = terminated or truncated

            all_prompt_ids.extend(llm_output.prompt_ids)
            all_response_ids.extend(llm_output.response_ids)
            all_response_mask.extend([1] * len(llm_output.response_ids))

            obs = next_obs
            info = step_info

        final_reward = float(reward)

        return AgentLoopOutput(
            prompt_ids=all_prompt_ids,
            response_ids=all_response_ids,
            response_mask=all_response_mask,
        )

这段代码中有四个关键点。

第一，_format_obs_to_prompt 负责把环境观测转为 LLM 可理解的文本。例如观测是字典：

{"room": "A", "has_key": False, "door": "locked"}

可以格式化为：

当前房间：A
是否持有钥匙：否
门状态：锁住
可选动作：go_left, go_right, pick_key, open_door

第二，_parse_action 把 LLM 自由文本解析为环境 step() 期望的动作格式。例如离散动作可以解析为整数索引，复杂动作可以解析为 JSON。

第三，response_mask 标记模型生成 token。文档也提到可选地把环境观测 token 加入 response_ids，并用 mask=0 标记。

第四，奖励可以在 run() 内计算，也可以通过 AgentLoopOutput 的额外元数据传递，再在自定义 Reward Loop 或 Reward Function 中读取。verl 允许通过 custom_reward_function.path 和 custom_reward_function.name 指定自定义奖励函数。

编写完成后，需要在 YAML 中注册：

actor_rollout_ref:
  rollout:
    agent:
      agent_loop_config_path: path/to/your_agent_loop.yaml

your_agent_loop.yaml 中指定类路径：

- name: my_gym_agent
  _target_: my_package.MyGymAgentLoop

这条路径适合快速验证环境接口和动作解析逻辑。若要支持多轮 rollout、记忆管理、并行环境、轨迹级奖励聚合和 GiGPO，则建议使用 verl-agent 的三层环境封装。

3. verl-agent 的分层结构与执行链路

3.1 两层架构

verl-agent 可以理解为两层：

层级	职责	关键模块
verl 核心层	分布式训练编排、Worker 管理、模型训练/推理	`verl/trainer/main_ppo.py`、`verl/workers/`、`ray_trainer.py`
verl-agent 扩展层	多轮 rollout、环境管理、记忆系统、轨迹级奖励	`agent_system/`

PPO 训练主循环仍由 verl 的 RayPPOTrainer 驱动，但轨迹采集环节由 verl-agent 的 TrajectoryCollector 接管。

3.2 环境系统三层结构

agent_system/environments/ 下通常包含：

agent_system/environments/
├── env_package/          # 各环境实现
├── prompts/              # Prompt 模板
├── env_manager.py        # make_envs 与 EnvironmentManager
└── base.py               # EnvironmentManagerBase

一个完整环境包通常分三层。

第一层：Worker（Ray Remote Actor）

Worker 真正持有单个环境实例，最接近 gym.env。它以 seed 和 env_kwargs 初始化，对外暴露：

reset(idx/seed) -> (obs, info)
step(action) -> (obs, reward, done, info)

文档以 WebshopWorker 为例：懒加载底层环境以避免 CUDA 初始化问题；在 info 中注入 available_actions；对原始奖励做标准化 shaping。

第二层：Envs 包装器

Envs 包装器不直接与环境交互，而是创建并管理一组 Ray Worker，将批量调用分发到各 Worker 并行执行。它负责 Worker 初始化、seed 分配、批量 reset() / step() 映射。

第三层：EnvironmentManager

EnvironmentManager 继承 EnvironmentManagerBase，负责高层编排：观测构建、记忆管理、动作投影。统一实现三个核心方法：

reset(kwargs) -> (obs_dict, infos)
step(text_actions) -> (obs, rewards, dones, infos)
success_evaluator(...)

obs_dict 通常包含 text、image、anchor 等字段，以支持纯文本和多模态环境。动作投影通过 projection_f 将 LLM 自由文本映射或校验为环境合法动作格式。

prompts/ 下每个环境有同名 Python 文件，定义系统提示和每步观测格式化模板。EnvironmentManager 在 build_text_obs 时调用这些模板。

env_manager.py 中的 make_envs 是环境系统唯一入口。它根据 config.env.env_name 路由，实例化对应 Envs 包装器、Projection 函数和 EnvironmentManager。简化逻辑如下：

if "search" in config.env.env_name.lower():
    _envs = build_search_envs(...)
    projection_f = partial(search_projection)
    envs = SearchEnvironmentManager(_envs, projection_f, config)
elif "alfworld" in config.env.env_name.lower():
    _envs = build_alfworld_envs(...)
    projection_f = partial(alfworld_projection)
    envs = AlfWorldEnvironmentManager(_envs, projection_f, config)

所有环境遵循“一个 Coordinator 管理一组 Ray remote Worker”的两层并行设计。好处包括水平扩展、资源隔离、Group 语义和 Ray 容错恢复。

3.3 多轮 rollout、记忆与奖励

multi_turn_rollout/trajectory_collector.py 中的 TrajectoryCollector 负责编排整个多轮 rollout。它从环境 reset 开始，循环执行“调用 LLM → 执行环境 step → 存储记忆”，直到环境终止，产出完整轨迹。

memory/ 提供可插拔记忆模块，决定每一步送进 LLM 的历史上下文包含哪些内容。SimpleMemory 保存近期观测/动作，按固定策略构建每步输入；SearchMemory 面向 Search 类任务。EnvironmentManager 在每一步通过 memory.fetch() 构建观测，在环境 step 后通过 memory.store() 更新记忆。

reward_manager/ 与 multi_turn_rollout/reward_manager.py 协同，将环境返回的逐步信号或最终信号转化为 RL 训练所需的轨迹级奖励。默认核心类是 EpisodeRewardManager。

一次 PPO 训练迭代中，各模块协作如下：

verl/trainer/main_ppo.py 启动，RayPPOTrainer 驱动训练循环。
进入 rollout 阶段，TrajectoryCollector 接管轨迹采集。
每一步：EnvironmentManager 调用 memory.fetch() 构建当前观测 → 送 LLM → 动作交回环境 step() → memory.store() 更新记忆。
环境终止后，奖励管理模块将轨迹级奖励聚合到 TrajectoryCollector 输出中。
RayPPOTrainer 拿到带奖励轨迹，计算优势，执行 PPO 更新。

4. verl-agent 的奖励机制

4.1 默认机制：轨迹级奖励的累积与写入

在默认配置下，verl-agent 使用 EpisodeRewardManager 处理轨迹级奖励。数据流分两个阶段。

阶段一：采集过程中累积原始奖励。

TrajectoryCollector 在多轮循环中把每一步环境奖励累加到 episode_rewards 数组：

episode_rewards = np.zeros(batch_size, dtype=np.float32)
episode_rewards += rewards * active_masks

设第 $i$ 条轨迹第 $t$ 步奖励为 $r_{i,t}$ ，第 $t$ 步是否仍未结束用 $m_{i,t}\in\{0,1\}$ 表示，则累计奖励可写为：

$E_i = \sum_{t=1}^{T_i} r_{i,t} \cdot m_{i,t}$

其中：

$E_i$ ：第 $i$ 条轨迹的累计奖励；
$T_i$ ：第 $i$ 条轨迹的交互步数；
$r_{i,t}$ ：第 $i$ 条轨迹第 $t$ 步环境奖励；
$m_{i,t}$ ：active mask，只有环境尚未结束时才更新，避免已结束环境继续累加。

轨迹终止后，gather_rollout_data 将累积好的 episode_rewards 和 episode_lengths 打包进 DataProto 的 non_tensor_batch，键名为 'episode_rewards' 和 'episode_lengths'。此时奖励仍是轨迹级标量。

阶段二：采集完成后张量化。

EpisodeRewardManager.__call__ 从 non_tensor_batch 读取：

episode_rewards = data_item.non_tensor_batch['episode_rewards']
episode_lengths = data_item.non_tensor_batch['episode_lengths']

根据 normalize_by_length 决定是否长度归一化：

$normalize_by_length=True E i , 否则 s_i = \begin{cases} \dfrac{E_i}{T_i}, & \text{如果 normalize\_by\_length=True} \\ E_i, & \text{否则} \end{cases}$

其中 $s_i$ 是第 $i$ 条轨迹最终写入的轨迹级分数。

然后创建与 responses 形状相同的全零 reward_tensor，仅在每条轨迹最后一个有效响应 token 位置填入该标量：

reward_tensor = torch.zeros_like(data.batch['responses'], dtype=torch.float32)
reward_tensor[i, valid_response_length - 1] = torch.tensor(score, ...)

形式化地，设 reward_tensor 形状为 $\times R$ ，其中：

$B$ ：batch size；
$R$ ：response 序列长度；
$L_i$ ：第 $i$ 条轨迹有效响应 token 数。

则：

$reward_tensor [ i , j ] = { s i , j = L i − 1 0 , 其他位置 \text{reward\_tensor}[i, j] = \begin{cases} s_i, & j = L_i - 1 \\ 0, & \text{其他位置} \end{cases}$

这里 $L_i-1$ 是最后一个有效响应 token 的索引，因为索引从 0 开始。这样做的原因是：在稀疏奖励场景中，整条轨迹的奖励通常只在终止时获得；把它放在最后一个有效响应 token 上，PPO/GRPO 等算法可以通过 response mask 将该轨迹级信号关联到整条轨迹。对应经典设定，同一优势值会分配给轨迹内所有参与训练的 token，无法区分单步动作好坏。

4.2 GiGPO：独立通道支持单步奖励

GiGPO 为解决粗粒度信用分配而设计。它显式引入 step-level 奖励与优势计算，但不是让 EpisodeRewardManager 处理每一步奖励，而是在轨迹收集阶段单独记录每一步的即时奖励，形成与轨迹奖励并行的数据通道。

step_norm_reward 定义在 gigpo/core_gigpo.py，核心逻辑如下：

def step_norm_reward(step_rewards: torch.Tensor,
                     eos_mask: torch.Tensor,
                     index: np.array,
                     epsilon: float = 1e-6):
    """Compute step-level advantage."""
    scores = step_rewards
    scores[i] = (scores[i] - id2mean[index[i]]) / (id2std[index[i]] + epsilon)
    step_advantages = scores.unsqueeze(-1).tile([1, response_length]) * eos_mask
    return step_advantages

设参与计算的样本数为 $N$ ，第 $i$ 个 step 样本的即时奖励为 $r_i$ ，其分组编号为 $g_i = \text{index}[i]$ 。对每个分组 $g$ ，先计算组内均值和标准差：

$\mu_g = \frac{1}{|G_g|}\sum_{j\in G_g} r_j$

$\sigma_g = \sqrt{\frac{1}{|G_g|}\sum_{j\in G_g}(r_j-\mu_g)^2}$

其中：

$G_g$ ：属于分组 $g$ 的样本集合；
$G_g|$ ：该分组样本数；
$\mu_g$ ：该分组即时奖励均值；
$\sigma_g$ ：该分组即时奖励标准差。

然后对每个样本做归一化：

$s_i = \frac{r_i - \mu_{g_i}}{\sigma_{g_i} + \epsilon}$

其中：

$s_i$ ：第 $i$ 个 step 样本归一化后的分数；
$g_i$ ：第 $i$ 个样本所属分组；
$\epsilon$ ：防止除零的小常数。

接着把标量 $s_i$ 广播到该 step 对应响应的所有 token 上，并用 mask 过滤无效 token：

$A_{i,t} = s_i \cdot M_{i,t}$

其中：

$A_{i,t}$ ：第 $i$ 个 step 样本第 $t$ 个 token 的 step-level 优势；
$M_{i,t}\in\{0,1\}$ ：有效 token mask，对应代码中的 eos_mask 或 response mask；
$t$ ：该 step 响应中的 token 位置。

step_norm_reward 的真正调用者是 compute_gigpo_outcome_advantage，内部会调用：

step_advantages = step_norm_reward(step_rewards, response_mask, ...)

GiGPO 的数据流中，step_rewards 是必需字段。TrajectoryCollector 在每一步与环境交互时，除了更新 episode_rewards，还会将每一步即时奖励记录到轨迹数据的独立字段。轨迹收集完成后，DataProto 同时包含：

episode_rewards：供 EpisodeRewardManager 计算轨迹级优势；
step_rewards：供 GiGPO 的 compute_gigpo_outcome_advantage 计算 step-level 优势。

GiGPO 通过双层分组实现细粒度信用分配：在 episode 级分组之下嵌套 step 级分组，利用跨轨迹中重复出现的环境状态作为“锚点状态”，比较从同一状态出发的不同动作的相对优势。这样无需额外 critic 模型即可实现逐步骤信用分配。

4.3 两者关系

EpisodeRewardManager 与 step_norm_reward 并非直接“支持”关系。它们分属两个独立阶段，通过 DataProto 中的数据字段协作：

环境 step 返回单步 reward
  ├─→ 累积到 episode_rewards → gather_rollout_data 存入 non_tensor_batch
  │     → EpisodeRewardManager 读取 → 归一化 → 全零 reward_tensor
  │     → 在每条轨迹最后一个有效 token 填 score → 轨迹级奖励
  └─→ 单独记录 step_rewards → GiGPO compute_gigpo_outcome_advantage
        → step_norm_reward 分组归一化 → 广播为 step-level 优势

奖励粒度	支持情况	对应机制
轨迹级奖励	默认、原生支持	`EpisodeRewardManager` + GRPO/PPO
单步奖励	通过 GiGPO 算法支持	`step_norm_reward`、step-level 分组、锚点状态

5. 将自定义 Gym 环境接入 verl-agent

在 verl-agent 中，外部仿真器应在 env_package/ 下新建子目录，按 Worker → Envs 包装器 → EnvironmentManager 三层实现，最后在 env_manager.py 注册。

5.1 创建环境子目录与 Worker

例如 env_package/my_sim/，实现：

MySimWorker：Ray remote actor，内部持有 gym.env 实例，封装 reset / step；
MySimEnvs：Worker 池管理器，创建并行 Worker 并分发批量调用；
__init__.py：导出上述类。

Worker 伪代码：

import ray

@ray.remote
class MySimWorker:
    def __init__(self, seed, env_kwargs):
        self.env = make_gym_env(seed=seed, **env_kwargs)

    def reset(self, idx=None):
        obs, info = self.env.reset(seed=...)
        return obs, info

    def step(self, action):
        obs, reward, terminated, truncated, info = self.env.step(action)
        return obs, reward, terminated, truncated, info

要点：

Worker 是真正持有单个环境实例的地方，gym.env 包装类几乎可以原样放在 step 中：接收 LLM 解析后的动作，调用 env.step()，返回标准 Gym 五元组。
可参考 WebshopWorker：懒加载避免 CUDA 初始化问题；在 info 中注入 available_actions；对奖励做标准化 shaping。
如果环境只给终止时总奖励，则每步 reward=0，终止时返回总奖励；TrajectoryCollector 会累加得到轨迹奖励。
如果希望使用 GiGPO 的 step-level 优势，需要让 Worker/Manager 暴露每一步即时奖励，并确保轨迹收集侧记录 step_rewards 字段。

MySimEnvs 负责创建并管理一组 Ray Worker，分配唯一 seed，将批量 reset() / step() 分发到各 Worker。无需自行实现并行逻辑，Ray 层会处理调度。

5.2 定义 Prompt 模板

创建 prompts/my_sim.py，提供系统提示和每步观测格式化模板。如果观测是结构化数据，在此定义如何转为 LLM 可读文本。EnvironmentManager 在 build_text_obs 时调用这些模板。

例如：

MY_SIM_SYSTEM_PROMPT = """你是一个在房间中行动的智能体。
请根据当前观测选择合法动作。
在 <think> </think> 中简要推理，然后输出动作。"""

MY_SIM_OBS_TEMPLATE = """当前观测：
{observation}

可选动作：
{available_actions}
"""

5.3 在 `env_manager.py` 中注册

需要完成四件事：

实现 build_my_sim_envs(...) 构建函数；
实现 my_sim_projection(text) 动作解析函数，将 LLM 文本转为 gym.env 可接受的 action；
实现 MySimEnvironmentManager(EnvironmentManagerBase)，重写 reset / step / build_text_obs；
在 make_envs 中新增路由分支。

伪代码：

def build_my_sim_envs(config):
    return MySimEnvs(config.env.num_envs, config.env.seed, config.env.env_kwargs)

def my_sim_projection(text: str):
    return parse_action(text)

class MySimEnvironmentManager(EnvironmentManagerBase):
    def reset(self, kwargs):
        obs, infos = self.envs.reset(kwargs)
        return self.build_text_obs(obs, infos), infos

    def step(self, text_actions):
        actions = [self.projection_f(t) for t in text_actions]
        obs, rewards, dones, infos = self.envs.step(actions)
        return self.build_text_obs(obs, infos), rewards, dones, infos

    def build_text_obs(self, obs, infos):
        # 调用 prompts/my_sim.py 模板
        ...

# make_envs 中新增
elif "my_sim" in config.env.env_name.lower():
    _envs = build_my_sim_envs(config)
    projection_f = partial(my_sim_projection)
    envs = MySimEnvironmentManager(_envs, projection_f, config)

Manager 统一约定：

方法	返回	用途
`reset(kwargs)`	`(obs_dict, infos)`	初始化所有子环境，返回格式化观测
`step(text_actions)`	`(obs, rewards, dones, infos)`	执行 LLM 文本动作，返回格式化新观测
`success_evaluator(...)`	任务成功判定	用于奖励计算或评估

obs_dict 通常包含 text、image、anchor 等字段，以支持纯文本和多模态环境。

5.4 训练配置指定环境名

在 PPO/GiGPO 等训练 YAML 中设置 env.env_name 为环境标识，make_envs 会自动路由。env.rollout.n 控制 rollout 数量，env.max_steps 控制单条轨迹最大交互步数。命令行示例：

python3 -m verl.trainer.main_ppo \
    algorithm.adv_estimator=gigpo \
    env.env_name=my_sim \
    env.rollout.n=8 \
    env.max_steps=50 \
    data.train_files=/path/to/train.parquet \
    data.val_files=/path/to/val.parquet

5.5 奖励传递

环境终止时返回的轨迹级 reward 会经由 Worker step() → Envs step() → Manager step() 逐层上传，最终进入 verl 奖励计算流程：

轨迹级奖励：环境每步返回的 reward 被 TrajectoryCollector 累加为 episode_rewards，gather_rollout_data 存入 DataProto.non_tensor_batch；EpisodeRewardManager 读取后做可选长度归一化，创建全零 reward_tensor，在每条轨迹最后一个有效响应 token 位置填入轨迹标量。
单步奖励（GiGPO）：如果使用 algorithm.adv_estimator=gigpo，需要在轨迹收集阶段单独记录每一步即时奖励为 step_rewards 字段，供 compute_gigpo_outcome_advantage 调用 step_norm_reward 计算 step-level 优势。EpisodeRewardManager 不处理这部分数据。
奖励 shaping：如果奖励需要折扣、归一化等特殊处理，可在 Worker 的 step 返回前完成 shaping，参考 WebshopWorker。
自定义奖励函数：也可通过 custom_reward_function.path 和 custom_reward_function.name 指定自定义奖励函数，在其中访问轨迹元数据。

6. 总结

接入自定义 Gym 环境到 verl / verl-agent 有两条路径。快速验证可以继承 AgentLoopBase，在 run() 中完成“观测格式化 → LLM 生成动作 → 动作解析 → 环境 step → 记录 token 与 mask”的循环。完整训练则按 verl-agent 的三层环境封装实现 Worker、Envs 包装器和 EnvironmentManager，并在 make_envs 中注册路由。

奖励机制上，默认配置使用 EpisodeRewardManager：先累计每条轨迹的奖励，再在采集完成后按需做长度归一化，最后在每条轨迹最后一个有效响应 token 位置写入轨迹级标量。若使用 GiGPO，则额外记录 step_rewards，由 step_norm_reward 按分组计算均值和标准差，做归一化后广播为 step-level 优势。两部分通过 DataProto 中的不同字段协作，而不是由 EpisodeRewardManager 直接处理单步奖励。