用verl做实验：奖励函数自定义全过程

用verl做实验：奖励函数自定义全过程

强化学习在大语言模型后训练中正变得越来越关键——但真正让RL落地的，从来不是算法本身，而是你能否快速、可靠、可复现地定义和验证自己的奖励逻辑。很多团队卡在第一步：想试试“更贴合业务目标”的奖励，却困在框架黑盒里，改不动、调不稳、看不到中间结果。

verl 不是另一个需要从头造轮子的 RL 框架。它把 LLM 后训练中最耗时的工程部分做了深度解耦：Actor、Critic、Reward Model、Rollout、Ref Model 各自独立，又通过清晰的数据流连接。而其中最常被定制、也最影响最终效果的模块，就是奖励函数（reward function）。

本文不讲理论推导，不堆公式，不跑通默认示例就收工。我们聚焦一个真实场景：如何从零开始，在 verl 中完整实现一个自定义奖励函数，并验证它是否按预期工作。你会看到：

• 奖励函数在 verl 架构中到底“长什么样”、挂在哪里、谁调用它
• 如何绕过预置 reward model，直接写 Python 函数做规则/启发式打分
• 怎样注入外部服务（如轻量级分类模型）作为动态奖励信号
• 如何在训练过程中实时打印、记录、对比不同样本的奖励值
• 一个可立即运行的最小闭环示例（含完整代码片段）

全程基于 verl 官方镜像，无需编译、不改源码、不碰 Ray 底层通信——所有操作都在配置和脚本层完成。

1. 奖励函数在 verl 中的位置与职责

在 verl 的 HybridFlow 架构中，奖励函数不是隐藏在某个 trainer 类里的私有方法，而是一个显式声明、独立部署、可热替换的计算单元。它的核心职责非常明确：

对 Actor 生成的每个 response，结合原始 prompt 和其他上下文，输出一个标量 reward 值（float）

这个值将直接影响 PPO 或 GRPO 的策略梯度更新方向。它不参与反向传播（除非你用可微 reward model），但它决定了“什么行为该被鼓励”。

1.1 两种奖励接入方式：Model-based vs Function-based

verl 支持两类奖励来源，它们在配置和使用上截然不同：

本文聚焦后者——Custom Reward Function。它更适合快速实验、A/B 测试、冷启动阶段探索 reward shape，也是理解 reward 如何影响 policy 的最佳入口。

1.2 它在数据流中具体在哪一环？

看一张简化的 verl 训练数据流图（文字描述）：

[Dataset] ↓（prompt + chosen/rejected） [Actor Model] → 生成 response A / B ↓（response A, prompt） [**Reward Function**] → 计算 reward_A ↓（response B, prompt） [**Reward Function**] → 计算 reward_B ↓（reward_A, reward_B） [PPO Loss] → 构建 KL 散度 + reward 差分项 → 更新 Actor

关键点：

• Reward Function 是同步、逐样本调用的，不涉及 batch 维度（除非你主动 vectorize）
• 它接收的是(prompt: str, response: str, **kwargs)这样的干净输入，不暴露模型参数、梯度、device 信息——你只管打分
• 输出必须是单个float，verl 会自动处理 NaN、inf 等异常值（转为 0）

这意味着：你可以用正则表达式检查 response 是否包含禁用词；可以用 jieba 统计中文关键词密度；可以调用 requests 请求本地 FastAPI 服务；甚至可以临时加载一个 ONNX 小模型做情感打分——只要它返回一个数字。

2. 从零编写一个可运行的自定义奖励函数

我们以一个典型需求为例：电商客服对话场景中，要求模型回复必须包含“已为您登记”或“已提交工单”等确定性动作短语，且不能出现“稍后回复”“正在处理”等模糊表述。满足则给 +1 分，否则给 -0.5 分。

这不是一个能靠 RM 学出来的模式，而是强业务规则。我们把它写成一个 verl 兼容的 reward function。

2.1 函数定义：简洁、无副作用、可测试

# rewards/simple_action_checker.py import re def check_action_clarity(prompt: str, response: str, **kwargs) -> float: """ 检查客服回复是否包含明确动作短语，且无模糊表述 返回：+1.0（明确动作）、-0.5（模糊表述）、0.0（其他） """ # 明确动作关键词（支持中英文） positive_keywords = [ r"已为您登记", r"已提交工单", r"已创建case", r"已记录问题", r"已转交技术团队" ] # 模糊表述黑名单 negative_keywords = [ r"稍后回复", r"正在处理", r"尽快给您答复", r"我们会跟进", r"稍等一下" ] # 检查正面关键词（至少匹配一个） has_positive = any(re.search(kw, response) for kw in positive_keywords) # 检查负面关键词（任意匹配即触发） has_negative = any(re.search(kw, response) for kw in negative_keywords) if has_positive and not has_negative: return 1.0 elif has_negative: return -0.5 else: return 0.0

