强化学习开发指南

from dashscope.finetune.reinforcement import (
    AbstractRewardProcessor, RewardInput, RewardOutput, Reward, TaskStatus,
    reward_func, sub_reward_func, aggregate_func
)

@reward_func("SafetyProcessor")
class SafetyProcessor(AbstractRewardProcessor):
    """多维度安全性评分"""
    BANNED = {"hack", "exploit", "attack"}

    @sub_reward_func("toxicity", sub_weight=0.7)
    def toxicity(self, input: RewardInput) -> RewardOutput:
        """毒性检测子维度（权重 0.7，同步）"""
        messages = input.agent_output.messages
        content = messages[0].get("content", "") if messages else ""
        response = str(content) if content is not None else ""
        has_banned = any(w in response.lower() for w in self.BANNED)
        return RewardOutput(
            reward=Reward(
                reward_score=0.0 if has_banned else 1.0,
                reward_metrics={"score-1": 0.1},
            ),
            status=TaskStatus.SUCCESS,
            error=None,
        )

    @sub_reward_func("refusal", sub_weight=0.3)
    async def refusal(self, input: RewardInput) -> RewardOutput:
        """拒绝检测子维度（权重 0.3，异步）"""
        messages = input.agent_output.messages
        content = messages[0].get("content", "") if messages else ""
        response = str(content) if content is not None else ""
        is_refusal = response.strip().lower().startswith("i cannot")
        return RewardOutput(
            reward=Reward(
                reward_score=0.5 if is_refusal else 1.0,
                reward_metrics={"score-1": 0.2},
            ),
            status=TaskStatus.SUCCESS,
            error=None,
        )

    @aggregate_func
    async def aggregate(self, sub_rewards: dict[str, RewardOutput]) -> RewardOutput:
        """聚合所有子维度（异步）"""
        weights = self.get_weights()
        scores = self.get_scores(sub_rewards)
        if len(scores) != len(weights):
            raise ValueError("scores and weights must have the same length")

        # 0.5 * 0.3 + 1.0 * 0.7 = 0.85
        total = sum(scores[k] * weights[k] for k in scores)

        reward_metrics = self.get_reward_metrics(sub_rewards)
        user_reward_metrics = {"aggregate-1": 0.4}
        user_reward_metrics.update(reward_metrics)
        return RewardOutput(
            reward=Reward(
                reward_score=total,
                reward_metrics=user_reward_metrics,
            ),
            status=TaskStatus.SUCCESS,
            error=None,
        )

from dashscope.finetune.reinforcement import (
    AbstractGroupRewardProcessor, GroupRewardInput, GroupRewardOutput,
    GroupReward, Reward, TaskStatus
)

class DemoGroupRewardProcessor(AbstractGroupRewardProcessor):
    def setup(self) -> None:
        pass

    def process(self, input: GroupRewardInput) -> GroupRewardOutput:
        rewards = []
        for agent_output in input.agent_outputs:
            score = 0.0
            # ground_truth 出现在 messages 内容里 → 1.0；messages 非空 → 0.5；否则 0.0
            if input.ground_truth is not None and agent_output.messages:
                gt_str = str(input.ground_truth)
                for msg in agent_output.messages:
                    if isinstance(msg.get("content"), str) and gt_str in msg["content"]:
                        score = 1.0
                        break
                if score == 0.0 and len(agent_output.messages) > 0:
                    score = 0.5
            rewards.append(Reward(reward_score=score))

        return GroupRewardOutput(
            reward=GroupReward(rewards=rewards),
            status=TaskStatus.SUCCESS,
            error=None,
        )

client = AgenticRL()
(rollout_entity_ids,
 reward_entity_ids,
 group_reward_entity_ids,
 rollout_instance_ids,
 reward_instance_ids,
 group_reward_instance_ids) = await client.register_functions(
    functions=[
        # rollout-only：演示版，覆盖 trace_client/trace_tool 全分支
        AgenticRLFunctionComponent(
            type=FunctionType.ROLLOUT,
            fcmodel=FunctionComponentModel(
                classpath="functions.rollout.rollout_only.DemoRolloutProcessor")),
        # reward
        AgenticRLFunctionComponent(
            type=FunctionType.REWARD,
            fcmodel=FunctionComponentModel(
                classpath="functions.reward.reward.DemoRewardProcessor")),
        # reward-decorator：文件路径:类名 形式
        AgenticRLFunctionComponent(
            type=FunctionType.REWARD,
            fcmodel=FunctionComponentModel(
                classpath="functions/reward/reward_decorator.py:SafetyProcessor")),
        # group-reward
        AgenticRLFunctionComponent(
            type=FunctionType.GROUP_REWARD,
            fcmodel=FunctionComponentModel(
                classpath="functions.reward.group_reward.DemoGroupRewardProcessor")),
    ],
    lazy_load=False  # False = 立即获取 instance_id，用于测试
)

## 测试 Rollout
result = await AgenticRL.test_functions(
    instance_id=rollout_instance_ids[0],
    type=FunctionType.ROLLOUT,
    input_data=rollout_input
)

## 测试 Reward
result = await AgenticRL.test_functions(
    instance_id=reward_instance_ids[0],
    type=FunctionType.REWARD,
    input_data=reward_input
)

## 测试 Reward 装饰器版（第二个 reward 实例，对应 refusal 分支 reward_score=0.85）
result = await AgenticRL.test_functions(
    instance_id=reward_instance_ids[1],
    type=FunctionType.REWARD,
    input_data=reward_decorator_input
)

## 测试 Group Reward
result = await AgenticRL.test_functions(
    instance_id=group_reward_instance_ids[0],
    type=FunctionType.GROUP_REWARD,
    input_data=group_reward_input
)

{
  "func_type": "rollout",
  "messages": [
    {"role": "user", "content": "Hello, please introduce yourself"}
  ],
  "tools": null,
  "ground_truth": "",
  "rollout_extra": {
    "rollout_id": "ro-ins-e8e92e94-b5c3-4546-b7b1-9832b8b948f2",
    "training_state": {"task_id": "task-001"}
  },
  "sampling_params": {
    "temperature": 0.7,
    "max_tokens": 1024,
    "max_turns": 25,
    "timeout": 60.0
  },
  "model_resource": {
    "model_name": "qwen3.5-9b",
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    "api_key": "your-api-key-here"
  },
  "request_metadata": {
    "protocol": "openai",
    "system_prompt": "",
    "job_id": "dummy-job-id",
    "sample_id": "dummy-sample-id",
    "rollout_id": "dummy-rollout-id",
    "attempt_id": "dummy-attempt-id"
  },
  "additional_field": "any extra data you want"
}