如何在verl中加入自定义奖励函数？

如何在verl中加入自定义奖励函数？

1. 引言

1.1 业务场景描述

在大型语言模型（LLM）的后训练过程中，强化学习（Reinforcement Learning, RL）已成为提升模型行为对齐能力的重要手段。而奖励函数作为RL训练的核心组成部分，直接决定了策略优化的方向和质量。标准的奖励机制往往难以满足特定任务的需求，例如数学推理、代码生成或复杂多轮对话等场景，需要结合领域知识设计更加精细化的评估逻辑。

verl是一个专为LLM后训练设计的高效强化学习框架，由字节跳动火山引擎团队开源，支持灵活扩展的RL算法与模块化组件集成。其核心优势之一是允许用户通过继承和重写接口来自定义关键模块，包括奖励计算逻辑。这使得开发者可以在不修改框架底层代码的前提下，快速实现面向具体任务的奖励函数。

1.2 痛点分析

现有通用奖励模型（如基于RM的打分）存在以下局限性：

• 缺乏细粒度控制：无法针对特定语义结构进行精准评分。
• 成本高：依赖额外的奖励模型推理，增加计算开销。
• 可解释性差：黑盒式打分难以调试与迭代。

因此，在实际项目中，我们常常需要引入规则驱动、工具调用辅助或混合式的自定义奖励函数，以提高训练效率和策略收敛质量。

1.3 方案预告

本文将详细介绍如何在verl框架中实现一个可插拔的自定义奖励函数，涵盖从交互系统设计、类继承实现到配置加载的完整流程，并以“数学解题正确性判断”为例展示实战步骤。最终目标是让读者掌握一套标准化的方法论，用于构建适用于各类任务的定制化奖励机制。

2. 技术方案选型

2.1 verl中的奖励机制架构

verl的奖励计算主要发生在两个层面：

1. Interaction 层：通过BaseInteraction子类的calculate_score()方法返回单次交互的奖励值；
2. Reward Model 层：可选地接入外部奖励模型（如基于对比学习的RM），但非必需。

对于轻量级、确定性的奖励逻辑（如格式校验、数值比对），推荐使用Interaction 层自定义实现，避免引入额外模型开销。

2.2 自定义方式对比

结论：若奖励逻辑可通过明确规则判定（如答案匹配、语法检查），应优先选择继承BaseInteraction类的方式，具备最高灵活性与最低延迟。

3. 实现步骤详解

3.1 环境准备

确保已安装 verl 并验证版本：

python -c " import verl print(f'verl version: {verl.__version__}') "

输出示例：

verl version: 0.1.0

3.2 定义自定义 Interaction 类

我们将创建一个名为MathReasoningInteraction的类，用于处理数学问题求解任务中的奖励计算。

from verl import BaseInteraction from typing import Optional, Dict, Any, Tuple import re class MathReasoningInteraction(BaseInteraction): def __init__(self, config: Dict[str, Any]): super().__init__(config) self._instance_dict = {} # 存储每个 instance 的上下文信息 async def start_interaction(self, instance_id: Optional[str] = None, **kwargs) -> str: """初始化交互实例""" if instance_id not in self._instance_dict: self._instance_dict[instance_id] = { "query": kwargs.get("query", ""), "ground_truth": kwargs.get("ground_truth", ""), "response": "", } return instance_id async def generate_response( self, instance_id: str, messages: list[dict], **kwargs ) -> Tuple[bool, str, float, Dict[Any, Any]]: """生成响应并计算奖励""" content = "" # 提取助手最后一轮回复 for i in range(len(messages) - 1, -1, -1): if messages[i].get("role") == "assistant": content = messages[i].get("content", "") break self._instance_dict[instance_id]["response"] = content reward = await self.calculate_score(instance_id) # 根据奖励决定是否终止对话 should_terminate_sequence = reward == 1.0 feedback_msg = ( "✅ 正确！你的回答完全匹配标准答案。" if should_terminate_sequence else "❌ 不完全正确，请重新思考。" ) return should_terminate_sequence, feedback_msg, reward, {} async def calculate_score(self, instance_id: str) -> float: """根据响应内容与真实答案计算奖励分数""" entry = self._instance_dict.get(instance_id) if not entry or not entry["response"] or not entry["ground_truth"]: return 0.0 response = entry["response"] ground_truth = entry["ground_truth"] # 提取数字答案（支持多种格式） pred_answer = self._extract_number(response) true_answer = self._extract_number(ground_truth) if pred_answer is None or true_answer is None: return 0.0 return 1.0 if abs(pred_answer - true_answer) < 1e-6 else 0.0 def _extract_number(self, text: str) -> Optional[float]: """从文本中提取最后一个出现的浮点数""" matches = re.findall(r"[-+]?\d*\.\d+|\d+", text.replace(",", "")) return float(matches[-1]) if matches else None async def finalize_interaction(self, instance_id: str) -> None: """清理交互实例""" if instance_id in self._instance_dict: del self._instance_dict[instance_id]

