跳到主要内容

经典 API 主线：一次 rollout 从头到尾

这章讲什么： 经典 API（verifiers/envs/environment.py）的 Environment 基类是整个库的心脏。这一章顺着「一道题怎么变成一条带分数的轨迹」的主线，把 Environment 的三层结构讲清：单条 rollout → 单/组打分 → 全数据集并发调度。

1. 三层结构：先看骨架​

Environment 是抽象基类（ABC），只强制子类实现一个方法：rollout()。围绕它，基类提供了从「一条」到「一批」的三层封装：

 evaluate() / generate()          ← 第③层：整个数据集，并发调度 + 保存
     │  拆成一道道题
     ▼
 run_rollout() / run_group()      ← 第②层：一条(或一组) rollout + 打分 + 序列化
     │
     ▼
 rollout()  [子类实现]            ← 第①层：一条轨迹怎么跑（本身在 MultiTurnEnv 里实现）

三层各管一件事：

层方法职责是否 final
①rollout跑出一条轨迹（抽象，子类实现）抽象
②run_rollout / run_group调 rollout + 让 rubric 打分 + 转成输出@final
③generate / evaluate全量并发、限流、断点续跑、保存结果——

@final 意味着这两层框架锁死、子类不许改；子类只被允许改「一条轨迹怎么跑」。这是 Verifiers 的核心约束：打分和调度是框架的事，环境作者只负责定义交互与奖励。

依据：Environment.rollout（verifiers/envs/environment.py:599 抽象方法）、run_rollout（:723，@final）、run_group（:769，@final）、generate（:821）、evaluate（:1218）。

2. 第①层准备：init_state 把「一道题」摊平成工作台​

每条 rollout 开始前，init_state 把一个 RolloutInput（一道题：prompt + answer + info）转成一个 State——rollout 全程的可变工作台。

关键动作（init_state，verifiers/envs/environment.py:537）：

  • 深拷贝输入，把 info 里可能的 JSON 字符串解析成 dict。
  • 把 prompt 规范化成 pydantic 消息对象（normalize_messages）。
  • 解析工具定义 tool_defs（可以来自 info["tool_defs"] 或环境自带）。
  • 初始化空的 trajectory、completion=None、reward=None、计时器 RolloutTiming、唯一 trajectory_id。

一句话：init_state 只搬数据、不作任何决策，它把一道题铺开成后续循环要往里填东西的空白工作台。

3. 第②层：跑一条 + 打分​

run_rollout 是单条的完整闭环（:723）。核心就三步（在内部 _run_rollout_state，:665）：

# 示意，非源码：单条 rollout 的闭环
state = await self.rollout(input, client, model, sampling_args)  # ① 跑轨迹
if self.score_rollouts:
    await self.rubric.score_rollout(state)                       # ② 打分
await self.rubric.cleanup(state)                                 # ③ 收尾

还有两个工程细节值得记：

  • 重试：run_rollout 用 maybe_retry 包一层，max_retries 次数内失败可重跑（:765）。
  • 服务器模式：如果 env_client 非空（环境跑在独立 ZMQ 进程里），这里会把请求转发给远端而不是本地执行（:741）。这是为大规模训练时「环境和训练器分离」准备的。

3.1 为什么还要 run_group？——RL 的分组打分​

run_group（:769）一次跑同一道题的 n 条 rollout。为什么要成组？因为像 GRPO 这类 RL 算法要在「同一道题的多个采样」之间比较，算 advantage（优势 = 这条比同组平均好多少）。

分组打分在 Rubric.score_group 里体现（下一章细讲）：它给每条算 reward，再减去组内均值得到 advantage（verifiers/rubrics/rubric.py:404-407）。Environment 用 requires_group_rollouts 属性（:212）告诉调度层「这个 rubric 需要成组跑」。

4. 第③层：generate —— 全量并发的发动机​

generate（:821）是最长的一段（300+ 行），但骨架清晰。它负责把一整个数据集跑完，管好并发、进度、保存、续跑。

