跳转至

第九章 Agentic RL(Agent强化学习)

⚠️ 时效性说明:本章涉及前沿模型/价格/榜单等信息,可能随版本快速变化;请以论文原文、官方发布页和 API 文档为准。

📌 定位说明:Agentic RL是2025年Agent领域的前沿方向——将强化学习(Reinforcement Learning)应用于LLM Agent的训练,使Agent通过"试错"自主学习如何更好地使用工具、规划任务和完成目标。这一章覆盖从SFT到GRPO等前沿训练方法,是我们教程超越datawhalechina/hello-agents第11章的关键内容。

📖 本章概览

主题 内容 预计学时
9.1 为什么Agent需要RL SFT的局限性与RL的优势 1小时
9.2 从SFT到RLHF到GRPO 训练范式的演进 2小时
9.3 奖励函数设计 如何为Agent行为定义奖励 2小时
9.4 GRPO训练实战 用GRPO训练一个工具调用Agent 3小时
9.5 Agent环境与Benchmark 训练和评估Agent的标准环境 1小时
9.6 前沿方向与展望 Agent RL的最新进展 1小时

9.1 为什么Agent需要强化学习?

9.1.1 SFT的局限性

当前大多数Agent都是通过提示工程(Prompt Engineering)和监督微调(SFT)来构建的。但这些方法有本质局限:

Text Only
SFT训练Agent的问题:

                      专家示范
   [状态] → [正确动作] → [结果]     ← SFT只学到了"模仿"

   但实际Agent运行时:
   [状态] → [动作A] → [失败] → 怎么办?SFT没教过如何从失败中恢复
   [状态] → [动作B] → [中间结果] → 下一步该怎么选?
   [状态] → [动作C] → [部分正确] → 如何优化?

SFT vs RL的核心区别

维度 SFT(监督微调) RL(强化学习)
训练信号 专家标注的"正确答案" 环境反馈的"奖励信号"
探索能力 只学已知的好策略 可以发现新的好策略
错误处理 不知道如何处理未见过的错误 从错误中学习恢复策略
长期规划 只优化单步正确性 优化整个轨迹的累积奖励
泛化能力 限于训练分布 更好地泛化到新场景
数据需求 需要大量高质量标注数据 可以通过与环境交互自动生成数据

9.1.2 Agent RL的核心思想

Python
# 概念性示意: Agent RL vs Agent SFT
from torch.nn.functional import cross_entropy

# SFT方式训练Agent(模仿学习)
def train_agent_sft(model, expert_demos):
    """
    给定专家的(状态, 动作)对,训练模型模仿。
    问题: 模型只学会了"在这个状态下,专家会做X",
    但不理解"为什么做X"以及"如果X失败了怎么办"。
    """
    for state, expert_action in expert_demos:
        loss = cross_entropy(model(state), expert_action)
        loss.backward()

# RL方式训练Agent(探索学习)
def train_agent_rl(model, environment):
    """
    让Agent在环境中自主探索,通过奖励信号学习最优策略。
    优点: 模型学会了"什么行为能带来好结果",包括错误恢复。
    """
    for episode in range(num_episodes):
        state = environment.reset()
        trajectory = []

        while not environment.done:
            # Agent自主选择动作(可能是好的,也可能是坏的)
            action = model.sample_action(state)

            # 环境返回奖励和新状态
            next_state, reward = environment.step(action)
            trajectory.append((state, action, reward))
            state = next_state

        # 根据整个轨迹的奖励更新模型
        update_policy(model, trajectory)

9.1.3 现实世界的例子

想象训练一个"代码调试Agent":

Text Only
SFT训练:
  输入: "这段代码有bug: ..."
  标签: "应该将第3行的 == 改为 ==="
  → 模型学会了常见bug的修复模式
  → 但遇到复杂bug时,它不知道如何一步步调试

RL训练:
  环境: 一个有bug的代码库 + 测试用例
  动作空间: [读文件, 搜索代码, 运行测试, 修改代码, ...]
  奖励:
    - 运行测试通过: +10
    - 正确定位bug文件: +3
    - 无效操作(读不存在的文件): -1
    - 超时: -5
  → 模型学会了: 先跑测试看哪个失败 → 读相关文件 → 定位问题 → 修复
  → 关键: 它还学会了"如果第一个修复不对,就回滚重试"

9.2 从SFT到RLHF到GRPO:训练范式演进

9.2.1 训练范式全景图

