# PR #1740 完整报告 - 仓库：`radixark/miles` - 标题：feat (async): support evaluation for fully-async training (dedicated fleet / pause-the-world / external service) - 合并时间：2026-08-04 12:23 - 原文链接：http://prhub.com.cn/radixark/miles/pull/1740 --- # 执行摘要 - 一句话：fully-async 评估：共享引擎 / 专用机群 / 外部黑盒三种姿态 - 推荐动作：值得精读。核心设计决策包括：按权重来源切分评估姿态、快照单一 owner 防泄漏、版本钉住与逐引擎读回校验、`--eval-sglang-*` 继承式参数与 tp-coupled 回退、黑盒契约保持目录进 / 结果出。建议重点阅读 `eval_dispatch.py`、`eval_fleet.py`、`checkpoint_eval.py` 三件套及其 fast 测试，可复用到任何需要异步评估的 RL 训练框架。合并后应补跑一次 head commit 上共享引擎暂停姿态的 GPU 验证。 # 功能与动机 PR body 明确指出："Fully-async rollout has no evaluation story: the producer never stops, so there is no quiet window to evaluate in, and running eval on the training engines is structurally broken — eval requests get aborted by every weight update, and no well-defined weight version is measured." 同时 "Heavy eval sets also steal training inference capacity even in sync mode"，即重型评估集即使在同步模式下也会挤占训练推理容量。因此需要一种让训练不被中断、且每次评估都钉住确定权重版本的方案。 # 实现拆解 1. **定义评估姿态与契约**：新增 `miles/rollout/checkpoint_eval.py`，定义 `CheckpointEvalFn`（黑盒契约：目录进、结果出）、`EvalSkip`（可归属跳过）、`retarget_args`（浅拷贝 args 并替换 router 与 GPU 规模）和 `is_checkpoint_eval_fn`。两种姿态按权重来源切分：共享训练引擎用阻断式调用钉住版本；快照姿态只通过 `(rollout_id, HF snapshot dir)` 交互，依赖 `update_weights_from_disk` 加载并把 `weight_version=str(rollout_id)` 作为版本。`args.eval_uses_snapshots` 是唯一判别量，由 `_resolve_rollout_functions` 派生，驱动侧 `EvalDispatcher` 与 `RolloutManager` 共享但互不 import。 2. **两种快照后端**：job 内专用 eval fleet（`--eval-num-gpus`）由 `miles/ray/rollout/eval_fleet.py` 的 `EvalFleet.pin` 实现——健康探测（无故障容错时调用 `RolloutServer.probe_and_mark_dead`）、`recover`、`wait_all_engines_alive`、逐引擎 `update_weights_from_disk` 加 `get_weight_version` 读回校验、路由器就绪探针，成功后返回 `GenerateState` 交给配置好的 eval fn。黑盒后端（`--eval-function-path` 指向 `CheckpointEvalFn` 子类）直接拿到快照目录自行加载与生成，`examples/fully_async/external_eval_fn.py` 是参考实现：通过 `MILES_EXTERNAL_EVAL_GPUS`/`MILES_EXTERNAL_EVAL_URL` 启动或挂接独立 sglang server，并用 `_pin` 完成 load 后版本读回。两种后端互斥，统一走 `RolloutManager._eval_checkpoint` 单一路径。 3. **驱动侧异步派发与快照所有权**：`miles/ray/rollout/eval_dispatch.py` 的 `EvalDispatcher` 负责有界并发（`--eval-max-in-flight`）、溢出策略（`--eval-overflow-policy backpressure|skip`）、快照导出（`actor_model.export_hf`）与 `_retire` GC（受 `--eval-keep-snapshots` 约束）。设计核心是快照单一 owner：导出与回收都在 dispatcher，每个结算点无论成功、跳过还是崩溃都恰好回收一个快照，避免任何失败路径向 `--eval-hf-dir`（文档建议 /dev/shm）泄漏模型大小目录。 4. **引擎配置与参数继承**：`miles/backends/sglang_utils/arguments.py` 用 `_add_prefixed_server_args` 注册 `--eval-sglang-*`（`argparse.SUPPRESS` 默认 + `BooleanOptionalAction`），`collect_eval_sglang_overrides` 收集用户显式覆盖；`miles/ray/rollout/rollout_server.py` 的 `_apply_eval_model_config`/`_eval_sglang_overrides` 把 eval 模型组填入 `SglangConfig`。TP 取自 `--eval-num-gpus-per-engine`；当 eval TP 与 rollout TP 不同时，`dp/pp/ep/attn_cp` 回退 1，避免 `tp=1 ep=8` 这类 SGLang 启动校验直接拒绝的配置。`sglang_overrides` 合并被移到 args 派生分支之后，修复了 per-group override 被 `--sglang-config` 条件分支覆盖的问题。 5. **导出重构与测试配套**：`miles/backends/megatron_utils/hf_export.py` 从 `model.py` 接收 `save_hf_model` 并新增 `export_hf_model_direct`——走 miles 自身 megatron→HF 转换器，导出覆盖范围与权重同步一致；`.complete` marker 在导出前清理、LoRA adapter 写完后补写；导出 barrier 移入 `finally`，rank 0 单点失败也不会让其他 rank 卡死直到 NCCL watchdog。测试侧：`tests/e2e/megatron/test_qwen3_4b_fully_async_eval.py` 以短 Qwen3-4B GRPO 跑覆盖共享 /fleet/external 三种姿态；fast 测试覆盖契约、fleet pin/recover、dispatcher 策略与快照所有权、manager 快照路径、`--eval-sglang-*` 继承与优先级；文档同步更新 `docs/user-guide/fully-async.md` 与 CLI 参考。 关键文件： - `miles/rollout/checkpoint_eval.py`（模块 评估契约；类别 source；类型 core-logic；符号 CheckpointEvalFn, evaluate_checkpoint, EvalSkip, retarget_args）: 新增黑盒评估契约，是快照姿态与外部后端的公共接口，定义 `CheckpointEvalFn`/`EvalSkip`/`retarget_args`。 - `miles/ray/rollout/eval_fleet.py`（模块 评估机群；类别 source；类型 core-logic；符号 EvalFleet, pin, _pin_fleet, _fleet_args）: 实现专用评估机群的核心 pin 逻辑：健康探测、恢复、逐引擎加载与版本读回校验。 - `miles/ray/rollout/eval_dispatch.py`（模块 派发器；类别 source；类型 core-logic；符号 EvalDispatcher, dispatch, drain, _export）: 驱动侧异步派发器，承载快照导出、有界并发、溢出策略与快照唯一 owner 的回收逻辑。 - `miles/backends/megatron_utils/hf_export.py`（模块 HF 导出；类别 source；类型 dependency-wiring；符号 export_hf_model_direct, save_hf_model, _is_hf_metadata_file, _get_hf_bridge）: HF 导出从 model.py 迁入并新增直连 megatron→HF 导出器，修复导出 barrier 的 rank 0 单点失败问题。 - `miles/ray/rollout/rollout_manager.py`（模块 管理器；类别 source；类型 core-logic；符号 eval, _eval_checkpoint, report_eval_skip）: 新增 `_eval_checkpoint` 单一路径：fleet pin 优先，再调用配置的评估 fn，所有失败降级为可归属跳过。 - `miles/backends/sglang_utils/arguments.py`（模块 参数层；类别 source；类型 configuration；符号 _add_prefixed_server_args, add_sglang_arguments, collect_eval_sglang_overrides）: 注册 `--eval-sglang-*` 继承式参数与布尔 `--no-` 形式，是 eval fleet 引擎配置的入口。 - `miles/ray/rollout/rollout_server.py`（模块 引擎服务；类别 source；类型 core-logic；符号 _eval_sglang_overrides, _apply_eval_model_config, _resolve_sglang_config, probe_and_mark_dead）: 在 SglangConfig 中追加 eval 模型组，应用 tp-coupled 回退与 eval overrides，并新增 `probe_and_mark_dead`。 - `miles/utils/arguments.py`（模块 参数校验；类别 source；类型 configuration）: 新增 eval 相关参数与校验矩阵，并修复 fully-async override 之后再解析默认 eval 路径的关键缺陷。 - `examples/fully_async/external_eval_fn.py`（模块 外部后端；类别 source；类型 core-logic；符号 ExternalSglangEvalFn, evaluate_checkpoint, _pin, dispose）: 参考 `CheckpointEvalFn` 实现：启动或挂接独立 sglang 服务，是外部黑盒姿态的模板。 - `examples/fully_async/run_qwen3_5_4b_fully_async_eval.py`（模块 示例脚本；类别 source；类型 core-logic；符号 ScriptArgs, prepare, execute, main）: 同一脚本通过 `--eval-backend fleet|external` 切换两种快照后端，是功能示例与 e2e 的参照。 - `tests/e2e/megatron/test_qwen3_4b_fully_async_eval.py`（模块 评估测试；类别 test；类型 test-coverage；符号 prepare, execute）: 覆盖共享 /fleet/external 三种姿态的 GPU e2e，是本次功能的主要验收测试。 关键符号：CheckpointEvalFn.evaluate_checkpoint, EvalSkip, retarget_args, EvalFleet.pin, EvalDispatcher.dispatch, RolloutManager._eval_checkpoint, FullyAsyncRolloutFn._call_eval, export_hf_model_direct, save_hf_model, collect_eval_sglang_overrides, _apply_eval_model_config, ExternalSglangEvalFn ## 关键源码片段 ### `miles/rollout/checkpoint_eval.py` 新增黑盒评估契约，是快照姿态与外部后端的公共接口，定义 `CheckpointEvalFn`/`EvalSkip`/`retarget_args`。 ```python # miles/rollout/checkpoint_eval.py —— 黑盒评估后端契约。 # 训练侧负责：导出快照、异步派发、结果记在快照对应 step、之后回收目录； # 后端负责：在 __init__ 里准备服务，在 evaluate_checkpoint 里消费快照目录。 import abc import copy import logging from argparse import Namespace from miles.rollout.base_types import RolloutFnEvalInput, RolloutFnEvalOutput, RolloutFnInput logger = logging.getLogger(__name__) class EvalSkip(Exception): # 抛出自定义异常即可让该评估点被登记为 eval/skipped_{reason}， # 而不是被当作崩溃处理 —— 训练循环永远不会因此中断。 def __init__(self, reason: str): super().__init__(reason) self.reason = reason def retarget_args(args: Namespace, router_ip, router_port, num_gpus: int, num_gpus_per_engine: int) -> Namespace: # 浅拷贝 args，只替换路由器地址与 GPU 规模；下游 GenerateState 与 # 生成函数都从 args 读取这些字段，因此同一套评估代码可无修改地 # 指向另一组引擎（共享引擎 / fleet / 外部服务共用这一份逻辑）。 eval_args = copy.copy(args) eval_args.sglang_router_ip = router_ip eval_args.sglang_router_port = router_port eval_args.rollout_num_gpus = num_gpus eval_args.rollout_num_gpus_per_engine = num_gpus_per_engine return eval_args class CheckpointEvalFn(abc.ABC): # 黑盒契约：目录进、结果出，中间是后端自己的事。 @abc.abstractmethod async def evaluate_checkpoint(self, checkpoint_dir: str, input: RolloutFnEvalInput) -> RolloutFnEvalOutput: ... async def __call__(self, input: RolloutFnInput) -> RolloutFnEvalOutput: # 只服务 eval；训练 fn 必须留在 --rollout-function-path。 assert input.evaluation, "CheckpointEvalFn only serves eval; keep the train fn on --rollout-function-path" assert input.hf_dir is not None, ( "no snapshot was dispatched — checkpoint eval fns require train_async.py " "and a snapshot source (--eval-hf-dir or --save-hf)" ) return await self.evaluate_checkpoint(input.hf_dir, input) def dispose(self) -> None: # 可选钩子：销毁 __init__ 里启动的进程 / 客户端，默认 no-op。 ... ``` ### `miles/ray/rollout/eval_fleet.py` 实现专用评估机群的核心 pin 逻辑：健康探测、恢复、逐引擎加载与版本读回校验。 ```python # miles/ray/rollout/eval_fleet.py —— job 内专用评估机群（--eval-num-gpus）。 # 机群只负责“把权重送到位”：pin 加载快照、逐引擎读回版本并确认一致， # 然后把 GenerateState 交还，由配置好的评估 fn 去生成 —— 与训练引擎上的路径完全一致。 import asyncio import logging from argparse import Namespace from miles.rollout.checkpoint_eval import EvalSkip, retarget_args from miles.rollout.inference_rollout.inference_rollout_common import GenerateState from miles.utils.http_utils import wait_http_ok logger = logging.getLogger(__name__) EVAL_WEIGHT_LOAD_TIMEOUT_SECS = 600.0 class EvalFleet: def __init__(self, args: Namespace, *, srv): self.args = args self._srv = srv self._state = GenerateState(self._fleet_args()) async def pin(self, checkpoint_dir: str, weight_version: str) -> GenerateState: # 健康检查三段式：无健康监控时自己探测，recover 重启死引擎，等全部复活。 try: if not self.args.use_fault_tolerance: await self._srv.probe_and_mark_dead() await self._srv.recover() await self._srv.wait_all_engines_alive() except Exception as e: logger.warning(f"Eval fleet unhealthy: {e}") raise EvalSkip("unhealthy") from e if not await self._pin_fleet(checkpoint_dir, weight_version): raise EvalSkip("pin_violation") # 引擎复活后路由器可能还在 503，用单 token 探针确认路由可用再放行。 try: await self._wait_router_ready() except Exception as e: logger.warning(f"Eval router not ready: {e}") raise EvalSkip("unhealthy") from e return self._state def _fleet_args(self) -> Namespace: router_ip, router_port = self.args.sglang_model_routers["eval"] return retarget_args( self.args, router_ip, router_port, self.args.eval_num_gpus, self.args.eval_num_gpus_per_engine ) async def _pin_fleet(self, checkpoint_dir: str, weight_version: str, *, retries: int = 2) -> bool: # 路由器在引擎间做负载均衡，任何一台版本不一致都会造成混合版本； # 所以逐台 update_weights_from_disk 后必须全部 get_weight_version 一致才算成功。 actors = [e.actor_handle for e in self._srv.engines] versions: list = [] for attempt in range(retries): try: await asyncio.wait_for( asyncio.gather( *[a.update_weights_from_disk.remote(checkpoint_dir, weight_version=weight_version) for a in actors] ), timeout=EVAL_WEIGHT_LOAD_TIMEOUT_SECS, ) versions = await asyncio.wait_for( asyncio.gather(*[a.get_weight_version.remote() for a in actors]), timeout=EVAL_WEIGHT_LOAD_TIMEOUT_SECS, ) except Exception as e: logger.warning(f"Weight pin to {checkpoint_dir} failed (attempt {attempt + 1}/{retries}): {e}") continue if versions and all(str(v) == weight_version for v in versions): return True logger.warning(f"Failed to pin weight_version={weight_version} to {checkpoint_dir} (got {versions})") return False ``` ### `miles/ray/rollout/eval_dispatch.py` 驱动侧异步派发器，承载快照导出、有界并发、溢出策略与快照唯一 owner 的回收逻辑。 ```python # miles/ray/rollout/eval_dispatch.py —— 驱动侧异步派发器（不阻塞训练循环）。 # 设计核心：快照的“导出”与“回收”由同一方持有，任何失败路径都不会泄漏 # 模型大小的目录；每个评估点无论成功、跳过还是崩溃，都恰好回收一个快照。 import logging import os import shutil import time from collections import deque import ray logger = logging.getLogger(__name__) class EvalDispatcher: def __init__(self, args, actor_model, rollout_manager): self.args = args self.actor_model = actor_model self.rollout_manager = rollout_manager self.pending: deque[tuple[int, ray.ObjectRef, str | None]] = deque() self._exported: list[str] = [] async def dispatch(self, rollout_id: int, hf_dir: str | None = None, force: bool = False) -> None: # 非快照姿态保持原有共享引擎调用形状（阻断式，让 FullyAsyncRolloutFn 自己暂停 producer）。 if not self.args.eval_uses_snapshots: await self.rollout_manager.eval.remote(rollout_id) return await self._reap_finished() if len(self.pending) >= self.args.eval_max_in_flight: if self.args.eval_overflow_policy == "skip" and not force: await self.rollout_manager.report_eval_skip.remote(rollout_id, "busy") return await self._settle(*self.pending.popleft()) # backpressure：先结算最老的 export_time = None exported_dir = None require_marker = hf_dir is None # 调用方自带的目录（如训练前基线）没有 .complete if hf_dir is None: if self.args.eval_hf_dir is None: hf_dir = self.args.save_hf.format(rollout_id=rollout_id) else: hf_dir = os.path.join(self.args.eval_hf_dir, f"step_{rollout_id}") try: export_time = await self._export(rollout_id, hf_dir) except Exception as e: logger.error(f"HF snapshot export for eval {rollout_id} failed: {e}") shutil.rmtree(hf_dir, ignore_errors=True) await self.rollout_manager.report_eval_skip.remote(rollout_id, "export_failed") return exported_dir = hf_dir ref = self.rollout_manager.eval.remote( rollout_id, hf_dir=hf_dir, export_time_seconds=export_time, require_marker=require_marker ) self.pending.append((rollout_id, ref, exported_dir)) async def _settle(self, rollout_id: int, ref, exported_dir: str | None) -> None: try: await ref except Exception: logger.exception(f"Async eval for rollout {rollout_id} raised") await self.rollout_manager.report_eval_skip.remote(rollout_id, "crashed") finally: self._retire(exported_dir) def _retire(self, exported_dir: str | None) -> None: if exported_dir is None: return self._exported.append(exported_dir) while len(self._exported) > self.args.eval_keep_snapshots: victim = self._exported.pop(0) shutil.rmtree(victim, ignore_errors=True) logger.info(f"GC'd consumed eval snapshot {victim}") ``` # 评论区精华 Review 中最有价值的交锋来自 ashtonchew 对共享引擎 eval 契约的核对：`resolve_rollout_function_paths()` 在 fully-async override 之前计算默认 eval path，导致 `--fully-async` 默认得到 `(FullyAsyncRolloutFn, InferenceRolloutFn)` 两个不同路径，`RolloutManager` 因而构造了第二个 `InferenceRolloutFn` 并绕过 `FullyAsyncRolloutFn._call_eval()` 的 producer pause。他建议的有界修复是 `eval_path = args.eval_function_path or rollout_path`（在 fully-async override 之后解析），并已用 resolver 测试复现；该问题由提交 b0aa5e5 修复。此外 gemini-code-assist 提出三处健壮性建议：`args.hf_checkpoint` 为 Hub ID 时 `iterdir()` 会崩溃（最终代码改为 `is_dir()` 判断后仅 warning）、用 `ValueError`/`FileNotFoundError` 替代 `assert`（避免 `-O` 下被 strip）、探针避免捕获过宽异常。 - 共享引擎 eval 路径解析失控：默认 eval_path 未经过 fully-async override (correctness): 采用有界修复：`eval_path = args.eval_function_path or rollout_path`，放在 fully-async override 之后；显式 `--eval-function-path` 不受影响。 - 快照所有权分散导致失败路径泄漏模型目录 (design): 快照单一 owner 收敛到 EvalDispatcher：每个结算点无论结果都恰好回收一个快照，`_retire` 按 `eval_keep_snapshots` 做 GC。 - eval TP 不同时继承 dp/pp/ep/attn_cp 导致引擎启动校验失败 (correctness): eval TP 与 rollout TP 不同时，tp-coupled 的四个维度回退为 1，用户可用 `--eval-sglang-*` 显式覆盖。 - args.hf_checkpoint 为 Hub ID 时 iterdir 崩溃 (correctness): 最终 `export_hf_model_direct` 先判断 `base_checkpoint.is_dir()`，非本地目录时仅记录 warning 并跳过 metadata 拷贝。 - 用 assert 做参数校验在 -O 下会被剥离 (design): Review 中提出但合并前未系统性替换；当前以 assert 为主，风险较低但值得后续统一处理。 # 风险与影响 - 风险： 1. **共享引擎暂停姿态未经验证**：PR body 自述 "These runs predate the rework and need repeating on the head commit. In particular the shared-engine numbers were produced by code in which the producer pause could not run, so that posture is effectively unverified"。合并时该姿态的实际暂停效果仍缺 GPU 验证。 2. **跨模块重构回归**：`save_hf_model` 从 `model.py` 迁入 `hf_export.py`，任何仍从旧路径 import 的插件或脚本会断裂；`_compute_server_args` 的 overrides 合并顺序修复会影响所有使用 `--sglang-config` 的多组（含 PD/ 多 LoRA）配置。 3. **外部依赖耦合**：retract 暂停模式需要 sgl-project/sglang#31962 与 #1750 才能可靠工作；外部服务姿态依赖 sglang 的 `update_weights_from_disk`/`model_info` 版本读回语义。 4. **快照竞争与资源**：快照在 /dev/shm 上 staging，`eval_keep_snapshots + eval_max_in_flight` 个模型大小目录可能超过容器 shm 配额；fleet pin 与 router 就绪之间存在引擎复活后的 503 窗口，依赖单 token 探针兜底。 5. **CI 覆盖盲区**：GPU e2e 依赖 `run-ci-eval` label 才能被执行，labelless 的绿跑并不覆盖该测试；合并后共用引擎姿态验证不足构成监控盲区。 - 影响：对用户：fully-async 训练首次获得完整评估方案，且不中断训练；新增 `--eval-num-gpus`、`--eval-function-path`、`--eval-sglang-*`、`--eval-max-in-flight`、`--eval-overflow-policy` 等参数。对系统：默认（同步训练且不配置快照姿态）路径保持原有调用形状不变，`eval_rollout_single_dataset` 的评分语义被单独 revert 回失败样本计 0.0，避免静默移动既有 eval 曲线；`hf_export` 迁移与参数解析顺序修复是潜在行为变化点，需要关注 `--sglang-config` 用户。对团队：提供了可复用的黑盒评估契约与参考实现，外部评测服务可用同一套训练 args，无需手工复制配置。影响程度中高，主要集中在 fully-async 用户群与 `--sglang-config` 多组配置。 - 风险标记：核心路径变更 , 共享引擎暂停未经验证 , 快照生命周期复杂度 , 外部 sglang 依赖 , 跨模块导出重构 , GPU e2e 依赖 CI label # 关联脉络 - PR #1717 Rewrite fully-async rollout as FullyAsyncRolloutFn on the class-based rollout API: 本 PR 的共享引擎评估直接挂接 `FullyAsyncRolloutFn._call_eval()` 的 producer 暂停语义，并复用其类式 API 与状态实例。 - PR #1716 Move fully-async rollout from examples into miles/rollout: fully-async rollout 迁入核心库后，本 PR 在其基础上补充完整的评估链路，二者属于同一功能演进线。