关键点说明：

• start_interaction：接收query和ground_truth初始化上下文；
• generate_response：调用calculate_score获取奖励；
• calculate_score：实现核心奖励逻辑——提取预测值并与真值比对；
• _extract_number：正则提取最后一个数字，适应自然语言输出格式。

3.3 注册并配置自定义类

在训练配置文件中注册该类路径，使verl能够动态加载。

创建config/interaction_config/math_interaction.yaml：

interaction: class_name: "your_module.interactions.MathReasoningInteraction" config: {}

⚠️ 注意：请将your_module.interactions替换为实际 Python 模块路径，确保该模块在 PYTHONPATH 中。

3.4 数据预处理：注入 ground truth

训练数据需包含extra_info.interaction_kwargs字段传递必要参数。

# 示例数据构造 question = "小明有5个苹果，吃了2个，还剩几个？" solution = "3" data = { "prompt": [ {"role": "user", "content": question} ], "extra_info": { "interaction_kwargs": { "query": question, "ground_truth": solution } } }

此字段将在start_interaction调用时传入，完成上下文初始化。

3.5 训练脚本调用

启动训练时指定 interaction 配置路径：

python3 -m verl.trainer.main_ppo \ data.train_batch_size=256 \ data.max_prompt_length=512 \ data.max_response_length=512 \ actor_rollout_ref.model.path=meta-llama/Llama-3.1-8B-Instruct \ actor_rollout_ref.rollout.name=sglang \ algorithm.adv_estimator=grpo \ +interaction_config_path=./config/interaction_config/math_interaction.yaml

4. 实践问题与优化

4.1 常见问题及解决方案

4.2 性能优化建议

✅ 启用梯度检查点减少显存占用

actor_rollout_ref: model: enable_gradient_checkpointing: True

✅ 使用序列打包提升吞吐

data: return_raw_chat: True

✅ 控制最大对话轮次防止无限循环

actor_rollout_ref: rollout: multi_turn: enable: True max_assistant_turns: 3

✅ 添加缓存机制加速重复查询

对高频出现的问题（如测试集题目），可在calculate_score中加入哈希缓存：

from functools import lru_cache @lru_cache(maxsize=1000) def cached_compare(pred: str, truth: str) -> float: pred_num = _extract_number(pred) truth_num = _extract_number(truth) return 1.0 if pred_num == truth_num else 0.0

5. 扩展应用：结合工具调用的复合奖励

当任务更复杂时（如涉及单位换算、公式推导），可结合SandboxFusion工具执行结果来计算奖励。

async def calculate_score_with_tool(self, instance_id: str) -> float: response = self._instance_dict[instance_id]["response"] # 调用代码解释器执行表达式 tool_result = await self.code_interpreter.execute( instance_id=instance_id, parameters={"code": f"result = {response}; result"} ) if tool_result.status == "error": return 0.0 exec_result = float(tool_result.text.strip()) return 1.0 if abs(exec_result - truth_value) < 1e-6 else 0.5 # 部分奖励

这种方式可用于开放域数学、物理建模等任务，显著提升奖励信号的准确性。

6. 总结

6.1 实践经验总结

本文详细介绍了如何在verl框架中实现自定义奖励函数，核心要点如下：

• 利用BaseInteraction接口扩展，实现灵活的奖励逻辑；
• 通过interaction_kwargs在数据层注入上下文信息；
• 结合正则解析、工具调用等方式提升奖励精度；
• 配置管理与模块路径规范是成功集成的关键。

6.2 最佳实践建议

1. 优先使用规则奖励：对于确定性任务，避免过度依赖RM模型；
2. 保持奖励函数无副作用：不要在calculate_score中修改全局状态；
3. 做好日志追踪：记录response与reward对应关系，便于调试分析。

通过上述方法，开发者可以高效构建面向特定任务的高质量奖励系统，显著提升 LLM 在强化学习训练中的表现。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。