Text Only
2020-2022: SFT (Supervised Fine-Tuning)
  模型 ← 专家标注数据
  └→ InstructGPT 第一阶段

2022-2023: RLHF (RL from Human Feedback)
  模型 ← PPO算法 ← 奖励模型 ← 人类偏好数据
  └→ ChatGPT, Claude 1

2023-2024: DPO (Direct Preference Optimization)
  模型 ← 直接从偏好对学习(绕过奖励模型)
  └→ Llama 2, Mixtral

2024-2025: GRPO (Group Relative Policy Optimization)
  模型 ← 组内相对排名 ← 规则奖励 + 结果验证
  └→ DeepSeek-R1, Qwen-Agent

2025+: Agentic RL
  模型 ← 环境交互 ← 工具使用奖励 + 任务完成度
  └→ 训练Agent的专用RL方法

9.2.2 PPO vs DPO vs GRPO 对比

Python
# 三种训练方法的核心区别(概念性代码)
import torch

class PPOTrainer:
    """
    PPO (Proximal Policy Optimization)
    经典RLHF方法,需要独立的奖励模型。
    """
    def train_step(self, prompts, old_model, reward_model):
        # 1. 当前模型和旧模型分别生成回复
        new_responses = self.model.generate(prompts)
        old_responses = old_model.generate(prompts)

        # 2. 奖励模型打分
        rewards = reward_model.score(prompts, new_responses)

        # 3. 计算优势函数
        advantages = self.compute_advantages(rewards)

        # 4. PPO裁剪更新
        ratio = self.model.log_prob(new_responses) / old_model.log_prob(new_responses)
        clipped_ratio = torch.clamp(ratio, 1 - self.epsilon, 1 + self.epsilon)
        loss = -torch.min(ratio * advantages, clipped_ratio * advantages).mean()

        # 需要: 奖励模型 + 旧模型副本 → 显存需求大
        return loss

class DPOTrainer:
    """
    DPO (Direct Preference Optimization)
    直接从偏好对学习,不需要独立的奖励模型。
    """
    def train_step(self, prompts, chosen, rejected, ref_model):
        # 1. 计算当前模型和参考模型的log概率
        pi_chosen = self.model.log_prob(prompts, chosen)
        pi_rejected = self.model.log_prob(prompts, rejected)
        ref_chosen = ref_model.log_prob(prompts, chosen)
        ref_rejected = ref_model.log_prob(prompts, rejected)

        # 2. DPO损失(隐式奖励)
        log_ratio_chosen = pi_chosen - ref_chosen
        log_ratio_rejected = pi_rejected - ref_rejected

        loss = -torch.log(
            torch.sigmoid(self.beta * (log_ratio_chosen - log_ratio_rejected))
        ).mean()

        # 需要: 偏好对数据(chosen/rejected) + 参考模型
        # 不需要: 独立奖励模型
        return loss

class GRPOTrainer:
    """
    GRPO (Group Relative Policy Optimization)
    DeepSeek提出,用组内相对排名代替绝对奖励。
    特别适合Agent场景!
    """
    def train_step(self, prompts, reward_fn):
        """
        GRPO的关键创新:
        1. 对每个prompt生成一组(G个)回复
        2. 用规则奖励函数给每个回复打分
        3. 在组内计算相对优势(不需要奖励模型)
        """
        group_size = 8  # 每个prompt生成8个回复

        all_responses = []
        all_rewards = []

        for prompt in prompts:
            # 1. 采样一组回复
            responses = [self.model.generate(prompt) for _ in range(group_size)]

            # 2. 规则奖励函数打分(不需要训练奖励模型!)
            rewards = [reward_fn(prompt, resp) for resp in responses]

            all_responses.extend(responses)
            all_rewards.extend(rewards)

        # 3. 组内归一化计算优势
        rewards_tensor = torch.tensor(all_rewards)
        # 关键:在每组内做归一化
        for i in range(0, len(rewards_tensor), group_size):
            group = rewards_tensor[i:i+group_size]
            mean = group.mean()
            std = group.std() + 1e-8
            rewards_tensor[i:i+group_size] = (group - mean) / std

        # 4. 策略梯度更新
        loss = 0
        for resp, advantage in zip(all_responses, rewards_tensor):
            log_prob = self.model.log_prob(resp)
            loss -= log_prob * advantage

        return loss / len(all_responses)

9.2.3 为什么GRPO特别适合Agent?

