企业官网建设流程全解析

verl 扩展指南：面向奖励函数、Agent Loop、Replay Buffer 与训练/推理引擎的自定义开发实战

【免费下载链接】verlverl/HybridFlow: A Flexible and Efficient RL Post-Training Framework项目地址: https://gitcode.com/GitHub_Trending/ve/verl

verl（HybridFlow）是一个灵活高效的 RL 后训练框架，其设计哲学是"一切皆可扩展"：从奖励函数、工具调用与 Agent 循环，到异步采样的 Replay Buffer 采样策略、训练器钩子与检查点回调，再到底层的推理框架、训练框架乃至硬件芯片，都预留了明确的扩展点。本文以 docs/extend_guide.rst 为主线，结合仓库源码与配置，按"RL 研究者—Agent 框架开发者—训练/推理框架开发者"三条角色路径，系统讲解 verl 的每类扩展接口的继承方式、配置接入方法与底层实现原理，帮助你快速把自定义能力接入 verl。

面向 RL 研究者：从奖励函数到 Agent 行为的自定义

1. 自定义奖励函数

verl 支持四类奖励函数，全部允许用户自定义：

• 基于规则（Rule-based）：数学、代码等带有 ground truth 的奖励，例如 GSM8K 答案匹配、代码执行结果校验；
• 判别式奖励模型（DisRM）：对轨迹/响应输出打分或分类的判别模型；
• 生成式奖励模型（GenRM）：以生成式方式评估响应的奖励模型；
• 混合奖励（Hybrid）：规则奖励 + GenRM/DisRM 的组合。

所有类型的奖励函数都支持用户自定义，详细的实现与接入方式参见 Reward Loop。从源码结构看，奖励计算的流式路径由verl/workers/reward_manager/下的实现承载，用户自定义奖励函数可通过数据集字段（如reward_model）与奖励管理器完成绑定。

2. 自定义工具调用（@function_tool 与 BaseTool）

verl 内置了 ReAct Agent 循环实现ToolAgentLoop（源码见 verl/experimental/agent_loop/tool_agent_loop.py），支持两种工具定义方式：

• 无状态函数式工具：用@function_tool装饰器装饰一个函数，工具即函数签名与 docstring 的自动解析；
• 有状态类式工具：继承BaseTool并实现execute方法，适合需要维护内部状态的工具（如沙箱、搜索服务客户端）。

定义好工具后，通过以下配置项在 rollout 中启用（对应 rollout.yaml 中的agent配置段）：

actor_rollout_ref.rollout.agent.default_agent_loop=tool_agent actor_rollout_ref.rollout.multi_turn.format=hermes # hermes, gpt-oss, qwen3_coder 等 actor_rollout_ref.rollout.multi_turn.function_tool_path=path/to/your_tools.py # 函数式工具路径 actor_rollout_ref.rollout.multi_turn.tool_config_path=path/to/your_tools.yaml # 类式工具路径

其中：

• default_agent_loop：默认使用的 agent loop 名称，未在数据集中显式指定agent_name时生效（源码见 agent_loop.py）；
• multi_turn.format：多轮对话的 chat template 格式，支持hermes、gpt-oss、qwen3_coder等；
• function_tool_path/tool_config_path：工具加载入口。AgentLoopWorker在每个 worker 上通过load_all_tools一次性加载全部工具，每条轨迹直接复用self.tools，避免重复加载开销（源码见 agent_loop.py）。

更多细节参见 Multi-turn Rollout Support、Agent Loop，以及端到端示例 agent_loop_tutorial.ipynb（内含 code sandbox 训练 ReAct agent 的完整流程）。

3. 自定义 Agent Loop：继承 AgentLoopBase

如果ToolAgentLoop无法满足需求，可以继承AgentLoopBase并实现run方法（抽象方法定义见 agent_loop.py）：

