企业官网建设流程全解析

AReaL RolloutWorkflow 全面解析：从接口抽象到分组采样与轨迹落盘的实战指南

【免费下载链接】AReaLThe RL Bridge for LLM-based Agent Applications. Made Simple & Flexible.项目地址: https://gitcode.com/GitHub_Trending/are/AReaL

导读：RolloutWorkflow是 AReaL 强化学习训练管线中负责"生成训练轨迹"的核心抽象，它统一了提示词 tokenize、调用推理引擎生成补全、计算奖励、打包张量字典供训练的全流程。本文以 docs/en/reference/rollout_workflow.md 为骨架，结合 areal/api/workflow_api.py、areal/infra/remote_inf_engine.py、areal/infra/workflow_executor.py 等源码，系统讲解接口契约、三种返回类型、轨迹 JSONL 落盘格式、Grouped Rollout 分组机制，以及自定义 Workflow 的完整步骤。读完你将能够独立实现、注册并调试一个自定义 RolloutWorkflow。

前置阅读与适用场景

本文面向希望深入理解 AReaL 代码库底层的开发者。文档明确给出两条定位说明：

1. 如果你的目标是直接使用高层 API 做 agentic RL 训练，请优先阅读 Agentic RL 教程，无需接触本节底层细节；
2. 直接继承RolloutWorkflow被视为 Legacy 模式，不应主动使用。对于新的 agentic RL 工作流，推荐使用 agent workflow 模式 —— 即实现async def run()方法的类，AReaL 会自动将其包装并支持代理（proxy）。

也就是说，RolloutWorkflow抽象是理解 AReaL rollout 数据流的关键入口：即便你最终走 agent workflow 路径，其底层仍然会解析并包装成RolloutWorkflow实例。

RolloutWorkflow 是什么

一个RolloutWorkflow定义了如何从输入数据生成训练轨迹，它封装了以下逻辑：

• Tokenize 提示词并准备模型输入（input_ids、gconfig等）
• 调用推理引擎（InferenceEngine）生成补全
• 为生成结果计算奖励（reward）
• 将结果打包成张量字典（tensor dict）供训练使用

从源码看，接口定义位于 areal/api/workflow_api.py：

class RolloutWorkflow(ABC): @abstractmethod async def arun_episode( self, engine: InferenceEngine, data: dict[str, Any] ) -> dict[str, Any] | None | dict[str, InteractionWithTokenLogpReward]: """Run a single episode of the workflow.""" ...

参数说明

三种返回类型

arun_episode支持三种返回值，对应三种不同的数据通路：

其中None语义在源码 docstring 中被特别强调："ReturningNoneimplies that this trajectory is rejected and will not be used for training."（workflow_api.py）。这为自定义 workflow 提供了一种天然的过滤机制：例如当 reward 过低时直接返回None丢弃该条轨迹。

另外，InteractionWithTokenLogpReward类型定义于 areal/experimental/openai/types.py，是 OpenAI 代理工作流中按 completion ID 组织 token 级 logprob 与奖励的载体。

张量字典（Tensor Dictionary）格式

当返回张量字典时，以下字段是训练侧约定的契约：

示例返回值：

return { "input_ids": torch.tensor([[1, 2, 3, 4, 5]], dtype=torch.int32), "attention_mask": torch.ones(1, 5, dtype=torch.bool), "loss_mask": torch.tensor([[0, 0, 1, 1, 1]], dtype=torch.int32), "logprobs": torch.tensor([[0.0, 0.0, -0.5, -0.3, -0.2]], dtype=torch.float32), "rewards": torch.tensor([1.0], dtype=torch.float32), "versions": torch.tensor([[0, 0, 1, 1, 1]], dtype=torch.int32), }

关于versions字段的底层原理

versions字段与 AReaL 的**异步权重更新 + 陈旧度管理（staleness）**机制紧密相关。在 remote_inf_engine.py 的agenerate实现中可以看到：