这个函数完全符合 verl 要求：

• 输入签名严格为(prompt, response, **kwargs)
• 输出是float
• 无全局状态、无文件 IO、无网络请求（纯 CPU 计算）
• 可单独导入测试：

# test_reward.py from rewards.simple_action_checker import check_action_clarity print(check_action_clarity("", "已为您登记，预计2小时内处理")) # → 1.0 print(check_action_clarity("", "正在处理，请稍等")) # → -0.5 print(check_action_clarity("", "好的，谢谢")) # → 0.0

2.2 注册到 verl 配置：三行搞定

verl 使用 Hydra 配置系统。你只需在训练配置 YAML 文件中指定函数路径：

# conf/reward/simple_action.yaml reward_model: _target_: rewards.simple_action_checker.check_action_clarity # 可选：传入额外参数，例如阈值、权重等 # threshold: 0.8 # weight: 1.2

然后在主训练配置（如conf/train/ppo_qwen.yaml）中引用它：

defaults: - reward: simple_action # ← 就是这里！加载上面的配置 # 其他配置保持不变... trainer: name: "PPOTrainer" ...

注意：_target_必须是可 import 的完整路径（模块名 + 函数名），且该模块需在 Python path 中。推荐将 rewards 目录放在项目根目录下，与examples/同级。

2.3 验证函数是否被正确加载与调用

光写对不够，要确认 verl 真的在用它。最直接的方式：在函数内加日志。

# rewards/simple_action_checker.py import logging import re logger = logging.getLogger(__name__) def check_action_clarity(prompt: str, response: str, **kwargs) -> float: logger.info(f"[RewardFn] Prompt: {prompt[:30]}... | Response: {response[:50]}...") # ...其余逻辑不变 result = 1.0 if (has_positive and not has_negative) else (-0.5 if has_negative else 0.0) logger.info(f"[RewardFn] → Reward = {result}") return result

启动训练时，加上--log-level INFO，你将在日志中清晰看到每条样本的 prompt、response 和对应 reward，无需进 debugger，一眼定位 reward 行为是否符合预期。

3. 进阶实践：接入外部服务与轻量模型

规则函数适合冷启动，但真实 reward 往往需要语义理解能力。下面展示两个生产级常用模式，均无需修改 verl 核心代码。

3.1 调用本地 FastAPI 服务（HTTP 方式）

假设你已部署一个轻量情感分析服务（FastAPI + transformers pipeline），地址http://localhost:8000/score，接受 JSON{ "text": "..." }，返回{ "score": 0.92 }。

# rewards/api_sentiment.py import requests import time def api_sentiment_reward(prompt: str, response: str, timeout: int = 5, **kwargs) -> float: """ 调用本地情感分析 API，返回归一化后的 score（0~1 → -1~1） 失败时返回 0.0，避免中断训练 """ try: payload = {"text": response} resp = requests.post("http://localhost:8000/score", json=payload, timeout=timeout) resp.raise_for_status() score = resp.json().get("score", 0.0) # 映射到 [-1, 1] 区间，便于 PPO 梯度更新 return float(score) * 2 - 1 except Exception as e: logger.warning(f"[API Reward] Request failed: {e}") return 0.0

关键设计点：

• 加了timeout和try/except，防止 reward service 挂掉拖垮整个训练
• 失败返回0.0（中性），而非报错——verl 会跳过该样本的梯度更新，但不 crash
• 用logger.warning记录失败，方便事后排查

配置方式同上，只需改_target_。

3.2 加载 ONNX 小模型（零依赖、低延迟）

如果你追求极致性能，可将一个蒸馏后的情感分类模型导出为 ONNX，用 onnxruntime 推理：

# rewards/onnx_sentiment.py import numpy as np import onnxruntime as ort from transformers import AutoTokenizer # 全局加载一次，避免每次调用都初始化 tokenizer = AutoTokenizer.from_pretrained("uer/roberta-finetuned-jd-binary-chinese") session = ort.InferenceSession("models/sentiment.onnx") def onnx_sentiment_reward(prompt: str, response: str, **kwargs) -> float: inputs = tokenizer( response, truncation=True, max_length=128, return_tensors="np" ) outputs = session.run(None, { "input_ids": inputs["input_ids"], "attention_mask": inputs["attention_mask"] }) # 假设输出是 [batch, 2] logits，取 positive class (index 1) prob = float(softmax(outputs[0])[0, 1]) return prob * 2 - 1 # same mapping def softmax(x): e_x = np.exp(x - np.max(x)) return e_x / e_x.sum()

优势：

• 无网络依赖，毫秒级响应
• 模型体积小（<50MB），可随镜像打包分发
• 完全离线，符合金融、政务等合规场景

4. 调试与可观测性：让 reward 不再是黑盒

自定义 reward 最怕“感觉不对但找不到原因”。verl 提供了天然的可观测性入口。