Python
# 概念性伪代码:以下辅助函数(has_valid_tool_call_format, extract_tool_calls,
# task_completed_successfully, count_steps, contains_dangerous_action)
# 展示奖励函数的设计思路,实际实现需根据Agent的输出格式定制。

def agent_reward_function(prompt: str, agent_trajectory: str) -> float:
    """
    GRPO的一大优势:可以使用基于规则的奖励函数。
    对于Agent,我们可以精确定义什么是"好的行为"。
    """
    reward = 0.0

    # 1. 格式奖励:Agent的输出是否符合预期格式?
    if has_valid_tool_call_format(agent_trajectory):
        reward += 1.0
    else:
        reward -= 2.0  # 格式错误是严重问题

    # 2. 工具使用奖励:是否正确使用了工具?
    tool_calls = extract_tool_calls(agent_trajectory)
    for call in tool_calls:
        if call.is_valid:
            reward += 0.5
        if call.result_used_in_reasoning:
            reward += 1.0  # 使用了工具结果进行推理

    # 3. 任务完成奖励:最终是否达成目标?
    if task_completed_successfully(prompt, agent_trajectory):
        reward += 5.0

    # 4. 效率奖励:用更少的步骤完成加分
    num_steps = count_steps(agent_trajectory)
    if num_steps <= 3 and task_completed_successfully(prompt, agent_trajectory):
        reward += 2.0  # 高效完成
    elif num_steps > 10:
        reward -= 1.0  # 效率低

    # 5. 安全奖励:避免危险操作
    if contains_dangerous_action(agent_trajectory):
        reward -= 10.0

    return reward

9.3 奖励函数设计

9.3.1 Agent奖励的层次结构

Python
import re
import json
from dataclasses import dataclass
from typing import Callable

@dataclass
class RewardComponent:
    """奖励组件"""
    name: str
    weight: float
    compute: Callable  # (prompt, trajectory) -> float
    description: str

class AgentRewardSystem:
    """
    分层Agent奖励系统
    四个层次: 格式 → 工具使用 → 推理质量 → 任务完成
    """

    def __init__(self):
        self.components = [
            # Level 1: 格式正确性 (基础要求)
            RewardComponent(
                name="format_compliance",
                weight=1.0,
                compute=self._format_reward,
                description="Agent输出是否符合预期的JSON/XML格式"
            ),

            # Level 2: 工具使用质量
            RewardComponent(
                name="tool_usage",
                weight=2.0,
                compute=self._tool_usage_reward,
                description="工具调用是否正确、是否充分利用了工具结果"
            ),

            # Level 3: 推理质量
            RewardComponent(
                name="reasoning_quality",
                weight=1.5,
                compute=self._reasoning_reward,
                description="推理过程是否逻辑清晰、是否有效推进任务"
            ),

            # Level 4: 任务完成度
            RewardComponent(
                name="task_completion",
                weight=3.0,
                compute=self._task_completion_reward,
                description="是否最终完成了用户的任务"
            ),
        ]

    def compute_total_reward(self, prompt: str, trajectory: str) -> dict:
        """计算总奖励及各组件分数"""
        results = {}
        total = 0.0

        for component in self.components:
            score = component.compute(prompt, trajectory)
            weighted = score * component.weight
            results[component.name] = {
                "raw_score": score,
                "weight": component.weight,
                "weighted_score": weighted,
            }
            total += weighted

        results["total"] = total
        return results

    def _format_reward(self, prompt: str, trajectory: str) -> float:
        """
        Level 1: 格式奖励
        检查Agent输出是否符合结构化格式要求
        """
        score = 0.0

        # 检查思考/行动/观察标记
        if "<think>" in trajectory and "</think>" in trajectory:
            score += 0.3
        if "<action>" in trajectory and "</action>" in trajectory:
            score += 0.3

        # 检查JSON工具调用格式
        # 注意:正则提取JSON工具调用较脆弱,可能匹配到非工具调用的JSON片段。
        # 生产环境建议使用结构化解析(如要求LLM输出固定JSON schema)而非正则。
        tool_calls = re.findall(r'\{.*?"name".*?"arguments".*?\}', trajectory)
        if tool_calls:
            for call in tool_calls:
                try:
                    json.loads(call)
                    score += 0.2
                except json.JSONDecodeError:
                    score -= 0.3  # 格式错误扣分

        return min(max(score, -1.0), 1.0)

    def _tool_usage_reward(self, prompt: str, trajectory: str) -> float:
        """
        Level 2: 工具使用奖励
        评估工具使用的合理性和有效性

        注意:当前使用简单的字符串匹配(如 '<action>' 标签)来检测工具调用,
        可能在trajectory包含讨论工具调用格式的文本时产生误报。
        生产环境建议使用结构化的工具调用记录而非从原始文本中提取。
        """
        score = 0.0

        # 是否使用了工具(大多数Agent任务都需要工具)
        tool_count = trajectory.count("<action>")
        if tool_count > 0:
            score += 0.3

        # 是否重复调用相同工具(不好的行为)
        actions = re.findall(r'<action>(.*?)</action>', trajectory)
        if len(actions) > len(set(actions)):
            score -= 0.2  # 重复调用扣分

        # 是否在观察后调整了策略(好的行为)
        observations = re.findall(r'<observation>(.*?)</observation>', trajectory)
        if len(observations) > 1:
            # 检查后续动作是否与前一个不同(说明Agent在学习)
            score += 0.3

        return min(max(score, -1.0), 1.0)

    def _reasoning_reward(self, prompt: str, trajectory: str) -> float:
        """
        Level 3: 推理质量奖励
        这通常需要LLM-as-Judge来评估
        """
        # 简单规则版本
        score = 0.0

        thoughts = re.findall(r'<think>(.*?)</think>', trajectory, re.DOTALL)

        for thought in thoughts:
            # 推理是否有实质内容(不是废话)
            if len(thought.strip()) > 50:
                score += 0.2
            # 是否引用了之前的观察
            if "observation" in thought.lower() or "result" in thought.lower():
                score += 0.1

        return min(max(score, -1.0), 1.0)

    def _task_completion_reward(self, prompt: str, trajectory: str) -> float:
        """
        Level 4: 任务完成奖励
        通常需要外部验证(如运行测试、检查答案)
        """
        # 这里需要根据具体任务类型来实现
        # 示例:代码生成任务
        if "<final_answer>" in trajectory:
            return 0.5  # 至少给出了答案
        return 0.0

