`reward_loop.py` — 这是奖励循环系统的核心文件¶

文件路径: verl/experimental/reward_loop/reward_loop.py

文件概述¶

这是奖励循环系统的核心文件，包含三个主要部分： 1. migrate_legacy_reward_impl - 配置迁移函数，将旧版奖励配置格式迁移为新版 2. RewardLoopWorker - 奖励计算工作器，支持规则奖励、模型奖励、自定义奖励 3. RewardLoopManager - 奖励循环管理器，创建和管理 Ray worker，控制奖励模型生命周期

一、配置迁移函数 migrate_legacy_reward_impl¶

这个函数用于向后兼容，把旧版的扁平配置结构迁移到新版的嵌套配置结构。

def migrate_legacy_reward_impl(config):
    """
    Migrate the legacy reward model implementation to the new one.
    """
    # 1. reward workers migration
    # config.reward_model.num_workers -> config.reward.num_workers
    if config.reward_model.num_workers is not None:
        config.reward.num_workers = config.reward_model.num_workers

    # 2. reward manager migration
    # config.reward_model.reward_manager -> config.reward.reward_manager
    if config.reward_model.reward_manager is not None:
        config.reward.reward_manager.name = config.reward_model.reward_manager

迁移涉及5个方面： - 奖励 worker 数量配置 - 奖励管理器配置 - 自定义奖励函数配置 - 奖励模型配置（GPU数量、路径等） - sandbox_fusion 配置

二、RewardLoopWorker - 奖励计算工作器¶

初始化¶

class RewardLoopWorker:
    def __init__(self, config, reward_tokenizer, val_reward_tokenizer, ...):
        self.config = config
        self.reward_tokenizer = reward_tokenizer
        self.val_reward_tokenizer = val_reward_tokenizer

        # 加载奖励管理器（核心组件）
        self.reward_manager = load_reward_manager(config, ...)

Worker 在初始化时会通过 load_reward_manager 加载合适的奖励管理器（如 NaiveRewardManager 或 DAPORewardManager），具体使用哪个取决于配置。

奖励计算方法 compute_reward¶

这是最核心的方法，根据配置决定使用哪种奖励计算方式：

def compute_reward(self, data: DataProto, ...):
    # 1. 规则奖励（Rule-based）
    if self.config.reward.get("custom_reward_function"):
        reward_tensor = self.reward_manager(data, ...)

    # 2. 判别式奖励模型（Discriminative RM）
    elif self.config.reward.reward_model.enable:
        reward_tensor = self._compute_rm_score(data, ...)

    # 3. 生成式奖励模型（Generative RM）
    elif self.config.reward.reward_model.rollout.enable:
        reward_tensor = await self._compute_generative_rm_score(data, ...)

    # 4. 自定义奖励（Custom）
    else:
        reward_tensor = self.reward_manager(data, ...)

生成式奖励模型的异步 HTTP 调用¶

对于需要 LLM 推理的生成式奖励（如用 GPT 打分），Worker 通过异步 HTTP 请求与 LLM 服务通信：

async def _compute_generative_rm_score(self, data, ...):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for prompt in prompts:
            task = self._send_request(session, prompt, ...)
            tasks.append(task)
        results = await asyncio.gather(*tasks)

三、RewardLoopManager - 奖励循环管理器¶

职责¶

Manager 负责整个奖励计算系统的生命周期管理：

class RewardLoopManager:
    def __init__(self, config, resource_pool, ...):
        self.config = config
        self.reward_model_manager = None

        # 创建 Ray worker
        self._create_reward_workers(resource_pool)

        # 如果启用了奖励模型，初始化模型管理器
        if config.reward.reward_model.enable:
            self.reward_model_manager = RewardModelManager(config, ...)

奖励模型的唤醒/休眠机制¶

为了节省 GPU 资源，Manager 支持在不需要时休眠奖励模型，需要时再唤醒：

def wake_up(self):
    """唤醒奖励模型到 GPU"""
    if self.reward_model_manager:
        self.reward_model_manager.wake_up()

def sleep(self):
    """将奖励模型移到 CPU 以释放 GPU 显存"""
    if self.reward_model_manager:
        self.reward_model_manager.sleep()

核心类/函数列表¶

名称	类型	说明
`migrate_legacy_reward_impl`	函数	配置格式迁移（旧版 -> 新版）
`RewardLoopWorker`	类	执行具体的奖励计算逻辑
`RewardLoopWorker.compute_reward`	方法	根据配置分发到不同奖励计算方式
`RewardLoopManager`	类	管理奖励计算系统的生命周期
`RewardLoopManager.wake_up/sleep`	方法	控制奖励模型的 GPU 占用

与其他模块的关系¶

依赖 reward_manager/ 子模块提供具体的奖励管理器实现
依赖 reward_model.py 的 RewardModelManager 管理奖励模型服务
被训练器（如 RayPPOTrainer）在训练循环中调用
使用 aiohttp 进行异步 HTTP 通信（与 LLM 推理服务交互）

小结¶

reward_loop.py 是奖励计算的调度中心。Worker 负责"做事"（计算奖励），Manager 负责"管理"（创建 Worker、控制模型生命周期）。这种分层设计使得奖励计算可以灵活支持多种模式：规则奖励、判别式模型打分、生成式 LLM 打分等。

reward_loop.py — 这是奖励循环系统的核心文件¶