奖励函数怎么写？verl自定义奖励实战教学

奖励函数怎么写？verl自定义奖励实战教学

强化学习训练大语言模型，最关键的不是算法本身，而是——你给模型的反馈是否真实、合理、可执行。在RLHF（基于人类反馈的强化学习）中，奖励函数就是那个“裁判”，它不教模型怎么思考，但它清楚地告诉模型：这句话写得好不好、这个推理步骤对不对、这张图生成得像不像。写错奖励函数，再强的PPO也会学偏；写好奖励函数，哪怕用最基础的REINFORCE，也能训出靠谱结果。

但现实是：很多工程师卡在第一步——不知道奖励函数该怎么写。查文档只看到reward_fn参数，点进去发现是个空函数；看示例代码全是调用现成的HFRewardModel，可自己业务场景里没有现成模型怎么办？要不要从头训一个奖励模型？要的话数据怎么构造？不要的话纯规则又怕太死板……

这篇文章不讲理论推导，不堆公式，也不复刻论文。我们直接打开verl框架，用一个真实可运行的数学推理任务（GSM8K风格），手把手带你：

• 从零写出一个可验证、可调试、可上线的自定义奖励函数
• 理解verl中奖励函数的输入结构、调用时机和返回约束
• 避开三个新手必踩的坑：token级reward错位、batch维度混乱、reward缩放失衡
• 最后给出一套渐进式奖励设计方法论：从规则起步 → 加入轻量模型打分 → 引入可验证逻辑 → 对齐人工偏好

全程代码可复制、命令可粘贴、效果可验证。你不需要提前装好集群，一台带GPU的开发机就能跑通。

1. 先搞清verl里奖励函数到底长什么样

在verl中，奖励函数不是黑盒API，而是一个明确签名、严格契约的Python函数。它被设计成可插拔、可组合、可独立测试的单元。理解它的接口，是写好reward的第一步。

1.1 verl奖励函数的标准签名

verl要求所有自定义奖励函数必须符合以下签名：

def reward_fn( batch: Dict[str, torch.Tensor], model_outputs: Dict[str, torch.Tensor], tokenizer: PreTrainedTokenizerBase, **kwargs ) -> torch.Tensor: """ Args: batch: 包含原始prompt、chosen response等的字典，key通常为'prompt', 'chosen' model_outputs: 模型前向输出，含logits、attention_mask等 tokenizer: 用于解码、定位token位置 **kwargs: 额外配置，如reward_weight、debug_mode等 Returns: reward: shape = (batch_size,) 的一维Tensor，每个元素对应一条样本的标量reward """ pass

注意三个硬性要求：

• 输入必须是字典，不能是list或tuple：verl内部按key取值，比如batch['prompt']必须存在
• 返回必须是float32类型的一维Tensor：shape为(N,)，不能是scalar、list、numpy array或二维Tensor
• reward值必须是标量（每条样本一个数）：不支持token-level reward（除非你手动聚合，如取mean/logprob）

这个签名看似简单，但90%的报错都源于违反其中一条。比如把return [1.2, 0.8]写成list，verl会在torch.stack()时崩溃；又比如忘了.to(torch.float32)，混合精度训练会静默失败。

1.2 它在训练流程中何时被调用？

很多人以为reward函数只在PPO step里算一次，其实不然。在verl的HybridFlow架构中，reward函数被两次调用：

这意味着：你的reward函数必须无副作用、幂等、不依赖外部状态。不能在里面写日志文件、不能修改batch原地、不能调用随机数（除非固定seed）。

1.3 verl内置奖励函数长啥样？（以HFRewardModel为例）

verl官方示例中常用HFRewardModel，它本质是加载一个预训练的奖励模型（如OpenAssistant/reward-model-deberta-v3-large）。我们拆解它的核心逻辑，就能反推出自定义函数该怎么做：