9.3.2 奖励工程的陷阱

Python
# ❌ 常见奖励设计错误

# 错误1: 奖励篇幅而非质量
bad_reward_1 = lambda _, traj: len(traj) / 1000  # lambda匿名函数:_表示忽略第一个参数,只用轨迹长度算奖励
# → Agent会生成冗长无用的内容

# 错误2: 只奖励最终结果,忽略过程
bad_reward_2 = lambda _, traj: 10.0 if "correct" in traj else 0.0  # lambda+三元表达式:包含correct得10分否则0分
# → Agent无法学习中间步骤的好坏

# 错误3: 惩罚过重导致过于保守
bad_reward_3 = lambda _, traj: -100.0 if any_error(traj) else 1.0
# → Agent学会了"什么都不做"以避免惩罚

# ✅ 好的奖励设计原则
"""
1. 过程奖励 > 结果奖励: 每一步都给反馈
2. 密集奖励 > 稀疏奖励: 避免大量0分
3. 相对排名 > 绝对分数: GRPO的核心优势
4. 多维度 > 单一分数: 分开评估格式/工具/推理/结果
5. 可验证 > 主观评价: 尽量用规则而非LLM-as-Judge
"""

9.4 GRPO训练实战

9.4.1 训练一个工具调用Agent

以下是用GRPO训练Agent进行工具调用的完整流程:

Python
"""
GRPO训练工具调用Agent的完整示例

目标: 训练一个Agent学会使用计算器/搜索/代码执行器工具解决数学问题
"""

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from torch.optim import AdamW

# === Step 1: 定义工具环境 ===

