百炼强化学习训练使用手册

5.2 自定义奖励函数基类和示例

"""
奖励计算基类

所有自定义奖励计算类都应该继承此基类
"""
from abc import ABC, abstractmethod

from core.models import RewardRequest, RewardResponse


class BaseRewardComputer(ABC):
    """
    奖励计算基类
    
    所有自定义奖励计算类必须继承此类并实现 compute_reward 方法
    """
    
    @abstractmethod
    def compute_reward(self, request: RewardRequest) -> RewardResponse:
        """
        计算奖励分数（必须实现）
        
        Args:
            request: 奖励计算请求
        
        Returns:
            奖励计算响应
        """
        pass
    
    def health_check(self) -> dict:
        """
        健康检查（可选实现）
        
        如果子类需要自定义健康检查逻辑，可以重写此方法。
        默认返回健康状态。
        
        Returns:
            包含 status 和可选 message 的字典
        """
        return {'status': 'healthy'}

实现示例：以下示例展示如何实现一个基于响应长度的奖励函数，使模型倾向于生成更长的响应内容。

"""
自定义奖励计算实现示例

展示如何创建自定义奖励计算类
"""
from core.base_reward import BaseRewardComputer
from core.models import RewardRequest, RewardResponse


class CustomRewardComputer(BaseRewardComputer):
    """
    自定义奖励计算类示例
    
    继承 BaseRewardComputer 并实现 compute_reward 方法
    """
    
    def compute_reward(self, request: RewardRequest) -> RewardResponse:
        """
        自定义奖励计算逻辑
        
        Args:
            request: 奖励计算请求
        
        Returns:
            奖励计算响应
        """
        try:
            # 示例：基于响应长度的简单奖励计算
            response_length = len(request.response)
            extra_params = request.extra_params
            
            # 你的自定义逻辑
            if response_length > 100:
                score = 0.9
            elif response_length > 50:
                score = 0.7
            else:
                score = 0.5
            
            # 可以访问 extra_params 来获取额外参数
            custom_param = extra_params.get('custom_param', 'default')
            
            return RewardResponse(
                score=score,
                status='success',
                extra_info={
                    'custom_logic': True,
                    'response_length': response_length,
                    'custom_param': custom_param
                }
            )
        except Exception as e:
            # 异常处理：返回错误响应
            return RewardResponse(
                score=0.0,
                status='fail',
                detail=f'Error in custom compute_reward: {str(e)}'
            )
    
    def health_check(self) -> dict:
        """
        自定义健康检查（可选）
        
        Returns:
            包含 status 和可选 message 的字典
        """
        return {
            'status': 'healthy',
            'message': 'Custom reward computer is healthy'
        }

5.3 系统及奖励函数配置

系统默认从 configs/config.yaml 文件中加载配置。

# 奖励计算服务配置文件

# 服务器配置
server:
  # 监听端口，默认 8000
  port: 8000

# 奖励计算配置
reward:
  # 自定义奖励计算类的完整路径
  # 格式: "module.path:ClassName"
  # 该类必须继承自 BaseRewardComputer
  # 如果不配置，则使用默认实现 (DefaultRewardComputer)
  class: "implementations.default_reward:DefaultRewardComputer"

# 健康检查配置（可选）
health_check:
  # 自定义健康检查函数的完整路径
  # 格式: "module.path:function_name"
  # 函数应该返回包含 'status' 键的字典
  # 如果不配置，则使用奖励计算类的 health_check 方法
  # function: "custom_module:custom_health_check"

# 日志配置
logging:
  # 是否启用日志输出，默认 true
  enable: true

# 线程池配置（可选）
# 用于 compute_reward 的并行计算
thread_pool:
  # 线程池最大线程数（>0 的整数）
  # 建议根据机器 CPU 核数和负载调整
  max_workers: 600
  # 允许排队等待的最大请求数（>0 的整数）
  # 达到该上限后新请求将直接返回 HTTP 失败响应
  queue_size: 600

6.1 训练集和测试集

训练集：rl-train.jsonl

测试集：rl-test.jsonl

7. 模型支持列表

说明：基于已训练的模型继续进行 GRPO、SFT、DPO 或 CPT 训练时，只需指定训练后的模型标识即可。