# 简化版HFRewardModel.reward_fn逻辑 def hf_reward_fn(batch, model_outputs, tokenizer, reward_model, reward_tokenizer): # 1. 将prompt+response拼成文本 texts = [] for i in range(len(batch['prompt'])): prompt = tokenizer.decode(batch['prompt'][i], skip_special_tokens=True) response = tokenizer.decode(model_outputs['response_ids'][i], skip_special_tokens=True) texts.append(prompt + response) # 2. 用reward tokenizer编码，送入reward model inputs = reward_tokenizer(texts, padding=True, truncation=True, return_tensors="pt") with torch.no_grad(): reward_logits = reward_model(**inputs).logits # shape: (B, 1) # 3. 返回reward值（通常取logits[0]） return reward_logits.squeeze(-1).float() # shape: (B,)

关键启示有三点：

• 它不碰原始logits：只用response_ids解码成文本，再喂给另一个模型——说明reward可以完全脱离Actor的内部表示
• 它做的是“端到端打分”：输入是字符串，输出是标量，中间过程对Actor透明
• 它依赖tokenizer一致性：prompt和response必须用同一个tokenizer解码，否则拼接错位

这给了我们极大自由：你可以用正则匹配关键词、调用外部API打分、运行本地小模型、甚至读Excel查表——只要最终能输出(B,)的Tensor。

2. 动手写第一个自定义奖励函数：数学答案校验器

我们以GSM8K数据集为背景：模型接收一道小学数学题（如“莉莉有5个苹果，吃了2个，还剩几个？”），需输出完整推理链并给出最终答案。理想reward应鼓励：
推理逻辑连贯
最终答案正确（数字匹配）
❌ 乱编答案（如“剩7个”）
❌ 答案藏在中间没标出（如不写“所以答案是3”）

下面这个函数，就是verl生产环境中真实可用的轻量级reward方案：

2.1 代码实现：基于正则的答案提取+数值比对