class ToolEnvironment:
    """Agent的工具环境"""

    def __init__(self):
        self.tools = {
            "calculator": self._calculator,
            "search": self._search,
            "code_executor": self._code_executor,
        }
        self.tool_descriptions = {
            "calculator": "Evaluate a mathematical expression. Input: expression (str)",
            "search": "Search for factual information. Input: query (str)",
            "code_executor": "Execute Python code and return output. Input: code (str)",
        }

    def _calculator(self, expression: str) -> str:
        try:
            # 安全的数学表达式求值:使用 AST 白名单检查而非字符集过滤
            import ast
            tree = ast.parse(expression, mode='eval')
            for node in ast.walk(tree):
                if not isinstance(node, (ast.Expression, ast.BinOp, ast.UnaryOp,
                                        ast.Constant, ast.Add, ast.Sub, ast.Mult,
                                        ast.Div, ast.Mod, ast.Pow, ast.USub, ast.UAdd)):
                    return "Error: invalid expression"
            return str(eval(compile(tree, '<expr>', 'eval'), {"__builtins__": {}}, {}))
        except Exception as e:
            return f"Error: {e}"

    def _search(self, query: str) -> str:
        # 模拟搜索(实际应用中接入真实搜索API)
        knowledge_base = {
            "pi": "Pi (π) is approximately 3.14159265358979",
            "speed_of_light": "Speed of light is 299,792,458 m/s",
            "earth_radius": "Earth's mean radius is 6,371 km",
        }
        for key, value in knowledge_base.items():
            if key in query.lower():
                return value
        return f"No results found for: {query}"

    def _code_executor(self, code: str) -> str:
        try:
            # 受限的代码执行环境
            local_vars = {}
            # exec动态执行代码字符串;{"__builtins__": {}}清空内置函数以限制权限,local_vars接收执行结果
            exec(code, {"__builtins__": {}}, local_vars)
            return str(local_vars.get("result", "No 'result' variable defined"))
        except Exception as e:
            return f"Execution error: {e}"

    def execute_tool(self, name: str, args: str) -> str:
        if name in self.tools:
            return self.tools[name](args)
        return f"Unknown tool: {name}"

    def get_tool_prompt(self) -> str:
        desc = "\n".join(
            f"- {name}: {desc}"
            for name, desc in self.tool_descriptions.items()
        )
        return f"Available tools:\n{desc}"

# === Step 2: 定义奖励函数 ===

def compute_agent_reward(prompt: str, response: str,
                         environment: ToolEnvironment,
                         expected_answer: str = None) -> float:
    """
    Agent行为的综合奖励函数
    """
    import re
    import json

    reward = 0.0

    # R1: 格式正确性 (±2.0)
    # 检查是否使用了正确的思考-行动格式
    has_thought = bool(re.search(r'<think>.*?</think>', response, re.DOTALL))
    has_action = bool(re.search(r'<action>.*?</action>', response, re.DOTALL))
    has_answer = bool(re.search(r'<answer>.*?</answer>', response, re.DOTALL))

    if has_thought:
        reward += 0.5
    if has_action or has_answer:
        reward += 0.5
    if not (has_thought or has_action or has_answer):
        reward -= 2.0  # 完全不遵循格式

    # R2: 工具调用有效性 (±3.0)
    tool_calls = re.findall(
        r'<action>\s*\{.*?"tool":\s*"(\w+)".*?"input":\s*"(.*?)".*?\}\s*</action>',
        response, re.DOTALL
    )

    valid_calls = 0
    for tool_name, tool_input in tool_calls:
        if tool_name in environment.tools:
            result = environment.execute_tool(tool_name, tool_input)
            if "Error" not in result:
                valid_calls += 1
                reward += 1.0
            else:
                reward -= 0.5
        else:
            reward -= 1.0

    # R3: 答案正确性 (±5.0)
    if expected_answer and has_answer:
        answer_match = re.search(r'<answer>(.*?)</answer>', response, re.DOTALL)
        if answer_match:
            agent_answer = answer_match.group(1).strip()
            if expected_answer.strip().lower() in agent_answer.lower():
                reward += 5.0
            else:
                reward -= 1.0

    # R4: 效率奖励 (±1.0)
    num_steps = len(tool_calls)
    if num_steps <= 3 and reward > 3:
        reward += 1.0  # 高效完成任务
    elif num_steps > 8:
        reward -= 1.0  # 过多步骤

    return reward

# === Step 3: GRPO训练循环 ===