• 每次生成请求发出前会调用request_version = self.get_version()固定当前权重版本；
• 当一次生成因权重更新被中断、分段续跑时，多个请求段可能对应不同版本，代码通过accumulated_versions.extend([request_version] * len(gen_result.output_tokens))将每个输出 token 归属于实际生成它的版本；
• 最终ModelResponse.output_versions逐 token 记录版本，供训练侧做 off-policy 分析（max_head_offpolicyness控制）。

因此，workflow 在构造张量字典时应当把resp.output_versions一并写入versions字段，这是轨迹落盘与版本分析的数据基础。

扩展字段：is_truncated与rollout_reward

源码 docstring（workflow_api.py）进一步补充了两个重要约定：

• is_truncated：当 workflow 能判断最终模型响应是否因达到长度上限而停止时，应在张量轨迹中附带一个每轨迹一个值的 bool 张量。PPO 会使用该元数据做 reward masking、value bootstrapping 与截断指标统计。
• rollout_reward：当一个arun_episode调用（一个逻辑 rollout）导出多行（例如多轮上下文化简并导出多行）且各行 reward 不一致、又开启了 reward 归一化时，需要在张量结果（或导出的InteractionWithTokenLogpReward）中提供唯一的、有限的rollout_reward标量作为组/批统计的参考，且所有提供值必须一致。显式参考可绕过行级 overlong 惩罚；actor 侧的 reward bias、scaling 与 clipping 对行和参考同时生效。

Workflow 执行上下文（Workflow Context）

在arun_episode内部，可以通过workflow_context模块访问执行上下文。每个 workflow 实例拥有相互隔离的上下文（基于 contextvars 实现）。

from areal.infra import workflow_context async def arun_episode(self, engine, data): # 获取当前执行上下文 ctx = workflow_context.get() # 判断是否处于评估模式 if ctx.is_eval: # 使用不同的评估参数 ... # 获取任务 ID 用于日志 task_id = ctx.task_id # 根据模式获取统计作用域（"rollout" 或 "eval-rollout"） scope = workflow_context.stat_scope()

WorkflowContext的完整字段定义在 areal/infra/workflow_context.py：

其中stat_scope()的实现为：评估模式返回"eval-rollout"，否则返回"rollout"（workflow_context.py），这与轨迹落盘目录的[rollout|eval-rollout]命名一一对应。

此外，同一模块还提供了get_aiohttp_session()与get_httpx_client()，为 workflow 内的 HTTP 调用（如调用外部工具、OpenAI 代理）提供按线程共享、自动复用与清理的连接池客户端，避免每个 episode 重复创建连接。

轨迹落盘（Trajectory Dumping）

当InferenceEngineConfig.dump_to_file=True时，轨迹会被自动保存到磁盘，用于调试与分析。

配置方式

rollout: dump_to_file: true fileroot: "/path/to/logs" tokenizer_path: "model/tokenizer" # 文本解码所必需

对应源码字段在 areal/api/cli_args.py：fileroot是日志与轨迹落盘的根目录，tokenizer_path用于轨迹文本解码（WorkflowExecutor._get_tokenizer会惰性加载 HF tokenizer），dump_to_file控制是否把轨迹写入 fileroot 下的文件。

输出位置

轨迹保存路径模板：

{fileroot}/{experiment_name}/{trial_name}/[rollout|eval-rollout]/{version}/{task_id}.jsonl

示例：

/tmp/areal/my_exp/trial1/rollout/5/42.jsonl

从源码看（workflow_executor.py），_get_dump_dir通过StatsLogger.get_log_path(experiment_name, trial_name, fileroot)解析日志根目录，再根据is_eval拼接rollout或eval-rollout子目录。

目录命名规则：{version}目录以批次全局最大版本（global_tail，即versions张量的全局最大值）命名。同一目录内的单条记录可能出现tail_version <= global_tail的情况——因为目录只代表"该批次曾触及的最高版本"（workflow_executor.py）。

