Agentic-RL 强化学习智能体训练

原文：Hello-Agents · 第十一章 Agentic-RL · 在线阅读

一句话

Agentic-RL 是将 LLM 作为可学习策略，嵌入智能体的感知-决策-执行循环，通过强化学习优化多步任务表现。

什么时候翻这页

• 需要训练具备推理、工具使用等高级能力的智能体
• 想让智能体学会自我改进和长期规划
• 需要优化多步任务的完成质量而非单步回答质量
• 希望从"对话助手"进化为"自主智能体"

核心概念

Agentic RL vs PBRFT

LLM 训练流程

1. 预训练(Pretraining): 学习语言基本规律和世界知识

   • 使用海量文本数据
   • 自监督学习: 根据上文预测下一个词
   • 目标函数: $\mathcal{L}{\text{pretrain}} = -\sum{t=1}^{T} \log P(x_t | x_1, x_2, ..., x_{t-1}; \theta)$

2. 后训练(Post-training): 对齐人类偏好和价值观

   • 监督微调(SFT): 学习遵循指令和对话格式
   • 奖励建模(RM): 学习人类偏好
   • 强化学习微调: 优化语言模型生成质量

Agentic RL 六大核心能力

1. 推理(Reasoning): 多步逻辑推导，学习推理策略
2. 工具使用(Tool Use): API/工具调用，学会何时用、如何用
3. 记忆(Memory): 长期信息保持，学习记忆管理
4. 规划(Planning): 行动序列规划，学会动态规划
5. 自我改进(Self-Improvement): 自我反思优化，从错误中学习
6. 感知(Perception): 多模态理解，视觉推理和工具使用

奖励函数类型

1. 准确率奖励(AccuracyReward):

   • $r_{\text{acc}}(a, a^) = \begin{cases} 1 & \text{if } a = a^ \ 0 & \text{otherwise} \end{cases}$
   • 优点: 简单直接；缺点: 奖励稀疏

2. 长度惩罚(LengthPenaltyReward):

   • $r_{\text{length}}(a, a^, l) = r_{\text{acc}}(a, a^) - \alpha \cdot \max(0, l - l_{\text{target}})$
   • 优点: 鼓励简洁表达；缺点: 可能抑制详细推理

3. 步骤奖励(StepReward):

   • $r_{\text{step}}(a, a^, s) = r_{\text{acc}}(a, a^) + \beta \cdot s$
   • 优点: 鼓励可解释推理；缺点: 可能导致冗余步骤

怎么做

快速上手示例

import sys
import json

from hello_agents.tools import RLTrainingTool

# 创建RL训练工具
rl_tool = RLTrainingTool()

# 1. SFT训练
sft_result_str = rl_tool.run({
    "action": "train",
    "algorithm": "sft",
    "model_name": "Qwen/Qwen3-0.6B",
    "output_dir": "./models/quick_test_sft",
    "max_samples": 10,
    "num_epochs": 1,
    "batch_size": 2,
    "use_lora": True
})

# 2. GRPO训练
grpo_result_str = rl_tool.run({
    "action": "train",
    "algorithm": "grpo",
    "model_name": "Qwen/Qwen3-0.6B",
    "output_dir": "./models/quick_test_grpo",
    "max_samples": 5,
    "num_epochs": 1,
    "batch_size": 2,
    "use_lora": True
})

# 3. 评估模型
eval_result_str = rl_tool.run({
    "action": "evaluate",
    "model_path": "./models/quick_test_grpo",
    "max_samples": 10,
    "use_lora": True
})

完整训练流程

class AgenticRLPipeline:
    def __init__(self, config_path):
        self.config = json.load(open(config_path))
        self.rl_tool = RLTrainingTool()
        self.results = {}
    
    def stage1_prepare_data(self):
        """阶段1: 数据准备"""
        # 加载并转换数据集
        pass
    
    def stage2_sft_training(self):
        """阶段2: SFT训练"""
        result = self.rl_tool.run({
            "action": "train",
            "algorithm": "sft",
            "model_name": self.config["model"]["base_model"],
            "output_dir": self.config["sft"]["output_dir"],
            "num_epochs": self.config["sft"]["num_epochs"],
            "batch_size": self.config["sft"]["batch_size"],
            "use_lora": True,
        })
        return json.loads(result)["output_dir"]
    
    def stage3_sft_evaluation(self, model_path):
        """阶段3: SFT评估"""
        result = self.rl_tool.run({
            "action": "evaluate",
            "model_path": model_path,
            "max_samples": self.config["eval"]["max_samples"],
            "use_lora": True,
        })
        return json.loads(result)
    
    def stage4_grpo_training(self, sft_model_path):
        """阶段4: GRPO训练"""
        result = self.rl_tool.run({
            "action": "train",
            "algorithm": "grpo",
            "model_name": sft_model_path,
            "output_dir": self.config["grpo"]["output_dir"],
            "num_epochs": self.config["grpo"]["num_epochs"],
            "batch_size": self.config["grpo"]["batch_size"],
            "use_lora": True,
        })
        return json.loads(result)["output_dir"]
    
    def stage5_grpo_evaluation(self, model_path):
        """阶段5: GRPO评估"""
        result = self.rl_tool.run({
            "action": "evaluate",
            "model_path": model_path,
            "max_samples": self.config["eval"]["max_samples"],
            "use_lora": True,
        })
        return json.loads(result)
    
    def run(self):
        """运行完整流程"""
        # 按顺序执行各阶段
        pass