class MyAgentLoop(AgentLoopBase): async def run(self, sampling_params: dict[str, Any], **kwargs) -> AgentLoopOutput: """与 LLM server 和外部环境交互，运行 agent loop。 Args: sampling_params (Dict[str, Any]): LLM sampling 参数。 **kwargs: 来自 `verl.utils.dataset.RLHFDataset` 的数据集字段。 Returns: AgentLoopOutput: Agent loop 输出。 """ ...

重要警告：自定义 loop 需要在 TITO（token-in-token-out）模式下自行请求 LLM server，务必遵守一条黄金法则：永远不要对已经解码过的 token 重新编码（never re-encode tokens you've decoded）。

定义完成后，通过配置文件指定自定义 agent loop 类：

actor_rollout_ref.rollout.agent.agent_loop_config_path=path/to/your_agent.yaml

agent_loop_config_path指向一个包含多个 agent loop 配置的 YAML 文件，每个配置是一个 Hydra instantiate 条目，例如（rollout.yaml 中的注释给出了同样格式的示例）：

- name: react_agent _target_: recipe.langgraph_agent.react_agent_loop.ReactAgentLoop tools: ["get_current_temperature"] - name: math_expression _target_: recipe.langgraph_agent.example.math_expression.MathExpressionReactAgentLoop min_terms: 2 max_terms: 6

加载机制上，AgentLoopWorker.__init__会通过OmegaConf.load读取该配置文件，并把每个{name: config}条目写入全局_agent_loop_registry（源码见 agent_loop.py）；真正执行时通过hydra.utils.instantiate实例化 loop，并注入trainer_config、server_manager、tokenizer、processor、dataset_cls、data_config与tools等运行期依赖（源码见 agent_loop.py）。

run方法返回的AgentLoopOutput是 agent loop 的标准化输出契约（定义见 agent_loop.py），包含prompt_ids、response_ids、response_mask（1 表示 LLM 生成 token，0 表示工具响应 token）、response_logprobs、reward_score、num_turns、metrics等字段。AgentLoopWorker._agent_loop_postprocess会随后完成 padding、attention mask/position id 计算、reward score 与 teacher logprobs（在线蒸馏场景）计算，最终聚合为DataProto批数据，因此自定义 loop 只需正确产出这些字段。

4. 异步训练中自定义 Replay Buffer 采样策略

异步训练时，agent 框架把生成的轨迹流式写入TransferQueue，trainer 通过ReplayBuffer（源码见 verl/trainer/ppo/v1/replay_buffer.py）从 TransferQueue 采样一个 batch 用于训练。

verl 提供了默认采样策略，但用户经常需要自定义。做法是继承ReplayBuffer并实现sample方法：

class UserCustomReplayBuffer(ReplayBuffer): def sample(self, global_steps: int, partition_id: str, batch_size: int) -> tuple[KVBatchMeta, dict]: """从 replay buffer 中采样一个 batch。 Args: global_steps (int): 当前训练的全局步数。 partition_id (str): TransferQueue 的分区，如 "train" 或 "val"。 batch_size (int): 批量大小。 Returns: KVBatchMeta: 一个 batch 的数据。 dict: 辅助指标，如 off-policy staleness 统计。 """ ...

然后通过配置接入（ppo_trainer.yaml 中trainer.v1.sampler段的字段与之对应）：

trainer.v1.sampler.custom_sampler.path = "path/to/your/sampler.py" trainer.v1.sampler.custom_sampler.name = "UserCustomReplayBuffer"

该路径/名称机制在默认sample的 docstring 中有明确说明（replay_buffer.py）。

理解默认实现有助于设计自定义策略。默认ReplayBuffer以 TransferQueue 为 KV 存储：key 格式为{uid}_{session_id}_{index}，其中uid是采样 prompt 时自动生成的唯一 id，session_id是 GRPO 组采样的会话 id，index是会话内轨迹序号。除轨迹外，原始 prompt 也以{uid}为 key 存入，并通过statustag 追踪 GRPO 组采样状态：pending（已采样未开始）、running（所有 session 运行中）、finished（全部正常结束）、failure（至少一个 session 失败）。只有finished或failure状态的 prompt 组才会进入终态处理流程（replay_buffer.py）。