4.1 两种调度模式​

              generate(inputs, ...)
                     │
        ┌────────────┴─────────────┐
  independent_scoring=True     默认(grouped)
        │                           │
  每条 rollout 独立起任务       按 example_id 分组，每组一个任务
  run_rollout()                run_group()  ← 同题的 n 条留在同一端点
                                 (共享 KV cache 更省)

默认走分组：按 example_id 把同一道题的多条聚在一起，交给 run_group，并且刻意让同组落在同一个推理端点上——这样同题多次采样能复用 KV cache，省显存省时间（注释见 :1087）。

4.2 并发、限流、进度​

  • 限流：maybe_semaphore(max_concurrent) 造一个信号量，with_sem 包住每个任务，控制同时在飞的 rollout 数（:928、:1065）。
  • 异步收割：asyncio.as_completed 谁先跑完先处理谁，实时更新进度条与平均奖励（:1105）。
  • 回调：on_start / on_progress / on_log 三个钩子，默认实现是 tqdm 进度条，可被 TUI 覆盖（:855-:920）。

4.3 断点续跑（resume）​

这是个实用设计。如果给了 results_path 且已有部分结果，generate 会：

  1. 载入已完成的输出，按 rollouts_per_example 截断多余的（:1005-:1019）。
  2. 用 filter_inputs 算出还没跑的输入，只跑剩下的（:1021）。
  3. 全跑完了就直接返回（:1024）。

意味着一次大评测中途挂了，重跑不会从零开始。

依据：Environment.generate（verifiers/envs/environment.py:821），分组逻辑 :1078-:1103，续跑 :993-:1036，信号量 maybe_semaphore / with_sem（verifiers/utils/async_utils.py）。

4.4 evaluate 只是 generate 的薄封装​

evaluate（:1218）做的事很少：从 eval_dataset 取题（_get_eval_inputs，:1199），必要时 shuffle、按 rollouts_per_example 复制，然后直接调 generate。还有 evaluate_sync / generate_sync 同步版，处理 Jupyter 里已有事件循环的情况（用 nest_asyncio，:1170）。

5. 数据集怎么被喂进来​

Environment.__init__ 接受 dataset 可以是现成的 HuggingFace Dataset，也可以是一个「构造函数」DatasetBuilder（延迟到用时才 build）。_format_dataset（:343）会：

  • 给每行加一个整数 example_id（分组打分要用）。
  • 用 _ensure_prompt（:276）把 question 列 + system_prompt + few_shot 拼成标准的 prompt 消息列表。

所以你的数据集只要有 question / answer 两列，框架自动补出 prompt。

6. 一次调用的完整时序（把本章串起来）​

env.evaluate(client, model)
   └─ _get_eval_inputs()                取 eval 数据 → List[RolloutInput]
   └─ generate(inputs)
        └─ 分组 by example_id
        └─ for each group: run_group()   [限流 + 并发]
             └─ _run_group_states()
                  └─ rollout() ×n        [子类：MultiTurnEnv 实现，见 03 章]
                  └─ rubric.score_group() 算 reward + advantage
        └─ 收割 → GenerateOutputsBuilder → GenerateOutputs

下一步该看什么：

7. 代码地图​

主题文件符号
环境基类verifiers/envs/environment.pyEnvironment
建工作台verifiers/envs/environment.pyEnvironment.init_state
单条闭环verifiers/envs/environment.pyEnvironment.run_rollout / _run_rollout_state
分组闭环verifiers/envs/environment.pyEnvironment.run_group / _run_group_states
全量调度verifiers/envs/environment.pyEnvironment.generate
评测封装verifiers/envs/environment.pyEnvironment.evaluate / evaluate_sync
数据集格式化verifiers/envs/environment.pyEnvironment._format_dataset / _ensure_prompt
按 id 载入环境verifiers/utils/env_utils.pyload_environment
公开 API / 懒加载verifiers/__init__.py_LAZY_IMPORTS / __getattr__