企业官网建设流程全解析

AReaL 指标跟踪系统深度解析：从流式 Rollout 到批量训练的统一统计管线

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

本篇技术指南围绕 AReaL 仓库的指标跟踪（Metrics Tracking）机制展开，系统讲解以areal.utils.stats_tracker为核心的统一指标收集、分布式归约与多后端日志落盘方案。文章将带你掌握「流式指标 + 批量指标」双范式的工作方式、scalar()/denominator()/stat()三种记录 API 的适用场景、层级作用域与命名跟踪器的用法，以及如何通过stats_logger配置把指标送入 wandb、SwanLab、TensorBoard 等实验跟踪后端。读完即可在自定义 Rollout 工作流或训练引擎中正确埋点，并理解每一步指标从收集、聚合到提交的完整链路。

一、指标跟踪系统在 AReaL 中的定位

AReaL 是面向 LLM Agent 应用的强化学习训练框架，一次训练运行同时存在两类分布式的统计来源：

• Rollout 工作器：异步执行工作流（如 RLVRWorkflow），每个工作流独立完成一次采样与奖励计算，完成时间天然不等；
• 训练引擎：在数据并行（DP）rank 之间同步处理批次，例如 PPOActor 的ppo_update()，需要保证所有 rank 统计结果一致。

这两类来源的同步模型完全不同，因此 AReaL 提供了两种针对各自场景优化的指标范式：

整个体系构建在 areal/utils/stats_tracker.py 的DistributedStatsTracker之上，由 areal/utils/stats_logger.py 的StatsLogger负责最终提交到外部日志后端。

二、核心组件：DistributedStatsTracker

DistributedStatsTracker是线程安全的统计收集器，通过Lock保护内部状态，提供了四类核心能力：

• 命名跟踪器（Named Trackers）：不同组件拥有隔离的指标命名空间；
• 层级作用域（Hierarchical Scoping）：把相关指标组织成逻辑分组，导出时自动拼出group/sub/key形式的键；
• 分布式聚合（Distributed Aggregation）：导出时根据传入的reduce_group跨工作器自动归约；
• 多种归约类型（Reduce Types）：平均值、求和、最小/最大值、标量加权平均。

模块底部暴露了模块级便捷函数（见 stats_tracker.py），因此可以直接这样使用：

from areal.utils import stats_tracker # 默认跟踪器（训练指标） stats_tracker.scalar(learning_rate=0.001) # 命名跟踪器（Rollout 指标） stats_tracker.get("rollout").scalar(reward=0.5)

get(name)会按名称创建/复用独立的DistributedStatsTracker实例（默认名称""对应训练指标跟踪器），export_all()则遍历所有已注册的跟踪器并合并结果，若跨 rank 出现重复键会发出告警（见 stats_tracker.py）。

三、两种日志范式详解

3.1 流式指标：面向 Rollout 工作器

Rollout 工作器异步执行工作流，每个工作流在完成时单独记录标量，指标累积在工作器进程内部列表中；记录期间工作器之间无需任何同步，归约全部推迟到导出阶段由控制器完成。

以 RLVRWorkflow._collect_samples 为真实示例：

async def _collect_samples(self, engine, req, prompt_str, task_data): resp = await engine.agenerate(req) reward = await self._compute_rewards(resp, prompt_str, task_data) # 记录单个标量 - 追加到内部列表 # `workflow_context.stat_scope()` 自动区分评估/训练作用域 stats_tracker.get(workflow_context.stat_scope()).scalar(reward=reward) return resp, reward

这里workflow_context.stat_scope()（实现在 areal/infra/workflow_context.py）根据当前上下文返回"eval-rollout"（评估模式）或"rollout"（训练模式），从而把评估与训练的 Rollout 指标自动隔离到不同命名空间。

在自定义工作流中可以记录任意其他标量，例如交互轮数、最大 token 数等：

async def run(self, data, **extra_kwargs): # `workflow_context.stat_scope()` 自动区分评估/训练作用域 stats_tracker.get(workflow_context.stat_scope()).scalar( num_turns=num_turns, max_tokens=max_tokens, reward=reward ) return reward

控制器聚合：RolloutController.export_stats()（见 areal/infra/controller/rollout_controller.py）通过_collective_rpc从所有工作器收集各自已经完成本地归约的export_stats结果，再由_merge_worker_stats合并：标量均值按__count计数做加权平均，<base>/avg|min|max形式的分布类指标保留加权均值与极值，PRM 的 count/sum 类指标按 SUM 语义直接累加（见 rollout_controller.py）。这样即使各工作器处理的工作流数量不等、完成时间不一，最终均值也能按真实样本数加权，而不是简单求算术平均。