输出格式（JSONL）

每一行 JSON 包含：

{ "task_id": 42, "sample_idx": 0, "seqlen": 512, "prompt_len": 128, "head_version": 5, "tail_version": 6, "version_rle": [[5, 100], [6, 200]], "reward": 1.0, "prompt": "<|im_start|>user\nWhat is 2+2?<|im_end|>\n<|im_start|>assistant\n", "completion": "The answer is 4.<|im_end|>" }

字段说明：

落盘实现的源码细节

_dump_trajectory（workflow_executor.py）揭示了若干值得注意的实现事实：

• 依赖完整字段：落盘要求轨迹包含input_ids、rewards、loss_mask、attention_mask四个字段，否则返回失败原因；缺少versions时回退到当前推理引擎版本并给出 warning。
• 跳过空补全：seqlen == 0或mask[-1] != 1（补全为空）的样本会被跳过。
• 版本计算：_compute_output_versions先按loss_mask过滤版本序列，再计算head = min(...)、tail = max(...)与 RLE（相邻相同版本合并计数）；无输出 token 时 head/tail 为-1。
• 多轮分段：_split_trajectory_for_dump中，只有当掩码中生成段（gen）数量 > 1 时才解码segments（单轮保持两次解码的成本）；角色依次为prompt（首个非生成段）、gen（生成段）、context（后续非生成段）。
• 批次展开：同一task_id下的批次内每条样本按sample_idx写入同一文件的不同行。
• 原始奖励保留：若轨迹携带original_rewards，则额外输出original_reward字段。

Grouped Rollout：同提示词多采样

分组 rollout 会对同一个输入提示词多次运行同一 workflow，产生多样化的补全，供 GRPO 等需要每提示词多样本的算法使用。

配置方式

提交时设置group_size：