另外两个与采样密切相关的内置配置值得了解：

• max_off_policy_threshold（默认 8）与max_off_policy_strategy（drop或wait）：控制轨迹最多可跨越的模型版本数，drop丢弃超限轨迹，wait为"无丢弃"模式、阻塞采样直到达到阈值的在途 prompt 完成；该机制仅对异步 trainer 生效，同步采样为 on-policy，策略为 NO-OP；
• sync_refill_failed_groups（默认False）：是否让同步 trainer 替换没有任何轨迹的失败组（要求gen_batch_size=1）。

同步（ReplayBuffer）与异步（ReplayBufferAsync）采样在 DAPO 组过滤（filter_groups_metric，过滤配置指标在所有轨迹上完全相同的组）、失败组处理与 off-policy 淘汰上的行为差异，体现在_terminal_eviction_reasons的矩阵中（replay_buffer.py）。

5. 自定义同步/异步 trainer 行为：Hook 体系

用户可能需要改变 trainer 的默认行为，例如：

• 过采样（over-sampling）：采样比 batch size 更多的轨迹；
• 动态过滤（dynamic filtering）：过滤掉组内响应全部正确或全部错误的样本（如 DAPO 式过滤）。

verl v1 PPO trainer（源码见 verl/trainer/ppo/v1/ 下的trainer_base.py、trainer_sync.py、trainer_colocate_async.py、trainer_separate_async.py）提供了一组生命周期 Hook：

• on_init_end
• on_train_begin
• on_train_end
• on_validate_begin
• on_validate_end
• on_step_begin
• on_step_end
• on_sample_begin
• on_sample_end

这组 Hook 同样被sync、colocate_async、separate_async三种 trainer 用于改变模型引擎、LLM server 与 checkpoint 引擎的行为，是 v1 PPO 训练流程扩展的核心入口。

6. 在 trainer 检查点时运行自定义代码：CheckpointCallback

子类化verl.trainer.ppo.checkpoint_callback.CheckpointCallback（源码见 verl/trainer/ppo/checkpoint_callback.py）并覆写on_save（语义与 HuggingFacetransformers的TrainerCallback.on_save事件类似），然后把 trainer 指向该类的完整限定名（该类必须在 driver 上可导入）：

trainer.checkpoint_callback_class=my_package.module.MyCheckpointCallback

该 Hook 在 v1 PPO trainer 中每次检查点保存后于 driver 进程运行，完整的 Hook 语义参见 checkpoint.rst。

源码层面需要注意两个行为细节（均记录在 checkpoint_callback.py 的模块 docstring 中）：

• on_save(trainer, global_step, checkpoint_dir, async_save=False, **kwargs)必须在签名中接受**kwargs，以便未来版本传递更多上下文而不破坏用户子类；
• Hook 抛出的异常会向上传播并中止训练——检查点回调通常执行持久性关键副作用（上传分片、注册模型版本等），静默吞掉失败可能导致无信号丢 checkpoint。若需要 best-effort 语义，请在 Hook 体内自行try/except；
• 当async_save=True（Megatron 异步 checkpointing）时，回调触发时 worker 侧写入可能仍在进行、latest_checkpointed_iteration.txt尚未写入，此时不能假定 checkpoint 已持久化。

build_checkpoint_callback在未设置该配置时返回 no-op 回调，因此 trainer 调用点无需判空（checkpoint_callback.py）。

面向 Agent 框架开发者：替换 AgentLoopManager 与接入黑盒 Agent

7. 用自有 Agent 框架替换 AgentLoopManager

