如何在verl中自定义reward函数？附完整示例-平芜编程栈

如何在verl中自定义reward函数？附完整示例

强化学习训练中，reward函数是驱动模型行为演化的“指挥棒”。在LLM后训练场景下，一个设计得当的reward函数，往往比算法本身更能决定最终效果的上限。verl作为专为大语言模型后训练打造的高效RL框架，将reward函数的定义与集成做到了极致简洁——它不强制你写复杂的类继承或注册机制，而是通过清晰、可插拔的函数式接口，让你专注业务逻辑本身。

本文将带你从零开始，在verl中实现一个真正可用、可调试、可复用的自定义reward函数。我们不讲抽象概念，不堆砌配置项，只聚焦三件事：它为什么重要、它长什么样、你如何把它跑起来。无论你是刚接触RL的新手，还是正在调优线上模型的工程师，都能从中获得可立即落地的实践路径。

1. 为什么必须自定义reward函数？

在verl的典型训练流程中（尤其是GRPO这类无Reward Model的变体），reward函数不是可选项，而是整个训练循环的核心计算节点。它直接决定了：

每个token生成动作的价值判断
Advantage信号的原始来源
Actor策略更新的方向与强度

官方默认的reward函数（如rule_based_reward）仅提供基础模板，例如简单匹配关键词或长度惩罚。但真实业务场景远比这复杂：

电商客服回复需同时满足准确性（是否正确解答用户问题）、安全性（是否含违规话术）、服务性（是否主动提供解决方案）
内容创作助手需平衡信息密度（关键点是否覆盖）、可读性（句子是否通顺自然）、风格一致性（是否符合品牌口吻）
代码生成模型需评估语法正确性、逻辑完整性（是否能编译运行）、注释质量（是否解释了关键步骤）

这些维度无法靠单一规则穷举，更不能依赖黑盒Reward Model带来的延迟与不确定性。此时，一个可编程、可分段调试、可与业务系统对接的自定义reward函数，就成了提升模型表现最直接、最可控的杠杆。

2. verl中reward函数的本质与位置

在verl架构中，reward函数不是一个独立模块，而是一个被明确注入到训练主干中的纯函数（pure function）。它的本质非常朴素：接收一个batch数据，返回一个与之形状对齐的reward张量。

2.1 函数签名与输入结构

自定义reward函数必须严格遵循以下签名：

def reward_fn(batch: DataProto) -> torch.Tensor: """ Args: batch: 包含rollout生成结果的数据容器，关键字段包括： - batch['prompt_token_ids']: prompt部分的token ID序列 (B, L_prompt) - batch['response_token_ids']: model生成的response token ID序列 (B, L_response) - batch['prompt_attention_mask']: prompt的attention mask (B, L_prompt) - batch['response_attention_mask']: response的attention mask (B, L_response) - batch['log_probs']: 每个token的log概率 (B, L_response) - batch['prompt_lengths']: 每条prompt的实际长度 (B,) - batch['response_lengths']: 每条response的实际长度 (B,) Returns: torch.Tensor: token-level reward张量，shape为 (B, L_response) """ pass

关键理解：verl默认采用token-level reward，而非sequence-level。这意味着你的函数需要为response中每一个token输出一个标量分数。这为精细化控制（如对关键动词加权、对结尾句号降权）提供了可能。

2.2 在训练流程中的执行时机

从ray_trainer.py的源码可见，reward函数在_step方法中被明确调用：

# 在advantage计算前，reward_fn是唯一的数据加工入口 reward_tensor = self.reward_fn(batch) # ← 就在这里！ batch.batch['token_level_scores'] = reward_tensor # 后续所有advantage、loss计算都基于此 batch = compute_advantage(batch, ...)

这个位置至关重要：它位于old_log_prob和ref_log_prob计算之后，但在任何梯度更新之前。这意味着你可以安全地访问所有中间状态（如log概率、token ID、mask），并基于它们做出综合判断，而无需担心破坏训练稳定性。

3. 从零构建一个实用的自定义reward函数