3.2 批量指标：面向训练引擎

训练引擎在数据并行 rank 之间同步处理批次。指标以带布尔掩码分母的全批次张量记录，导出时执行 all-reduce，保证各 rank 拿到完全一致的统计结果。

以 PPOActor.ppo_update 中的记录逻辑为例（与官方文档给出的模式一致）：

def ppo_update(self, data): loss_mask = data["loss_mask"].bool() reward_score = data["rewards"] # 定义分母（布尔掩码） stats_tracker.denominator( n_seqs=torch.ones_like(reward_score, dtype=torch.bool), n_valid_tokens=loss_mask, ) # 使用分母引用记录张量指标 stats_tracker.stat( advantages=data["advantages"], # [batch, seq_len] kl_rewards=data["kl_rewards"], # [batch, seq_len] denominator="n_valid_tokens" ) stats_tracker.stat( task_reward=reward_score.float(), # [batch] seq_len=seqlens.float(), # [batch] denominator="n_seqs" )

这段代码体现了「先定义分母、再引用分母记录」的核心约定：n_seqs按序列粒度统计task_reward、seq_len，n_valid_tokens按 token 粒度统计advantages、kl_rewards。从源码看，PPOActor 中还会记录correct_seq_len、incorrect_seq_len、prompt_len、no_eos_ratios、group_loss_weight等序列级指标，以及eps_clip、c_clip、use_dual_clip等超参标量，全部通过scalar()走默认跟踪器。

导出行为：训练引擎的export_stats()（见 areal/engine/fsdp_engine.py）把self.data_parallel_group作为归约组传给export_all()：

def export_stats(self) -> dict[str, float]: # 跨数据并行组 all-reduce return stats_tracker.export_all(reduce_group=self.data_parallel_group) # 所有 DP rank 接收相同的结果

底层_aggregate（见 stats_tracker.py）会按归约类型调用_avg_of、_sum_of、_min_of、_max_of等实现：平均值是「掩码内元素求和 ÷ 掩码内元素计数」的加权平均；标量则是先 all-reduce 求和、再 all-reduce 计数，最终除以计数得到加权均值，并同时输出key__count。

四、记录 API 与归约类型参考

4.1 三种记录方法

源码层面的约束值得注意（见 stats_tracker.py）：

• denominator()的值必须是非空的 PyTorch bool 张量，否则直接抛ValueError；
• stat()的值必须是非空的 float 张量，且不允许使用SCALAR归约类型；
• stat()引用的分母必须已经通过denominator()注册过，且每次记录的 shape 必须与分母一致（内部有assert x.shape == y.shape校验）；
• 所有张量在记录时执行detach().clone()，切断与计算图的联系，避免梯度残留。

4.2 归约类型

使用stat()时指标默认归约类型为AVG_MIN_MAX，一个键会导出三个输出键：

stats_tracker.stat(loss=tensor, denominator="valid") # 导出：{"loss/avg": 0.5, "loss/min": 0.1, "loss/max": 0.9}

可用的归约类型由ReduceType枚举定义（见 stats_tracker.py）：

细节上，AVG_MIN_MAX的min/max仅在掩码为真的位置取值（torch.where(d, v, float("inf")).min()），空批次会返回None并从结果中剔除；SCALAR类型同时输出key__count，正是控制器做跨工作器加权平均所需的计数键。

五、作用域、计时与命名跟踪器

5.1 层级作用域

使用with语句可以嵌套组织指标，导出时自动拼接出完整键名：

with stats_tracker.scope("ppo_actor"): with stats_tracker.scope("update"): stats_tracker.stat(loss=loss_tensor, denominator="valid") # 键："ppo_actor/update/loss/avg"

实现上，Scope上下文管理器在进入/退出时维护一个scope_stack（见 stats_tracker.py），_get_full_key()把作用域栈与当前键用/拼接。此外还提供了两个配套工具：

• scope_func_wrapper(name)：装饰器形式的作用域，等价于在函数体内包一层with；
• disable_scope()：临时清空作用域栈，用于绕过当前作用域记录全局键。

5.2 计时

record_timing(key)用time.perf_counter()测量代码块执行时间，并固定记录在timeperf/前缀下：

with stats_tracker.record_timing("rollout"): batch = actor.prepare_batch(dataloader, workflow) # 键："timeperf/rollout"