自定义奖励函数

from typing import List
import re

def custom_reward_function(
    completions: List[str],
    **kwargs
) -> List[float]:
    """
    自定义奖励函数
    
    Args:
        completions: 模型生成的完成文本列表
        **kwargs: 其他参数,通常包含ground_truth等
        
    Returns:
        奖励值列表(每个值在0.0-1.0之间)
    """
    ground_truths = kwargs.get("ground_truth", [])
    rewards = []
    
    for completion, truth in zip(completions, ground_truths):
        reward = 0.0
        
        # 提取答案并计算奖励
        numbers = re.findall(r'-?\d+\.?\d*', completion)
        if numbers:
            try:
                pred = float(numbers[-1])
                truth_num = float(truth)
                error = abs(pred - truth_num)
                
                # 根据误差给予不同奖励
                if error < 0.01:
                    reward = 1.0
                elif error < 1.0:
                    reward = 0.8
                elif error < 5.0:
                    reward = 0.5
                
                # 额外奖励:鼓励展示推理步骤
                if "step" in completion.lower() or "=" in completion:
                    reward += 0.1
                    
            except ValueError:
                reward = 0.0
        
        rewards.append(min(reward, 1.0))
    
    return rewards

分布式训练配置

DDP 配置 (multi_gpu_ddp.yaml):

compute_environment: LOCAL_MACHINE
distributed_type: MULTI_GPU
num_processes: 4
machine_rank: 0
num_machines: 1
gpu_ids: all
mixed_precision: fp16

ZeRO-2 配置 (deepspeed_zero2.yaml):

compute_environment: LOCAL_MACHINE
distributed_type: DEEPSPEED
num_processes: 4
machine_rank: 0
num_machines: 1
gpu_ids: all
mixed_precision: fp16
deepspeed_config:
  gradient_accumulation_steps: 4
  gradient_clipping: 1.0
  offload_optimizer_device: none
  offload_param_device: none
  zero3_init_flag: false
  zero_stage: 2

启动命令:

# DDP训练
accelerate launch --config_file multi_gpu_ddp.yaml train_script.py

# ZeRO-2训练
accelerate launch --config_file deepspeed_zero2.yaml train_script.py

命令 / 代码速查

安装 HelloAgents 框架

# 安装HelloAgents框架(第11章版本)
pip install "hello-agents[rl]==0.2.5"

# 或者从源码安装
cd HelloAgents
pip install -e ".[rl]"

数据集加载

from hello_agents.tools import RLTrainingTool
import json

rl_tool = RLTrainingTool()

# 加载SFT格式数据集
sft_result = rl_tool.run({
    "action": "load_dataset",
    "format": "sft",
    "max_samples": 5
})
sft_data = json.loads(sft_result)

# 加载RL格式数据集
rl_result = rl_tool.run({
    "action": "load_dataset",
    "format": "rl",
    "max_samples": 5
})
rl_data = json.loads(rl_result)

创建奖励函数

# 准确率奖励
accuracy_result = rl_tool.run({
    "action": "create_reward",
    "reward_type": "accuracy"
})

# 长度惩罚奖励
length_result = rl_tool.run({
    "action": "create_reward",
    "reward_type": "length_penalty",
    "max_length": 1024,
    "penalty_weight": 0.001
})

# 步骤奖励
step_result = rl_tool.run({
    "action": "create_reward",
    "reward_type": "step",
    "step_bonus": 0.1
})

模型导出

from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel

# 加载基础模型
base_model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")

# 加载LoRA权重
model = PeftModel.from_pretrained(base_model, "./models/grpo_model")

# 合并权重
merged_model = model.merge_and_unload()

# 保存合并后的模型
merged_model.save_pretrained("./models/merged_model")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer.save_pretrained("./models/merged_model")

与 Python / Claude Code 手册的联系

• Python 手册: 使用 Python 进行模型训练和推理，涉及数据处理、模型加载、训练循环等
• Claude Code 手册: Claude Code 可用于辅助编写训练脚本、调试模型、分析训练结果

初学者易错点

1. 混淆 PBRFT 和 Agentic RL: PBRFT 优化单轮对话质量，Agentic RL 优化多步任务完成度
2. 奖励函数设计不当: 只使用准确率奖励可能导致训练初期缺乏有效反馈
3. 忽略数据质量: 训练前应检查数据格式、空值、重复样本等问题
4. 超参数设置不合理: 学习率、batch size 等参数需要根据任务特点调整
5. 分布式训练配置错误: 确保配置文件与实际硬件环境匹配

相关词条

• 强化学习 (Reinforcement Learning)
• 监督微调 (Supervised Fine-Tuning, SFT)
• 群组相对策略优化 (Group Relative Policy Optimization, GRPO)
• 低秩适配 (LoRA)
• GSM8K 数据集