我们以一个典型的技术文档问答助手场景为例，构建一个兼顾准确性、安全性和可读性的reward函数。目标是让模型在回答“如何在Linux中查看磁盘使用率？”时，不仅给出正确命令（df -h），还要避免泄露敏感信息，并用清晰的中文组织语言。

3.1 基础骨架：函数定义与数据准备

首先，创建一个独立的Python文件（如my_reward.py），定义函数主体：

# my_reward.py import torch import re from typing import List, Dict, Any from verl.data.data_proto import DataProto def reward_fn(batch: DataProto) -> torch.Tensor: """技术文档问答场景下的多维度reward函数""" # 1. 提取关键数据 prompts = batch.batch['prompts'] # List[str], 长度为B responses = batch.batch['responses'] # List[str], 长度为B response_token_ids = batch.batch['response_token_ids'] # (B, L_response) response_mask = batch.batch['response_attention_mask'] # (B, L_response) # 2. 初始化reward张量，初始值为0 B, L = response_token_ids.shape reward_tensor = torch.zeros(B, L, dtype=torch.float32, device=response_token_ids.device) # 3. 对每个样本进行逐条处理 for i in range(B): prompt = prompts[i] response = responses[i] mask = response_mask[i] # (L,) # 计算该样本的token-level reward token_rewards = _compute_single_sample_reward(prompt, response, mask) reward_tensor[i, :len(token_rewards)] = token_rewards return reward_tensor

这个骨架完成了三件事：安全提取数据、初始化输出张量、遍历样本。接下来，我们填充核心逻辑_compute_single_sample_reward。

3.2 核心逻辑：多维度评分与融合

我们将reward拆解为三个正交维度，分别计算后再加权融合。这种设计便于后期单独调整某一项权重，也利于调试：

def _compute_single_sample_reward(prompt: str, response: str, mask: torch.Tensor) -> List[float]: """为单个prompt-response对计算token-level reward""" # Step 1: 基于规则的准确性打分（Accuracy Score） acc_score = _accuracy_score(prompt, response) # Step 2: 安全性检查（Safety Score） safety_score = _safety_score(response) # Step 3: 可读性打分（Readability Score） readability_score = _readability_score(response) # Step 4: 融合三者，生成token-level reward # 策略：将总分均匀分配给每个有效token，但对关键token（如命令、动词）额外加成 tokens = response.split() token_rewards = [0.0] * len(tokens) if len(tokens) == 0: return [0.0] # 基础分：总分平均分配 base_reward = (acc_score + safety_score + readability_score) / len(tokens) for j in range(len(tokens)): token_rewards[j] = base_reward # 关键词加成：识别并增强关键token的reward keyword_bonus = _keyword_bonus(tokens) for j, bonus in enumerate(keyword_bonus): if j < len(token_rewards): token_rewards[j] += bonus # 过滤掉padding位置的reward（对应mask为0的位置） valid_rewards = [] for j, m in enumerate(mask.tolist()): if j < len(token_rewards) and m == 1: valid_rewards.append(token_rewards[j]) return valid_rewards def _accuracy_score(prompt: str, response: str) -> float: """检查response是否包含正确技术要点""" # 简化版：检查是否包含'df'和'-h'（Linux磁盘命令） if 'df' in response and '-h' in response: return 1.0 elif 'df' in response: return 0.7 else: return 0.0 def _safety_score(response: str) -> float: """检查是否存在高风险内容""" # 禁止词列表（实际应使用更完善的过滤器） dangerous_patterns = [ r'rm\s+-rf', r'chmod\s+777', r'echo\s+.*\s+>>\s+/etc/passwd', r'wget\s+http://.*\.sh' ] for pattern in dangerous_patterns: if re.search(pattern, response, re.IGNORECASE): return -5.0 # 严重惩罚 # 允许合理建议，但禁止绝对化指令 if '必须' in response or '一定要' in response: return 0.5 return 1.0 def _readability_score(response: str) -> float: """基于简单启发式评估可读性""" # 句子数量（鼓励分点说明） sentences = re.split(r'[。！？；]+', response) sentences = [s.strip() for s in sentences if s.strip()] # 长度适中（过短信息不足，过长易混乱） word_count = len(response.split()) if 10 <= word_count <= 50: length_score = 1.0 elif word_count < 5: length_score = 0.3 else: length_score = 0.6 # 分点结构加分 bullet_points = len(re.findall(r'^\s*[-•●]\s+', response, re.MULTILINE)) structure_score = min(1.0, bullet_points * 0.3) return (length_score + structure_score) / 2 def _keyword_bonus(tokens: List[str]) -> List[float]: """为关键token提供额外reward加成""" bonus = [0.0] * len(tokens) # 命令关键词（如df, ls, cd） command_keywords = {'df', 'ls', 'cd', 'ps', 'top', 'grep', 'awk', 'sed'} # 动词关键词（如显示、查看、列出、检查） verb_keywords = {'显示', '查看', '列出', '检查', '获取', '运行'} for i, token in enumerate(tokens): clean_token = token.strip('.,!?;:"()[]{}') if clean_token.lower() in command_keywords: bonus[i] = 0.5 elif clean_token in verb_keywords: bonus[i] = 0.3 return bonus