engine.submit( data=sample, workflow=MyWorkflow, workflow_kwargs={...}, group_size=4, # 每个输入运行 workflow 4 次 )

或通过 CLI 配置：

rollout: group_size: 4

submit的完整签名（remote_inf_engine.py）还包括reward_normalization、drop_incomplete_group、min_usable_group_size等分组控制参数。

工作原理

当group_size > 1时，workflow 会被包装进GroupedRolloutWorkflow。核心流程：

1. 包装器使用asyncio.gather并发运行arun_episode共group_size次（源码中每个 slot 会先写入completed_results[sample_idx]再汇总）；
2. 结果按类型合并：
   • 张量字典：沿 batch 维度拼接（concat_padded_tensors）；
   • InteractionWithTokenLogpReward字典：合并为单个字典（合并前校验 completion ID 唯一性，重复会抛出WorkflowContractError）；
3. 每个 slot 在可用时返回正常结果类型，不可用时返回None。None是有意不透明的：分类与重试策略由生产者负责；
4. 包装器只等待原始 slot，既不会对不可用 slot 重试，也不会复制可用结果；
5. 可用 slot 恰好保留一次并拼接，其实际数量在奖励与 advantage 归一化时作为 prompt 组的边界；
6. 当reward_normalization=True时，交互奖励会在组通过最小规模过滤后、在可用 rollouts 之间做归一化；drop_incomplete_group=True仍然要求每个原始 slot 都成功，幸存 rollout 若缺失行奖励会导致整组被丢弃；
7. min_usable_group_size默认为1。v1 RL trainer 在奖励或 advantage 归一化使用组统计时将其设为2（因为该统计至少需要两个观测）；n_samples: 1的单例目标组本身完整，保持最小值为1。设置actor.min_usable_group_size会覆盖派生值，组统计生效时显式低于2的值会被拒绝。低于最小值的组返回None，异步收集器会接着取下一个就绪的 prompt 组。批量相对的 PPO 与 REINFORCE 保留可用的单例。

实现参考 remote_inf_engine.py，GroupedRolloutWorkflow的关键路径可简化为：

class GroupedRolloutWorkflow(RolloutWorkflow): async def arun_episode(self, engine, data): # 并发运行 N 次 results = await asyncio.gather( *[self.workflow.arun_episode(engine, data) for _ in range(self.group_size)] ) # 正常结果可用；None 不可用 valid_results = [r for r in results if r is not None] if len(valid_results) < self.min_usable_group_size: return None # 按结果类型合并 if all_interaction_dicts(valid_results): return merge_dicts(valid_results) else: return concat_padded_tensors(valid_results)

注意：实际实现（remote_inf_engine.py）比上述骨架更严谨——它通过asyncio.create_task为每个 slot 建立独立任务，为每个 slot 注入包含sample_idx、group_size与共享processor_cache的WorkflowContext，并在子任务异常时取消兄弟任务、排空取消处理器、调用_afinalize_processor_cache_group释放组级资源；结果还要求 slot 索引严格等于[0..group_size)，否则抛出运行时错误。

关于逻辑 rollout 与行奖励的关键约定

以下几点是该机制中容易踩坑的契约（详见 workflow_api.py 与文档原文）：

• 每次 v1arun_episode调用就是一个逻辑 rollout。上下文化简（context compaction）与agent.export_style: individual可以导出多行而不会中止整个组；收集器通过RolloutGroup元数据（携带连续行数与可选奖励参考）在rollout_group轨迹键下记录，批拼接时这些元数据进入TrajBatchMeta，切分时再还原到各轨迹。可用组大小按逻辑 rollout 计数（含n_samples: 1）。
• 奖励归一化对组统计与批统计都使用每个逻辑 rollout 一个参考。当一个 rollout 内各行奖励不同时，workflow 必须提供有限的rollout_reward标量；所有提供的参考必须一致；省略时由相等的行奖励提供参考。同一中心化与缩放应用于每行自身奖励；留一法（leave-one-out）将整个逻辑 rollout 从其参考基线中排除；参考标准差 ≤ 归一化 epsilon 时保留中心化但除数为1。内置 v1 agent workflow 会对individual导出显式提供终止奖励，从而保留折扣行奖励与任何已提供的参考；自定义 workflow 必须自行声明参考，框架不会推断任何终止、求和或均值分数。
• 显式参考不受内置行长度 overlong 惩罚影响；actor 奖励 bias、缩放与 clipping 对行与参考同时生效。无显式参考时，被惩罚的行奖励仍须一致。advantage 归一化保持其既有的掩码 token 统计与逐 token leave-one-out 行为；逻辑计数仅用于选择单例回退。GAE 仍在每行上单独运行。

损失权重语义：PPO 系列 actor 损失默认保持全局按 token 加权。因此，包含更多有效响应 token 的部分组会比更小或更短的组贡献更多损失权重。这是既有向后兼容的估计器，并不隐含"提示词等权"的声明。

导出指标：分组 rollout 会导出target_slot_count、usable_slot_count、trainable_slot_count、fully_masked_group、singleton_slot_group、pre_filter_usable_slot_yield、pre_filter_trainable_slot_yield（见_record_group_stats，remote_inf_engine.py），这些统计的是原始 rollout 调用次数；最终接受/拒绝计数是should_accept_fn运行后收集器的指标。PPO 训练还分别报告逻辑可用组大小与有效 token 损失权重分布，包括按规模统计的group_loss_weight_size_<N>指标。

输出形状

以group_size=4、workflow 返回[1, seq_len]张量、四个 slot 全部可用为例，分组输出形状为[4, seq_len]。不完整但被接受的组以其实际可用数作为首维。

实现自定义 Workflow

步骤 1：继承RolloutWorkflow

from areal.api.workflow_api import RolloutWorkflow class MyWorkflow(RolloutWorkflow): def __init__(self, tokenizer, gconfig, **kwargs): self.tokenizer = tokenizer self.gconfig = gconfig async def arun_episode(self, engine, data): # 1. 准备输入 input_ids = self.tokenizer.encode(data["prompt"]) # 2. 生成补全 req = ModelRequest( rid=uuid.uuid4().hex, input_ids=input_ids, gconfig=self.gconfig, tokenizer=self.tokenizer, ) resp = await engine.agenerate(req) # 3. 计算奖励 reward = self.compute_reward(resp, data) # 4. 返回张量字典（或返回 None 拒绝） if reward < 0: return None return self.build_tensor_dict(resp, reward)

其中ModelRequest与ModelResponse定义于 areal/api/engine_api.py，engine.agenerate(req)是异步生成入口（remote_inf_engine.py），内部会处理版本固定、中断续跑、超时重试与 stop reason 解析，返回的resp.output_tokens、resp.output_logprobs、resp.output_versions应被写入张量字典。

步骤 2：注册到 trainer

trainer.train( workflow=MyWorkflow, workflow_kwargs={ "tokenizer": tokenizer, "gconfig": config.gconfig, }, )

文档同时建议参考 areal/workflow 目录下的具体实现示例（如 areal/workflow/multi_turn.py、areal/workflow/rlvr.py），它们展示了真实算法中张量字典与奖励计算的完整写法。

Workflow 解析（Workflow Resolution）

WorkflowLike类型（workflow_api.py）允许以多种方式指定 workflow，训练系统会自动解析为RolloutWorkflow实例（解析逻辑见 remote_inf_engine.py 的_resolve_workflow）：

解析优先级与约束值得注意：

• None：仅当agent.mode == "online"时有效，走配置驱动的在线代理模式，否则抛错；
• 已是RolloutWorkflow实例：workflow_kwargs被忽略（打 warning）；
• RolloutWorkflow子类：必须提供workflow_kwargs，否则抛ValueError；
• 字符串路径：先import_from_string动态导入，再按"类 → 实例 → agent 类 → agent 实例"判定；非RolloutWorkflow的类/对象一律按 agent workflow 处理（需要proxy_addr，否则抛错）；
• group_size > 1：最后统一包一层GroupedRolloutWorkflow。

Agent workflow 与 deprecated 基类

任何带兼容async def run(data, **extra_kwargs)方法的类都可以作为 agent workflow。workflow_api.py 中的AgentWorkflow基类已被标记为Deprecated（自 1.0.0 起），其元类会在任何子类实例化时触发DeprecationWarning，提示"不再需要继承该类，任何带兼容async def run()方法的类都能工作"。extra_kwargs中 AReaL 会提供base_url、http_client、api_key等代理连接参数。

总结与调试建议

RolloutWorkflow是连接推理引擎与训练器的枢纽。掌握本文内容后，你可以：

• 理解 agentic RL 高层 API 底层的数据流与张量契约；
• 在自定义 workflow 中正确处理versions（异步权重更新场景）、is_truncated（PPO 截断处理）与rollout_reward（组奖励归一化）等扩展字段；
• 利用dump_to_file轨迹落盘检查单条样本的prompt/completion、版本区间与多轮segments；
• 正确配置group_size、min_usable_group_size、reward_normalization与drop_incomplete_group来适配 GRPO 等需要组采样的算法；
• 通过实例、类、字符串路径或 agent workflow 四种方式注册自定义 rollout 逻辑。

调试时有两处值得关注：一是WorkflowExecutor的check_trajectory_format配置（默认False，仅用于隔离调试自定义 workflow 的轨迹格式，RL 训练时应关闭）；二是轨迹落盘要求轨迹必须携带input_ids、rewards、loss_mask、attention_mask四个字段且tokenizer_path已配置，否则_dump_trajectory会返回失败原因而静默跳过。

进一步阅读：Agentic RL 教程、添加自定义 Workflow 指南、Agent Workflow 参考。

【免费下载链接】AReaLThe RL Bridge for LLM-based Agent Applications. Made Simple & Flexible.项目地址: https://gitcode.com/GitHub_Trending/are/AReaL