AgentLoopManager（源码见 agent_loop.py）是 verl 中 agent 框架的一个参考实现，被设计为可被其他 agent 框架完全替换——例如 NVIDIA Nemo-Gym、AWS Bedrock AgentCore、SWE-agent 等（见 agent_loop.py 的模块 docstring）。接入自有框架只需满足两个要求：

1. 实现一个非阻塞的generate_sequences方法；
2. rollout 结束后把轨迹字段（如prompt_ids、response_ids、response_mask等）写入TransferQueue。

class MyAgentLoopManager: @classmethod @auto_await async def create( cls, config: DictConfig, llm_client: LLMServerClient, teacher_client: dict[str, LLMServerClient] = None, reward_loop_worker_handles: list[ray.actor.ActorHandle] = None, ): """创建 agent loop manager。 Args: config (DictConfig): 主入口的完整配置。 llm_client (LLMServerClient): LLM server 的 client。 teacher_client (dict[str, LLMServerClient]): 多个 teacher server 的 client。 reward_loop_worker_handles (List[ray.actor.ActorHandle]): 流式奖励计算的 actor handles。 """ ... def generate_sequences(self, prompts: TensorDict) -> None: """向 agent 框架提交一批 prompt 用于 rollout，不阻塞。框架应在 rollout 完成后把轨迹字段 （如 prompt_ids, response_ids, response_mask ...）写入 TransferQueue。 Args: prompts (TensorDict): 来自训练或验证数据集的一批 prompt。 """ ...

定义后，通过配置指定自定义 manager 类（注意+前缀表示向配置新增字段）：

+actor_rollout_ref.rollout.agent.agent_loop_manager_class=my_package.module.MyAgentLoopManager

作为参考，内置AgentLoopManager的create类方法会通过auto_await异步创建并初始化一组AgentLoopWorker（默认使用ray.remote(AgentLoopWorker)，数量由rollout.agent.num_workers控制，默认 8），并按 round-robin 策略把 worker 调度到集群中的存活节点上（agent_loop.py）；generate_sequences则把输入 batch 切分后并行分发到各 worker，再聚合结果（agent_loop.py）。

8. 以黑盒方式接入 Claude Code / Codex / Trae 等 Agent

verl 启动了子项目verl-project/uni-agent，其中提供一个 agent gateway，用于把外部 Agent 框架以黑盒方式接入训练，核心能力包括：

• Message API：提供 OpenAI/v1/chat/completions与 Anthropic/v1/messages兼容 API；
• Token-in-token-out：把user, tool消息编码为 token id 并请求 LLM server，把响应 id 解码并将工具解析为assistant消息；
• 轨迹追踪（Trajectory tracking）：消息前缀匹配，前缀变化时派生新轨迹；
• 会话管理（Session management）：支持多个活动会话并发管理。

详细设计可查阅 Agent Gateway 的 RFC 与实现文档（对应 verl 仓库中的 agent_loop.rst 及相关 issue/子项目资料）。

面向训练/推理框架开发者：接入自有引擎与硬件

9. 推理框架开发者：用 VERL_USE_EXTERNAL_MODULES 注册自定义推理框架

verl 提供了环境变量钩子VERL_USE_EXTERNAL_MODULES用于加载外部模块。你可以在自己的模块中定义注册钩子，并设置该环境变量来动态注册自定义模块，主要包括两类扩展点：

• RolloutReplica：自定义 rollout replica 类，决定如何启动你自己的推理 server；
• ServerAdapter：自定义 server adapter 类，决定如何用你自己的推理 server 更新权重。

verl-project/vexact就是通过这种方式与 verl 集成的。vexact 在register.py中定义了注册钩子：

def _load_vexact_replica(): """Lazy loader for VeXactReplica to avoid circular imports.""" from vexact.integrations.verl.async_server import VeXactReplica return VeXactReplica # Register VeXact rollout replica (for server mode) RolloutReplicaRegistry.register("vexact", _load_vexact_replica) # Register VeXact rollout base (for hybrid mode with device mesh) _ROLLOUT_REGISTRY[("vexact", "async")] = "vexact.integrations.verl.rollout.ServerAdapter"