PPOTrainer内部大量使用该机制记录各阶段耗时，例如rollout（收集批次）、critic_values、ref_logp、teacher_logp、rollout_onload、rollout_pause、rollout_offload以及eval等（见 areal/trainer/rl_trainer.py 的_onload_rollout/_offload_rollout/train方法）。这些timeperf/*指标会以SCALAR类型随训练指标一起导出，用于分析训练吞吐与各阶段开销。

5.3 命名跟踪器

为不同组件隔离指标：

# 训练指标（默认跟踪器） stats_tracker.scalar(grad_norm=1.5) # Rollout 指标 stats_tracker.get("rollout").scalar(reward=0.8) # 评估指标 stats_tracker.get("eval-rollout").scalar(reward=0.9) # 从所有跟踪器导出 all_stats = stats_tracker.export_all(reduce_group=group)

export_all()会通过all_gather_object同步各 rank 已注册的跟踪器名集合，从而保证所有 rank 导出的键一致（见 stats_tracker.py）。

另外，如果配置了evaluator.eval_before_train: true，系统会在第一个训练 step 开始前先执行一次评估并导出指标（见 rl_trainer.py），用于在微调开始前估计初始模型的性能基线。该配置项定义于 areal/api/cli_args.py。

六、完整数据流：从收集到记录

综合 RLVRWorkflow、RolloutController、FSDPPPOActor 与 PPOTrainer 的实现，指标从产生到落盘的完整流程如下：

Rollout 工作器 训练工作器 ─────────────── ─────────────── workflow.arun_episode() actor.ppo_update(batch) │ │ ▼ ▼ get("rollout").scalar(r=0.5) stat(tensor, denom=mask) │ │ ▼ ▼ export_stats(reduce_group=None) export_stats(reduce_group=dp_group) {reward: 0.5, reward__count: 1} → all_reduce 跨 DP rank │ │ ▼ │ RolloutController.export_stats() │ → 加权平均跨工作器 │ │ │ └────────────────┬───────────────────────┘ ▼ PPOTrainer._export_and_commit_stats() │ ▼ StatsLogger.commit(stats) │ ┌────────────┼────────────┐ ▼ ▼ ▼ wandb tensorboard swanlab

PPOTrainer._export_and_commit_stats()（见 rl_trainer.py）在每个训练步结束时依次收集三类指标并提交：

def _export_and_commit_stats(self, epoch, epoch_step, global_step): # 1. 从所有组件收集指标 stats = self.actor.export_stats() # 训练指标（all-reduced） stats.update(self.rollout.export_stats()) # Rollout 指标（控制器聚合） stats.update(self.eval_rollout.export_stats()) # 评估指标 # 2. 发送到日志后端（仅 rank 0） self.stats_logger.commit(epoch, epoch_step, global_step, stats)

值得注意的是，eval_rollout的export_stats()同样经由RolloutController的控制器聚合逻辑，因此评估指标（键前缀eval-rollout/）与训练 Rollout 指标（键前缀rollout/）的聚合路径一致。

七、StatsLogger：日志后端

7.1 职责与生命周期

StatsLogger 把聚合后的指标发送到外部日志后端，由PPOTrainer在初始化时自动创建（StatsLogger(config, ft_spec)，见 rl_trainer.py）。它在init()中完成各后端的初始化：仅在 rank 0 执行以避免重复日志；在commit()中过滤内部计数键并写入所有后端；在close()中收尾（wandb.finish()、swanlab.finish()、trackio.finish()与summary_writer.close()）。它还实现了state_dict()/load_state_dict()，支持随训练状态一起保存与恢复last_commit_step，用于断点续训时保证日志 step 连续。

7.2 支持的后端

说明：当前仓库的 StatsLogger 实现 中除文档列举的 wandb / SwanLab / TensorBoard 外，还支持 Trackio 后端（trackio.init()+trackio.log()）。Trackio 的配置类TrackioConfig定义于 areal/api/cli_args.py，支持disabled/online/local三种模式，以及可选的space_id（部署 HF Space 远端看板）。

init()中的细节还包括：wandb 支持通过wandb_base_url/wandb_api_key环境注入，id_suffix="timestamp"时自动追加时间戳以支持多 run 续写（resume="allow"）；SwanLab 的api_key缺省时读取SWANLAB_API_KEY环境变量；完整实验配置会经redact_sensitive_config()脱敏后连同version_info（commit_id、branch、is_dirty、version）一并写入各后端的config字段，便于回溯实验环境。

7.3 StatsLogger.commit()

commit()过滤掉__count内部键后，按global_step写入所有后端：

def commit(self, epoch, step, global_step, data): if dist.is_initialized() and dist.get_rank() != 0: return # 仅 rank 0 记录 # 过滤掉 __count 键（用于内部加权平均） data = {k: v for k, v in data.items() if not k.endswith("__count")} # 记录到所有后端 wandb.log(data, step=global_step) swanlab.log(data, step=global_step) if self.summary_writer: for key, val in data.items(): self.summary_writer.add_scalar(key, val, global_step)

同时commit()还会通过tabulate_stats()把指标打印为表格日志（print_stats），并在日志中输出Epoch/Step/Train step进度信息。_last_commit_step保证断点续训后日志 step 单调递增，log_step = max(global_step, self._last_commit_step + 1)。

7.4 配置示例

在实验配置中配置日志后端（对应StatsLoggerConfig，定义于 areal/api/cli_args.py）：

stats_logger: experiment_name: "gsm8k_grpo" trial_name: "run_001" fileroot: "/path/to/logs" wandb: mode: "online" # "online"、"offline" 或 "disabled" project: "my-project" entity: "my-team" swanlab: mode: "online" # "online"、"local" 或 "disabled" project: "my-project" tensorboard: path: "/path/to/tensorboard/logs" # null 禁用

配置字段与源码对应的取值约束：

• experiment_name/trial_name/fileroot为必填项（缺失会报错），日志根目录由get_log_path()计算为{fileroot}/logs/{user}/{experiment_name}/{trial_name}；
• wandb.mode合法值：online、offline、disabled、shared（默认disabled）；wandb.project缺省时回退到experiment_name，wandb.name缺省时回退到trial_name，group缺省为{experiment_name}_{trial_name}；
• swanlab.mode合法值：cloud、local、disabled、offline（默认disabled），project缺省回退experiment_name；
• tensorboard.path为null时禁用 TensorBoard（SummaryWriter不创建）；
• trackio.mode合法值：disabled、online、local（默认disabled）。

八、最佳实践与常见陷阱

1. 选择正确的范式：对标量（学习率、奖励、超参、耗时）使用scalar()；对批量 PyTorch 张量（通常是训练指标，如 loss、advantages、seq_len）使用带分母的stat()。
2. 先定义分母：始终在stat()之前调用denominator()建立掩码关系——stat()引用不存在的分母会直接抛ValueError，这是源码层面的硬约束。
3. 使用命名跟踪器：使用stats_tracker.get(workflow_context.stat_scope()).scalar(...)将 Rollout（"rollout"）和评估（"eval-rollout"）指标与训练指标隔离，避免键名冲突。
4. 注意作用域前缀：在with stats_tracker.scope(...)块内记录的所有键都会带上作用域前缀，跨组件统计时要注意导出键名的拼接规则。
5. __count键是内部实现细节：SCALAR归约会导出key__count用于加权平均，commit()会自动过滤，控制器合并时也会跳过，不要在自定义埋点中刻意手工构造该后缀键。
6. 分布式键一致性：export_all()通过all_gather_object同步各 rank 的跟踪器集合与键元数据；若不同 rank 对同一键的归约类型/分母元数据不一致，会在合并时抛出ValueError，因此请确保所有 rank 按相同顺序、相同类型记录同一批键。

九、测试与验证

仓库在 tests/test_stats_tracker.py 中覆盖了指标系统的关键行为，包括：

• 当某个 rank 缺失某键（key_sync_group元数据同步）时导出不崩溃（test_export_scalar_key_missing_on_this_rank_does_not_crash）；
• 使用比默认导出组更宽的key_sync_group时，默认键仍保持原reduce_group对齐（test_export_keeps_default_reduce_group_when_key_sync_group_is_larger）；
• 跨归约组时 CPU 上的标量会先迁移到 NCCL 设备再执行 all-reduce（test_all_reduce_moves_cpu_stat_to_nccl_device_before_reduction）。

这些测试印证了export()中「per-key reduce_group 覆盖 + 元数据同步」的设计：特定键（如 CP-local 场景下的 loss/vocab_* 需要跨 DP+CP 归约）可以通过stat(..., reduce_group=...)单独指定归约组，而其余键仍跟随默认组。

十、小结

AReaL 的指标跟踪系统用「一套 API、两种范式」解决了强化学习训练中最棘手的分布式指标一致性问题：流式指标让异步 Rollout 工作器零同步地累积标量，导出时由RolloutController按样本数加权聚合；批量指标让训练引擎在 DP rank 之间通过 all-reduce 得到全局一致的统计。在此基础上，层级作用域、命名跟踪器、六种归约类型、record_timing计时以及StatsLogger对 wandb / SwanLab / TensorBoard / Trackio 的多后端支持，构成了从埋点、聚合到可视化的完整闭环。无论是扩展自定义工作流、新增训练指标，还是接入新的实验跟踪平台，都可以直接从本文的 API 参考与数据流图中获得可落地的指引。

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