class GRPOTrainer:
    """
    简化版GRPO训练器
    用于演示核心训练逻辑
    """

    def __init__(self, model, tokenizer, environment,
                 group_size=4, lr=1e-5, kl_coeff=0.1):
        self.model = model
        self.tokenizer = tokenizer
        self.env = environment
        self.group_size = group_size
        self.kl_coeff = kl_coeff
        self.optimizer = AdamW(model.parameters(), lr=lr)

        # 参考模型(用于KL散度约束)
        self.ref_model = AutoModelForCausalLM.from_pretrained(
            model.config.name_or_path
        )
        self.ref_model.eval()

    def train_step(self, batch: list[dict]) -> dict:
        """
        GRPO训练的一步

        batch: [{"prompt": str, "expected_answer": str}, ...]
        """
        all_log_probs = []
        all_ref_log_probs = []
        all_rewards = []
        all_advantages = []

        for item in batch:
            prompt = item["prompt"]
            expected = item.get("expected_answer")

            # 1. 为每个prompt生成group_size个回复
            group_rewards = []
            group_log_probs = []
            group_ref_log_probs = []

            for _ in range(self.group_size):
                # 采样生成
                inputs = self.tokenizer(prompt, return_tensors="pt")
                with torch.no_grad():
                    output = self.model.generate(
                        **inputs,
                        max_new_tokens=512,
                        do_sample=True,
                        temperature=0.7,
                        top_p=0.9,
                    )

                response = self.tokenizer.decode(
                    output[0][inputs.input_ids.shape[1]:],
                    skip_special_tokens=True
                )

                # 计算奖励
                reward = compute_agent_reward(
                    prompt, response, self.env, expected
                )
                group_rewards.append(reward)

                # 计算log概率
                log_prob = self._compute_log_prob(inputs, output)
                group_log_probs.append(log_prob)

                ref_log_prob = self._compute_ref_log_prob(inputs, output)
                group_ref_log_probs.append(ref_log_prob)

            # 2. 组内归一化(GRPO的核心)
            rewards_tensor = torch.tensor(group_rewards)
            mean = rewards_tensor.mean()
            std = rewards_tensor.std() + 1e-8
            advantages = (rewards_tensor - mean) / std

            all_advantages.extend(advantages.tolist())
            all_log_probs.extend(group_log_probs)
            all_ref_log_probs.extend(group_ref_log_probs)
            all_rewards.extend(group_rewards)

        # 3. 策略梯度更新 + KL约束
        policy_loss = 0
        kl_loss = 0

        for log_prob, ref_log_prob, advantage in zip(
            all_log_probs, all_ref_log_probs, all_advantages
        ):
            policy_loss -= log_prob * advantage
            kl_loss += (log_prob - ref_log_prob)  # KL散度近似

        total_loss = policy_loss + self.kl_coeff * kl_loss
        total_loss = total_loss / len(all_log_probs)

        self.optimizer.zero_grad()
        total_loss.backward()
        torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
        self.optimizer.step()

        return {
            "loss": total_loss.item(),
            "mean_reward": sum(all_rewards) / len(all_rewards),
            "reward_std": torch.tensor(all_rewards).std().item(),
        }

    def _compute_log_prob(self, inputs, output):
        """计算当前模型的log概率(简化近似)"""
        with torch.enable_grad():
            # 注意:output包含完整序列(prompt+生成),需用它同时作为input和labels
            # 这是粗略近似(包含了prompt部分),精确实现应只计算生成部分的log概率
            # 精确实现应屏蔽prompt token:
            #   prompt_len = inputs.shape[-1]
            #   labels = output.clone()
            #   labels[:, :prompt_len] = -100  # -100为CrossEntropyLoss的忽略索引
            #   outputs = self.model(input_ids=output, labels=labels)
            outputs = self.model(input_ids=output, labels=output)
            return -outputs.loss  # 负的平均交叉熵 ≈ 序列log概率的近似

    def _compute_ref_log_prob(self, inputs, output):
        """计算参考模型的log概率(简化近似)"""
        with torch.no_grad():
            outputs = self.ref_model(input_ids=output, labels=output)
            return -outputs.loss

# === Step 4: 训练数据准备 ===

TRAINING_DATA = [
    {
        "prompt": "What is 15% of 380? Use the calculator tool to compute this.",
        "expected_answer": "57"
    },
    {
        "prompt": "What is the circumference of the Earth in km? Search for Earth's radius and then calculate.",
        "expected_answer": "40030"
    },
    {
        "prompt": "Calculate the compound interest on $1000 at 5% for 3 years.",
        "expected_answer": "1157.625"
    },
]

# === Step 5: 训练主循环 ===

def main():
    """GRPO训练主流程"""
    # 初始化
    model_name = "Qwen/Qwen2.5-1.5B-Instruct"  # 用小模型演示
    model = AutoModelForCausalLM.from_pretrained(model_name)
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    env = ToolEnvironment()

    trainer = GRPOTrainer(
        model=model,
        tokenizer=tokenizer,
        environment=env,
        group_size=4,
        lr=1e-5,
    )

    # 训练循环
    num_epochs = 10
    for epoch in range(num_epochs):
        metrics = trainer.train_step(TRAINING_DATA)
        print(
            f"Epoch {epoch+1}/{num_epochs} | "
            f"Loss: {metrics['loss']:.4f} | "
            f"Mean Reward: {metrics['mean_reward']:.2f} | "
            f"Reward Std: {metrics['reward_std']:.2f}"
        )

    # 保存模型
    model.save_pretrained("./agent-grpo-trained")
    tokenizer.save_pretrained("./agent-grpo-trained")
    print("Training complete! Model saved to ./agent-grpo-trained")