用户只需设置环境变量即可加载 vexact：

export VERL_USE_EXTERNAL_MODULES=vexact.integrations.verl.register

10. 训练框架开发者：继承 BaseEngine 接入自有训练框架

verl 提供了统一的训练引擎抽象BaseEngine（源码见 verl/workers/engine/base.py）。基于该抽象，verl 原生支持多个主流训练框架：

训练框架开发者只需继承BaseEngine并实现全部接口，然后与推理框架一样通过VERL_USE_EXTERNAL_MODULES在 verl 中注册自己的训练引擎。

BaseEngine的接口设计（base.py）覆盖了训练引擎的全生命周期：

• 初始化：initialize()实例化或加载模型、优化器与 LR scheduler；
• 模式切换：train_mode()/eval_mode()上下文管理器（配合BaseEngineCtx完成参数/优化器的设备 offload 与回载）；
• 训练步骤：optimizer_zero_grad()、forward_backward_batch()、optimizer_step()、lr_scheduler_step()；train_batch()在基类中已经组合好"zero-grad → forward/backward → step → 记录 grad_norm"的标准流程；
• 推理：infer_batch()在torch.no_grad()下执行前向；
• 参数导出：get_per_tensor_param()/get_per_tensor_param_shard()/get_per_tensor_param_delta_shard()，分别支持全量导出、本地分片导出与 delta 增量导出（供delta_shardedcheckpoint 后端使用）；
• 并行信息：get_data_parallel_size()/get_data_parallel_rank()/get_data_parallel_group()；
• 检查点：save_checkpoint()/load_checkpoint()；
• 设备与 LoRA：to()、disable_adapter()。

引擎注册机制由EngineRegistry提供（base.py）：register(model_type, backend, device, vendor)装饰器按model_type → backend → (device, vendor)三级键注册引擎类，new()依据当前设备与厂商动态实例化；还支持VERL_ENGINE_DEVICE/VERL_ENGINE_VENDOR环境变量覆盖自动探测结果。

例如verl-hardware-plugin（FlagOS）在__init__.py中定义注册钩子：

from verl_hardware_plugin.engines import register_all_engines from verl_hardware_plugin.platforms import register_all_platforms register_all_platforms() register_all_engines()

用户同样通过环境变量加载：

export VERL_USE_EXTERNAL_MODULES=verl_hardware_plugin

关于训练引擎的更多设计，参见 Model Engine。

11. 硬件厂商：接入自有芯片

verl 主分支原生支持 NVIDIA GPU、华为 Ascend NPU 与 AMD GPU，并提供统一的插件系统支持其他硬件平台。硬件接入的完整说明参见：

• Multi-chip Support
• 外部硬件插件verl-project/verl-hardware-plugin（面向 MLU、XPU、MetaX 等芯片）

从仓库结构看，硬件抽象还体现在平台插件层 verl/plugin/platform/（含 6 个 Python 模块），与verl/utils/device.py中的设备/厂商探测逻辑配合，实现设备无关的引擎选择。

总结：verl 扩展点全景

把上述扩展点归纳为一张速查表：

verl 的扩展设计遵循一致的模式：抽象基类定义契约（AgentLoopBase、ReplayBuffer、CheckpointCallback、BaseEngine），注册表或配置文件完成绑定（_agent_loop_registry、EngineRegistry、agent_loop_config_path、custom_sampler），环境变量或命令行参数实现运行时装载（VERL_USE_EXTERNAL_MODULES、checkpoint_callback_class）。无论你是 RL 研究者、Agent 框架开发者还是训练/推理框架开发者，都可以在不改动 verl 核心代码的前提下，沿着上述扩展点把自有实现无缝接入训练流程。

【免费下载链接】verlverl/HybridFlow: A Flexible and Efficient RL Post-Training Framework项目地址: https://gitcode.com/GitHub_Trending/ve/verl