这段代码展示了verl reward函数的典型范式：纯Python逻辑 + 显式数据操作 + token粒度控制。它完全避开了框架内部的复杂API，你只需关注业务规则本身。

3.3 集成到verl训练配置

要让verl使用你的函数，只需两步：

步骤1：修改PPO训练配置文件（`ppo_trainer.yaml`）

在trainer部分添加reward_fn路径：

trainer: # ... 其他配置保持不变 reward_fn: "my_reward:reward_fn" # 格式：模块名:函数名 # 可选：为验证阶段指定不同的reward函数 val_reward_fn: "my_reward:reward_fn"

步骤2：确保模块可导入

将my_reward.py放在训练脚本同级目录，或将其所在目录加入PYTHONPATH。verl会自动通过importlib动态加载。

验证技巧：在函数开头添加print(f"Reward function called for prompt: {prompt[:30]}...")，运行训练时观察日志，即可确认函数是否被正确调用。

4. 调试与优化实战技巧

一个写得漂亮的reward函数，不等于一个效果好的reward函数。以下是我们在真实项目中总结出的调试黄金法则：

4.1 日志可视化：让reward“看得见”

在reward_fn中添加日志，记录每一步的计算结果：

def reward_fn(batch: DataProto) -> torch.Tensor: # ... 数据提取 ... # 添加调试日志（仅在rank 0上打印，避免刷屏） if torch.distributed.get_rank() == 0: print(f"[DEBUG] Prompt: '{prompts[0][:50]}...' | Response: '{responses[0][:50]}...' | " f"Acc: {acc_score:.2f} | Safety: {safety_score:.2f} | Read: {readability_score:.2f}") # ... 计算reward ... return reward_tensor

配合tensorboard，你还可以将reward统计信息写入日志：

from torch.utils.tensorboard import SummaryWriter writer = SummaryWriter(log_dir="./logs/reward_debug") def reward_fn(batch: DataProto) -> torch.Tensor: # ... 计算reward_tensor ... if torch.distributed.get_rank() == 0: writer.add_histogram("reward/token_level", reward_tensor.flatten(), global_step=global_step) writer.add_scalar("reward/mean_per_batch", reward_tensor.mean().item(), global_step=global_step) return reward_tensor

4.2 渐进式开发：从简单到复杂

切忌一上来就写一个“完美”的reward函数。推荐按以下顺序迭代：

Baseline（基线）：只返回一个常数（如全1.0）。验证训练流程是否通畅。
Accuracy Only（仅准确性）：只实现_accuracy_score，观察模型是否开始收敛到正确答案。
Add Safety（加入安全性）：引入safety_score，观察违规回复是否显著减少。
Refine Readability（精调可读性）：逐步增加readability_score的复杂度，观察生成文本的流畅度变化。

每次迭代后，用verl的eval工具对比不同版本的reward函数在验证集上的表现，用数据驱动决策。

4.3 性能陷阱规避