4.1 在训练循环中记录 reward 分布

修改你的 reward 函数，加入统计逻辑（推荐用wandb或内置logging）：

# rewards/trackable_action.py import logging from collections import defaultdict logger = logging.getLogger(__name__) stats = defaultdict(int) # 全局统计（注意：多进程下需用 ray.put） def trackable_action_reward(prompt: str, response: str, **kwargs) -> float: global stats # 分类统计 if "已为您登记" in response: stats["positive_action"] += 1 elif "正在处理" in response: stats["vague_response"] += 1 else: stats["neutral"] += 1 # 打印每 100 条的统计（避免刷屏） total = sum(stats.values()) if total % 100 == 0: logger.info(f"[RewardStats] Total: {total}, Positive: {stats['positive_action']}, " f"Vague: {stats['vague_response']}, Neutral: {stats['neutral']}") return check_action_clarity(prompt, response)

启动训练时，你会看到类似日志：

INFO:__main__:[RewardStats] Total: 100, Positive: 42, Vague: 18, Neutral: 40 INFO:__main__:[RewardStats] Total: 200, Positive: 87, Vague: 32, Neutral: 81

这比看 loss 曲线更能说明 reward 是否在引导模型走向你想要的行为。

4.2 可视化 reward 与 response 质量的关系

在examples/grpo_trainer/下新建一个analyze_rewards.py脚本：

# analyze_rewards.py import json from verl.data.sft_dataset import SFTDataset # 加载一批原始数据（非 tokenized） dataset = SFTDataset( data_path="data/my_customer_data.jsonl", tokenizer=None, # 不 tokenize，保留原始字符串 max_seq_len=2048 ) # 对前 50 条 sample 手动计算 reward rewards = [] for i in range(50): item = dataset[i] prompt = item["prompt"] response = item["response"] # 假设你的数据格式含此字段 r = check_action_clarity(prompt, response) rewards.append({ "prompt": prompt[:100] + "...", "response": response[:100] + "...", "reward": r }) # 保存为 JSON，供后续分析或前端展示 with open("reward_analysis.json", "w", encoding="utf-8") as f: json.dump(rewards, f, ensure_ascii=False, indent=2)

运行它，你将得到一份带 reward 标签的样本集，可直接导入 Excel 做交叉分析（例如：reward 为负的 response，有多少比例包含“无法解决”？），这是优化 reward 规则的黄金依据。

5. 常见陷阱与避坑指南

即使是最简单的 reward function，也容易踩坑。以下是 verl 用户高频反馈问题：

5.1 “Reward 值全是 0，模型不更新”

原因：reward function 返回了None、np.nan、或未 catch 的异常（如正则编译失败）。
检查：在函数末尾加assert isinstance(result, (int, float)), f"Reward must be number, got {type(result)}"
修复：确保所有分支都有return，且类型为float。

5.2 “Reward 值波动极大，训练不稳定”

原因：reward 未归一化（如直接返回 0~100 的 raw score），导致 PPO 的 KL penalty 失效。
建议：将 reward 缩放到[-1, 1]或[-2, 2]区间。可用 min-max scaling 或 z-score（需先采样统计）。

5.3 “修改 reward 函数后，训练没变化”

原因：Hydra 配置未生效（常见于defaults路径写错、YAML 缩进错误、或缓存未清）。
验证：在main_ppo.py的trainer初始化前，打印cfg.reward_model._target_，确认是否为你新写的路径。

5.4 “多 GPU 下 reward 计算变慢”

原因：reward function 内部做了 heavy IO（如每次读文件）或未共享的模型加载。
方案：

• 将模型/资源加载提到函数外（global scope 或__init__）
• 使用ray.put()共享大对象（适用于 multi-controller 场景）
• 优先选择无状态、纯函数式实现

6. 总结：奖励函数自定义的本质是“控制权回归”

用 verl 做 reward 自定义，真正的价值不在于技术多炫酷，而在于它把最关键的决策权——“什么行为值得鼓励”——交还给了你。

你不再需要：

• 等待 RM 微调收敛数天
• 在 reward model 的 logits 层反复调试 temperature
• 把业务规则硬编码进模型权重

你只需要：

• 写一个 Python 函数，定义你的规则或调用你的服务
• 在 YAML 里配一行_target_
• 启动训练，看日志，调逻辑，再迭代

这个过程快、透明、可控。它让你能在一个下午内，完成从“灵光一现的 reward idea”到“验证它是否真能提升线上指标”的闭环。

下一步，你可以：

• 将多个 reward 函数组合（加权平均、条件路由）
• 用 offline evaluation 脚本批量测试 reward 在 holdout 数据上的相关性
• 把 reward 分布监控接入 Prometheus + Grafana，实现生产环境实时告警

强化学习的终点不是算法本身，而是你对智能体行为的精准塑造力。而 verl，正是那把趁手的刻刀。

获取更多AI镜像

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