if __name__ == "__main__":
    main()

9.5 Agent环境与Benchmark

9.5.1 主要Agent训练/评估环境

环境 任务类型 工具 适用场景
WebArena Web浏览与操作 浏览器 Web Agent训练
SWE-bench 代码修复 代码编辑/执行 代码Agent训练
GAIA 通用助手任务 搜索/计算/文件 通用Agent评估
AgentBench 多领域任务 多种工具 综合Agent评估
ToolBench 工具调用 API调用 工具使用训练
InterCode 交互式代码 Shell/SQL/Python 代码执行Agent

9.5.2 构建自定义训练环境

Python
from abc import ABC, abstractmethod
from typing import Any

class AgentEnvironment(ABC):  # 抽象基类,定义Agent训练环境接口
    """Agent训练环境的基类"""

    @abstractmethod
    def reset(self, task: dict) -> str:
        """重置环境,返回初始观察"""
        pass

    @abstractmethod
    def step(self, action: str) -> tuple[str, float, bool]:
        """
        执行动作
        Returns: (observation, reward, done)
        """
        pass

    @abstractmethod
    def get_available_tools(self) -> list[dict]:
        """返回可用工具列表"""
        pass

class CodeFixEnvironment(AgentEnvironment):
    """
    代码修复环境 - 用于训练代码调试Agent
    """

    def __init__(self, test_cases: list[dict]):
        self.test_cases = test_cases
        self.current_task = None
        self.current_code = ""
        self.step_count = 0
        self.max_steps = 15

    def reset(self, task: dict) -> str:
        self.current_task = task
        self.current_code = task["buggy_code"]
        self.step_count = 0

        return (
            f"Fix the following code:\n```python\n{self.current_code}\n```\n"
            f"Test description: {task['test_description']}"
        )

    def step(self, action: str) -> tuple[str, float, bool]:
        self.step_count += 1

        if self.step_count >= self.max_steps:
            return "Max steps reached", -2.0, True

        # 解析动作
        if action.startswith("edit:"):
            # 编辑代码
            new_code = action[5:].strip()
            self.current_code = new_code
            return f"Code updated:\n{new_code}", 0.5, False

        elif action.startswith("run_tests"):
            # 运行测试
            passed, total, output = self._run_tests()
            reward = (passed / total) * 5.0 - 1.0
            done = passed == total
            return (
                f"Tests: {passed}/{total} passed\n{output}",
                reward,
                done
            )

        elif action.startswith("read:"):
            # 读取文件
            return f"File content:\n{self.current_code}", 0.0, False

        else:
            return "Unknown action", -1.0, False

    def _run_tests(self) -> tuple[int, int, str]:
        """运行测试用例"""
        passed = 0
        total = len(self.current_task.get("tests", []))
        output = []

        for i, test in enumerate(self.current_task.get("tests", [])):
            try:
                exec_globals = {}
                # exec将Agent修改后的代码与测试代码拼接执行,exec_globals为独立命名空间防止变量污染
                exec(self.current_code + "\n" + test["code"], exec_globals)
                passed += 1
                output.append(f"  ✅ Test {i+1}: PASSED")
            except Exception as e:
                output.append(f"  ❌ Test {i+1}: FAILED - {e}")

        return passed, total, "\n".join(output)

    def get_available_tools(self) -> list[dict]:
        return [
            {"name": "edit", "description": "Edit the code. Usage: edit:<new_code>"},
            {"name": "run_tests", "description": "Run the test suite"},
            {"name": "read", "description": "Read the current code"},
        ]

9.6 前沿方向与展望

9.6.1 2025年Agent RL的最新进展

方向 代表工作 核心思想
Agentic GRPO DeepSeek-R1 用GRPO训练Agent的推理和工具使用能力
Process Reward OpenAI PRM 在每一步给奖励(而非只在最终)
Self-Play Agent Arena Agent之间对抗训练提升能力
Curriculum Learning Adaptive Agent 从简单任务逐渐训练到复杂任务
Multi-Agent RL CAMEL, MetaGPT 训练多个Agent协作完成任务