虽然reward函数是CPU计算，但不当写法会成为性能瓶颈：

❌避免在循环内做IO操作：如每次调用都读取文件、请求API。
❌避免正则表达式过度回溯：对超长response使用re.compile()预编译模式。
利用向量化操作：对于批量相似计算（如字符串长度），用torch.tensor([len(r) for r in responses])替代Python循环。
缓存重复计算：如果多个样本共享相同prompt，可对prompt哈希后缓存其分析结果。

5. 高级用法：与外部系统联动

verl的reward函数设计天然支持与外部系统集成，这是其区别于其他框架的关键优势。

5.1 调用本地LLM进行细粒度评估

当你需要超越规则的语义理解时，可以调用一个轻量级本地LLM（如Phi-3-mini）作为reward judge：

# 在reward_fn顶部初始化一次 _judge_model = None _judge_tokenizer = None def _init_judge_model(): global _judge_model, _judge_tokenizer if _judge_model is None: from transformers import AutoModelForSequenceClassification, AutoTokenizer _judge_model = AutoModelForSequenceClassification.from_pretrained( "microsoft/phi-3-mini-4k-instruct", torch_dtype=torch.bfloat16 ).to("cuda") _judge_tokenizer = AutoTokenizer.from_pretrained("microsoft/phi-3-mini-4k-instruct") def _llm_judge_score(prompt: str, response: str) -> float: _init_judge_model() # 构造评估提示（Prompt Engineering） eval_prompt = f"""请评估以下AI回复的质量（1-5分）： Prompt: {prompt} Response: {response} 请只输出一个数字，不要任何解释。""" inputs = _judge_tokenizer(eval_prompt, return_tensors="pt").to("cuda") with torch.no_grad(): outputs = _judge_model(**inputs) score = torch.softmax(outputs.logits, dim=-1)[0, 4].item() # 假设5分对应最高logit return score * 5.0 # 映射到1-5分

注意：此方案会显著增加单步耗时，建议仅在小批量调试或离线评估时使用。生产环境应替换为更高效的模型或缓存机制。

5.2 对接业务数据库进行实时校验

对于需要强一致性的场景（如金融问答），reward函数可查询内部知识库：

def _database_check_score(response: str) -> float: """查询公司内部知识库，验证技术细节准确性""" try: # 伪代码：实际使用SQL或API db_result = query_knowledge_base("linux_disk_usage_command") if db_result and response.strip() in db_result.get("valid_commands", []): return 2.0 else: return -1.0 except Exception as e: logger.warning(f"DB check failed: {e}") return 0.0

这种能力让verl的reward函数真正成为连接AI模型与企业核心资产的桥梁。

6. 总结：掌握reward，就是掌握模型进化方向

在verl中自定义reward函数，本质上是一次从“调参”到“定义价值”的思维跃迁。它不再是你被动适应框架的约束，而是你主动塑造模型行为的画笔。

回顾本文，我们完成了：

理解本质：明确了reward函数在verl训练流中的核心地位与函数签名；
动手实践：构建了一个具备准确性、安全性、可读性三重维度的完整reward函数；
工程落地：掌握了从配置集成、日志调试到性能优化的全流程；
能力延伸：探索了与本地LLM、业务数据库等外部系统的深度联动方式。

记住，最好的reward函数，永远不是最复杂的那个，而是最贴近你业务目标、最容易被团队理解和维护的那个。现在，打开你的编辑器，把第一个print("Hello, Reward!")写进reward_fn里吧——模型进化的下一章，由你执笔。

7. 下一步：从reward到完整RL工作流

掌握了reward函数，你已站在了verl RL训练的中枢。接下来，你可以：

尝试将reward函数与val_reward_fn结合，构建A/B测试框架，科学评估不同reward策略的效果；
探索verl的GRPO与PPO算法切换，理解reward函数在有/无Critic模型下的不同作用机制；
结合verl的vLLMrollout引擎，将reward计算与高速推理流水线深度耦合，实现毫秒级反馈。

真正的强化学习，始于一个精心设计的reward。

--- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

如何在verl中自定义reward函数？附完整示例