import re import torch from typing import Dict, Any, Optional from transformers import PreTrainedTokenizerBase def math_answer_reward_fn( batch: Dict[str, torch.Tensor], model_outputs: Dict[str, torch.Tensor], tokenizer: PreTrainedTokenizerBase, answer_key: str = "answer", # 从batch中取真实答案的key debug_mode: bool = False, reward_scale: float = 1.0, wrong_penalty: float = -2.0 ) -> torch.Tensor: """ 数学题答案校验奖励函数 - 提取模型输出末尾的数字（支持整数/小数/分数格式） - 与batch中提供的标准答案比对 - 正确给+1.0，错误给-2.0，未提取到给0.0 """ device = model_outputs['response_ids'].device batch_size = len(batch[answer_key]) rewards = torch.zeros(batch_size, dtype=torch.float32, device=device) # 逐条处理 for i in range(batch_size): try: # 1. 解码模型输出 response_ids = model_outputs['response_ids'][i] # 过滤掉padding和special tokens mask = response_ids != tokenizer.pad_token_id clean_ids = response_ids[mask] response_text = tokenizer.decode(clean_ids, skip_special_tokens=True).strip() # 2. 提取标准答案（来自batch，通常是数字字符串） true_answer_str = str(batch[answer_key][i]).strip() # 3. 从response_text中提取最后一个数字（支持多种格式） # 匹配：整数(123)、小数(3.14)、分数(1/2)、带单位(5个) number_pattern = r'[-+]?\d*\.?\d+(?:/\d+)?(?=\D*$|\s*$)' matches = re.findall(number_pattern, response_text) pred_answer_str = matches[-1] if matches else "" # 4. 数值标准化与比对（处理分数、小数等） def normalize_number(s: str) -> Optional[float]: s = s.strip() if not s: return None # 处理分数：'3/4' -> 0.75 if '/' in s and s.count('/') == 1: try: a, b = s.split('/') return float(a.strip()) / float(b.strip()) except: return None # 处理普通数字 try: return float(s) except: return None true_num = normalize_number(true_answer_str) pred_num = normalize_number(pred_answer_str) # 5. 计算reward if true_num is not None and pred_num is not None: # 允许小浮点误差（如3.0 vs 3） if abs(true_num - pred_num) < 1e-6: rewards[i] = 1.0 * reward_scale else: rewards[i] = wrong_penalty else: # 无法解析答案，给中性分 rewards[i] = 0.0 except Exception as e: if debug_mode: print(f"[Reward Debug] Error on sample {i}: {e}") rewards[i] = 0.0 return rewards

2.2 关键设计点解析

• 为什么用正则而不是LLM解析？
  简单任务（如数学答案）用规则更稳定、更快、零成本。LLM解析可能把“答案是3”误判为“3.0”，而正则明确锚定结尾数字。

• 为什么只取最后一个数字？
  GSM8K标准格式要求模型在结尾写So the answer is 3.，我们信任这个约定，避免过度解读中间步骤。

• 为什么用normalize_number而不直接float()？
  真实数据含分数（1/2）、科学计数（1e2）、负数（-5），统一转换防报错。

• reward_scale和wrong_penalty为何要参数化？
  后期可调优：若模型总不敢答，调高reward_scale；若胡编答案多，调低wrong_penalty（如-0.5）降低惩罚力度。

2.3 如何集成到verl训练脚本？

只需在PPO配置中替换reward函数：

# config/ppo_gsm8k.yaml algorithm: name: "ppo" reward_fn: "math_answer_reward_fn" # 指向你定义的函数 reward_kwargs: answer_key: "answer" # batch中真实答案的key名 reward_scale: 2.0 wrong_penalty: -1.5 # 在训练启动脚本中注册函数 # train.py from my_reward_module import math_answer_reward_fn # verl会自动通过字符串名找到该函数

验证方式：在训练日志中搜索reward_mean，初期应在[-1.5, 2.0]间波动；几轮后若稳定在1.2+，说明模型开始学会输出正确答案。

3. 进阶技巧：让奖励更鲁棒、更智能

纯规则奖励虽快，但面对复杂任务（如代码生成、多跳推理）易失效。下面三个技巧，帮你平滑过渡到更强大的reward设计。

3.1 技巧一：加入“置信度加权”，避免过拟合表面模式

问题：规则奖励可能让模型学会“凑数字”。例如题干问“多少人”，它不管逻辑，硬在结尾写100。解决方案：用Actor模型自身的logprobs评估“答案位置”的置信度。

def confidence_weighted_reward_fn(...): # ... 前面提取答案逻辑不变 ... # 新增：计算答案token的平均logprob if 'logits' in model_outputs: logits = model_outputs['logits'][i] # shape: (seq_len, vocab_size) response_ids = model_outputs['response_ids'][i] # 找到答案token在response中的起始位置（需结合tokenizer） answer_tokens = tokenizer.encode(pred_answer_str, add_special_tokens=False) if len(answer_tokens) > 0: # 简单起见，取答案第一个token的logprob first_ans_token = answer_tokens[0] pos = (response_ids == first_ans_token).nonzero() if len(pos) > 0: logprob = torch.log_softmax(logits[pos[0][0]], dim=-1)[first_ans_token] # 将reward乘以exp(logprob)作为置信权重 rewards[i] *= torch.exp(logprob).item()

效果：模型不仅答对，还要“自信地答对”，抑制胡编。

3.2 技巧二：组合多个reward信号，用加权和

单一reward易片面。verl支持CompositeRewardFn，轻松组合：

from verl.utils.reward import CompositeRewardFn # 组合：答案正确性（主）+ 推理长度合理性（辅）+ 无有害内容（兜底） composite_reward = CompositeRewardFn( reward_fns=[ ("answer_correct", math_answer_reward_fn, {"weight": 0.6}), ("length_bonus", length_reward_fn, {"weight": 0.2, "target_len": 120}), ("safety_penalty", safety_reward_fn, {"weight": -0.2}), ] )

注意：权重和必须为1，且负权重用于惩罚项。verl会自动归一化。

3.3 技巧三：引入“可验证逻辑”，超越字符串匹配

对需要多步推理的任务（如“如果A>B且B>C，则A>C吗？”），纯答案比对不够。我们可以嵌入轻量逻辑检查器：

def logic_verification_reward_fn(...): # 1. 提取模型的推理步骤（用句号分割） steps = [s.strip() for s in response_text.split('。') if s.strip()] # 2. 构建符号逻辑图（简化版） facts = set() for step in steps[:3]: # 只看前3步，防过长 if "因为" in step or "由于" in step: # 提取"因为X，所以Y" -> 添加事实X pass # 3. 检查结论是否被前提逻辑蕴含（调用z3或简易规则引擎） if is_logically_entailed(facts, conclusion): return 1.0 else: return -0.5 # 比乱答惩罚轻，鼓励尝试

这种reward已接近“过程监督”，是迈向高质量RLHF的关键一步。

4. 避坑指南：新手常犯的5个致命错误

写reward函数不是写Python脚本，它运行在分布式、混合精度、动态batch的verl引擎中。这些错误不会立刻报错，但会让训练悄无声息地失败。

4.1 错误1：在reward函数里调用.cpu()或.item()

# ❌ 危险！强制同步，拖慢整个pipeline reward_val = rewards[i].item() # 同步等待GPU完成 # 正确：全程保持Tensor在GPU上 rewards[i] = torch.tensor(1.0, device=device, dtype=torch.float32)

4.2 错误2：忽略batch内padding导致的长度不一致

# ❌ 错误：假设所有response等长 response_ids = model_outputs['response_ids'] # shape: (B, L) last_token = response_ids[:, -1] # 取最后一列 —— 但padding位置可能是pad_id！ # 正确：用attention_mask定位真实结尾 attention_mask = model_outputs.get('attention_mask', torch.ones_like(response_ids, dtype=torch.bool)) seq_len = attention_mask.sum(dim=1) # 每条的真实长度 last_token = torch.gather(response_ids, 1, (seq_len-1).unsqueeze(1))

4.3 错误3：reward值域过大，导致Critic爆炸

# ❌ 危险：reward范围[-100, +100]，Critic梯度爆炸 rewards[i] = float(score * 100) # 来自外部API的原始分 # 正确：归一化到[-2, 2]或使用tanh压缩 rewards[i] = torch.tanh(torch.tensor(score * 0.1)).item()

4.4 错误4：修改了batch或model_outputs字典

# ❌ 危险：破坏verl内部状态 batch['prompt'] = batch['prompt'] + " [REWARD]" # verl后续会出错 # 正确：只读，不写 prompt_text = tokenizer.decode(batch['prompt'][i], skip_special_tokens=True)

4.5 错误5：没设device，导致CPU/GPU混用

# ❌ 错误：在GPU模型上返回CPU tensor return torch.tensor([1.0, 0.5]) # 默认在CPU # 正确：显式指定device device = model_outputs['response_ids'].device return torch.tensor([1.0, 0.5], device=device, dtype=torch.float32)

5. 总结：构建你自己的奖励工程方法论

写好reward函数，不是一次编码任务，而是一套持续迭代的工程实践。根据verl在真实项目中的经验，我们提炼出四步法：

5.1 第一步：从“可验证规则”起步（1天）

• 目标：快速获得baseline reward，验证训练流程
• 做法：用正则、关键词、数值比对等确定性逻辑
• 指标：reward_mean是否收敛？reward_std是否下降？

5.2 第二步：加入“轻量模型信号”（3天）

• 目标：提升reward的语义理解能力
• 做法：接入tiny-BERT、distilRoBERTa等<100M参数模型，做二分类（好/坏）
• 关键：用verl的HFRewardModel封装，避免重写IO

5.3 第三步：设计“可解释过程reward”（1周）

• 目标：让模型学会正确推理路径，不止答案对
• 做法：拆解任务为子步骤（如“提取实体→判断关系→得出结论”），每步给分
• 工具：用spaCy/NLTK做实体识别，NetworkX建模逻辑图

5.4 第四步：对齐“人工偏好分布”（持续）

• 目标：让reward函数逼近人类评委打分
• 做法：收集人工标注的pairwise preference（A>B），训练reward model
• verl支持：verl.data.preference_dataset可直接加载偏好数据

记住：最好的reward函数，是那个让你敢把模型交给客户的函数。它不一定最复杂，但一定最可靠、最透明、最容易调试。

获取更多AI镜像

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