9.6.2 Agent RL vs 传统RL

Text Only
传统RL (游戏/机器人):
  - 状态空间: 有限/连续
  - 动作空间: 有限/连续
  - 奖励: 即时且明确 (得分/距离)
  - 环境: 确定性或随机
  - 样本效率: 低 (需百万次交互)

Agent RL (LLM Agent):
  - 状态空间: 自然语言 (无限)
  - 动作空间: 自然语言 + 工具调用 (无限)
  - 奖励: 延迟且模糊 (任务完成度)
  - 环境: 高度不确定 (LLM本身有随机性)
  - 样本效率: 更高 (LLM预训练提供了强先验)

关键区别:
  LLM的预训练给Agent提供了强大的"世界知识"先验,
  RL只需要在此基础上学习"如何行动",
  而不需要从零学习"世界是怎样的"。

9.6.3 实践建议

  1. 先SFT,再RL:用少量高质量数据SFT建立基线,再用GRPO提升
  2. 从简单工具开始:先训练Agent使用1-2个简单工具,再逐步增加
  3. 过程奖励 > 结果奖励:给中间步骤也设计奖励信号
  4. 用规则奖励:尽可能用可验证的规则而非LLM-as-Judge
  5. 控制KL散度:防止模型在RL训练中偏离太远

📝 练习

练习1:设计奖励函数(基础)

为一个"问答Agent"设计奖励函数,该Agent可以使用搜索工具回答问题: - 考虑格式、工具使用、答案质量三个维度 - 写出完整的Python实现

练习2:实现简易GRPO(中级)

使用一个小型语言模型(如Qwen2.5-0.5B),实现: - 定义一个简单的数学计算环境 - 实现GRPO的核心训练逻辑 - 训练Agent正确使用计算器工具 - 记录训练过程中的奖励曲线

练习3:多Agent RL(高级)

设计一个双Agent训练场景: - Agent A负责生成代码 - Agent B负责测试和反馈 - 两个Agent通过交互共同提升 - 使用GRPO分别训练两个Agent


📚 参考资料

  1. DeepSeek: "DeepSeek-R1: Incentivizing Reasoning Capability in LLMs via Reinforcement Learning" (2025)
  2. Schulman et al.: "Proximal Policy Optimization Algorithms" (2017) — PPO基础
  3. Rafailov et al.: "Direct Preference Optimization" (2023) — DPO方法论
  4. Shao et al.: "DeepSeekMath: Pushing the Limits of Mathematical Reasoning via Reinforcement Learning" (2024) — GRPO首次提出
  5. Yao et al.: "ReAct: Synergizing Reasoning and Acting in Language Models" (2023)
  6. Wang et al.: "AgentBench: Evaluating LLMs as Agents" (2023)
  7. Zeng et al.: "AgentTuning: Enabling Generalized Agent Abilities for LLMs" (2023)

📝 本章小结

本章系统学习了Agentic RL(Agent强化学习)的核心知识:

  1. ✅ 理解了SFT训练Agent的局限性与RL的核心优势
  2. ✅ 掌握了从SFT到RLHF到DPO到GRPO的训练范式演进
  3. ✅ 理解了GRPO为何特别适合Agent场景(规则奖励 + 组内相对排名)
  4. ✅ 学会了分层Agent奖励系统设计(格式→工具使用→推理质量→任务完成)
  5. ✅ 完成了GRPO训练工具调用Agent的完整实战代码
  6. ✅ 了解了主要Agent训练/评估环境与前沿研究方向

✅ 学习检查清单

  • 能解释SFT和RL训练Agent的核心区别
  • 能说明PPO、DPO、GRPO三种方法的优缺点
  • 能解释GRPO为什么特别适合Agent场景
  • 能设计分层的Agent奖励函数(格式/工具/推理/完成度)
  • 了解奖励工程的常见陷阱(奖励长度、只看结果、惩罚过重)
  • 能实现简化版GRPO训练循环(采样→打分→归一化→策略梯度)
  • 了解主要Agent评估环境(WebArena、SWE-bench、GAIA等)
  • 理解Agent RL与传统RL的关键区别

🔗 下一步

下一章我们将学习GUI Agent,探索如何构建能够操作图形界面的智能代理。

继续学习: 10-GUI Agent


祝你学习愉快! 🎉


最后更新日期:2026-02-12 适用版本:AI Agent开发实战教程 v2026