DeepSeek原生应用与智能体开发实践（人工智能技术丛书）【行情 报价 价格 评测】-京东

王晓华DeepSeek开发AI应用入门书《DeepSeek原生应用与智能体开发实践》全文分享~持续更新-CSDN博客

目录

14.1  基于GRPO的平衡车自动控制实战

14.1.1  CartPole强化学习环境设置

14.1.2  基于GRPO的CartPole模型训练

14.1.3  基于GRPO后的CartPole模型演示

14.2  GRPO算法详解

14.2.1  从PPO对比GRPO

14.2.2  GRPO核心原理与案例演示

14.2.3  GRPO原理的补充问答

14.2.4  平衡车中的GRPO控制详解

14.3  本 章 小 结

---

PPO（Proximal Policy Optimization，近端策略优化）作为强化学习领域的一颗璀璨明星，以其独特的优化策略和出色的性能表现，在众多复杂任务场景中崭露头角。它巧妙地平衡了探索与利用之间的关系，通过限制策略更新的幅度，有效避免了策略在更新过程中出现的大幅波动，使得智能体能够在稳定的学习过程中逐步提升性能。在诸如机器人控制、游戏AI等领域，PPO算法展现出了强大的适应性和高效性，为解决实际问题提供了有力的工具。其通过不断地与环境进行交互，根据反馈信号调整策略，让智能体逐渐学会在复杂的环境中做出最优决策，仿佛为智能体赋予了一双洞察环境奥秘的慧眼。

然而，随着应用场景的不断拓展和任务的日益复杂，PPO算法也面临着一些挑战。例如，在处理高维状态空间和动作空间的问题时，其计算复杂度和样本需求量会显著增加，导致训练效率降低。而且，PPO算法对于超参数的设置较为敏感，不同的超参数组合可能会对算法的性能产生较大影响，这增加了算法调优的难度。

而GRPO（Guided Reinforcement Policy Optimization，引导式强化策略优化）算法则犹如一股新兴的力量，在强化学习的舞台上崭露头角。GRPO算法在继承PPO算法优势的基础上，进行了创新性的改进。它引入了一种引导机制，这种机制能够根据任务的特点和先验知识，为智能体的策略更新提供更有针对性的指导。就像是在黑暗中为智能体点亮了一盏明灯，让它在探索最优策略的道路上少走弯路。PPO与GRPO示意如图14-1所示。

图14-1  PPO与GRPO示意图

在实际应用中，GRPO算法展现出了比PPO算法更优越的性能。以自动控制为例，面对复杂多变的操作环境和瞬息万变的状况，GRPO算法能够更快速地收敛到最优策略，使自动控制更加紧凑和准确。它通过对操作过程中的各种因素进行精准分析和引导，让目标在不同的情况下都能做出最合适的决策。

而且，GRPO算法在处理高维数据和复杂任务时，具有更好的鲁棒性和适应性，能够有效应对各种突发情况和不确定性因素，为强化学习在更多领域的应用开辟了新的道路。未来，随着对GRPO算法研究的不断深入和完善，相信它将在强化学习领域发挥更加重要的作用，推动人工智能技术不断向前发展。

14.1  基于GRPO的平衡车自动控制实战

我们知道，基于PPO算法的火箭回收案例非常经典，从其实现过程可以看到，通过对整体的操作描述和控制，我们可以更好地对火箭降落的全过程进行优化。由于篇幅问题，我们直接把这个火箭回收案例代码放在配套资源中，请读者在学习本节之前，先通过运行案例代码弄清楚PPO算法。本节将延续这一自动火箭回收的经典案例，使用新的强化学习算法GRPO来完成一项新的强化学习控制技术。

14.1.1  CartPole强化学习环境设置

CartPole是用于强化学习的一种常用环境，在CartPole场景中，有一辆小车，智能体的任务是通过左右移动保持车上的杆竖直，若杆的倾斜度数过大，或者车子离初始位置左右的偏离程度过大，或者坚持时间到达最大帧，则游戏结束。在CartPole-V1环境中，最大帧是500。CartPole环境如图14-2所示。

图14-2  CartPole环境

下面是我们完成的一个用于演示CartPole的代码：

import gym

def main():

    env = gym.make('CartPole-v1', render_mode="human")

    for i_episode in range(20):

        observation = env.reset()

        for t in range(100):

            env.render()

            print(observation)

            action = env.action_space.sample()

            observation, reward, done, info, _ = env.step(action)

            if done:

                print("Episode finished after {} timesteps".format(t + 1))

                break

if __name__ == "__main__":

    main()

智能体的状态是一个维数为4的向量，每一维都是连续的，其动作空间是离散的，动作空间大小为2，详情参见表14-1、表14-2所示。

表14-1  CartPole环境的状态空间

维    度	状    态	最  小  值	最  大  值
0	车的位置	-2.4	2.4
1	车的速度	-Inf	Inf
2	杆的角度	-41.8°	41.8°
3	杆尖端的速度	-Inf	Inf

表14-2  CartPole环境的动作空间

标    号	动    作
0	向左移动小车
1	向右移动小车

在游戏中每坚持一帧，智能体能获得分数为1的奖励；坚持时间越长，则最后的分数越高，坚持最大帧即可获得最高的分数。

14.1.2  基于GRPO的CartPole模型训练

接下来，我们将首先使用基于GRPO强化学习方案，完成CartPole模型训练，代码如下所示：

# -*- coding: utf-8 -*-

"""

GRPO (Generalized Reward Policy Optimization) 算法实现

环境：CartPole-v1

功能：训练策略网络平衡小车立杆

"""

# 常用库

import time  # 时间统计

from tqdm import tqdm  # 进度条显示

import matplotlib.pyplot as plt  # 结果可视化

# PyTorch相关

import torch

from torch.nn import functional as F  # 神经网络函数

import gym  # 强化学习环境

from torch.distributions import Categorical  # 分类分布采样

import numpy as np  # 数值计算

class PolicyNet(torch.nn.Module):

    """策略网络定义"""

    def __init__(self, state_dim, action_dim):

        """

        初始化策略网络结构

        :param state_dim: 状态维度 (CartPole为4)

        :param action_dim: 动作维度 (CartPole为2)

        """

        super(PolicyNet, self).__init__()

        self.fc1 = torch.nn.Linear(state_dim, 128)  # 第一全连接层

        self.fc2 = torch.nn.Linear(128, action_dim)  # 第二全连接层

    def forward(self, state):

        """

        前向传播计算动作概率

        :param state: 输入状态 [batch_size, state_dim]

        :return: 动作概率分布 [batch_size, action_dim]

        """

        x = torch.nn.functional.relu(self.fc1(state))  # ReLU激活

        logits = self.fc2(x)  # 未归一化的动作分值

        return F.softmax(logits, dim=1)  # 转换为概率分布

def collect_trajectory_vectorized(envs, policy_net, trajectory_max_steps=500, device="cpu"):

    """

    从并行环境中收集轨迹数据

    :param envs: 并行环境对象 (vectorized environment)

    :param policy_net: 策略网络实例

    :param trajectory_max_steps: 单条轨迹最大步长

    :param device: 计算设备 (cpu/cuda)

    :return: (轨迹数据字典, 各环境总奖励)

    """

    group_size = envs.num_envs  # 并行环境数量

    seed_num = np.random.randint(0, 1000)  # 随机种子

    states, _ = envs.reset(seed=[seed_num] * group_size)  # 环境重置

    # 初始化存储容器

    all_states = []  # 状态序列 [T, group_size, state_dim]

    all_actions = []  # 动作序列 [T, group_size]

    all_log_probs = []  # 对数概率 [T, group_size]

    all_rewards = torch.zeros(group_size)  # 累计奖励 [group_size]

    all_dones = torch.tensor([False] * group_size)  # 终止标记 [group_size]

    # 轨迹收集循环

    for t in range(trajectory_max_steps):

        # 状态转张量

        states_tensor = torch.tensor(states, dtype=torch.float32, device=device)

        # 计算动作概率

        probs = policy_net(states_tensor)  # [group_size, action_dim]

        dist = Categorical(probs)  # 创建分类分布

        actions = dist.sample()  # 采样动作 [group_size]

        log_probs = dist.log_prob(actions).detach()  # 对数概率 [group_size]

        # 环境交互

        next_states, rewards, terminated, truncated, infos = envs.step(actions.cpu().numpy())

        dones = np.logical_or(terminated, truncated)  # 合并终止条件

        # 数据存储

        all_states.append(states)

        all_actions.append(actions)

        all_log_probs.append(log_probs)

        all_dones[dones] = True  # 更新终止标记

        # 奖励处理：终止环境奖励归零 + 位置惩罚

        rewards[all_dones] = 0  # 终止环境奖励置零

        rewards += -abs(next_states[:, 0])  # 添加水平位置惩罚

        all_rewards += rewards  # 累计奖励

        # 状态更新

        states = next_states

        # 提前终止条件：所有环境都终止

        if torch.all(all_dones):

            break

    # 后处理：归一化奖励并组织数据

    normalized_rewards = (all_rewards / trajectory_max_steps).to(device)  # 奖励归一化

    all_states = torch.tensor(all_states).permute(1, 0, 2).to(device)  # [group_size, T, state_dim]

    all_log_probs = torch.stack(all_log_probs).permute(1, 0).to(device)  # [group_size, T]

    all_actions = torch.stack(all_actions).permute(1, 0).to(device)  # [group_size, T]

    # 打包轨迹数据

    trajectories = {

        "all_states": all_states,

        "all_log_probs": all_log_probs,

        "all_actions": all_actions,

        "normalized_rewards": normalized_rewards

    }

    episode_rewards = normalized_rewards * trajectory_max_steps  # 计算实际奖励

    return trajectories, episode_rewards

def calc_advantages_with_grpo(trajectories):

    """

    计算标准化优势值

    :param trajectories: 轨迹数据字典

    :return: 标准化后的优势值 [group_size]

    """

    rewards = trajectories["normalized_rewards"]  # 提取归一化奖励

    mean_reward = torch.mean(rewards)        # 计算均值

    std_reward = torch.std(rewards) + 1e-8  # 计算标准差（防止除零）

    advantages = (rewards - mean_reward) / std_reward  # 标准化

    return advantages

def grpo_update(trajectories, net, optimizer, n_iterations=20, eps=0.2):

    """

    GRPO策略更新

    :param trajectories: 轨迹数据字典

    :param net: 策略网络

    :param optimizer: 优化器

    :param n_iterations: 策略更新迭代次数

    :param eps: PPO截断阈值

    :return: 本轮平均损失值

    """

    # 计算标准化优势值 [group_size, 1]

    advantages = calc_advantages_with_grpo(trajectories).unsqueeze(-1)

    # 解包轨迹数据

    all_states = trajectories["all_states"]  # [group_size, T, state_dim]

    all_log_probs = trajectories["all_log_probs"]  # [group_size, T]

    all_chosen_actions = trajectories["all_actions"]  # [group_size, T]

    batch_size = len(all_states)  # group_size

    # 多轮策略优化

    for i_iter in range(n_iterations):

        loss = 0.0

        # 遍历每个并行环境的轨迹

        for i in range(batch_size):

            # 提取单条轨迹数据

            states = all_states[i]  # [T, state_dim]

            log_probs = all_log_probs[i]  # [T]

            chosen_actions = all_chosen_actions[i]  # [T]

            advantage = advantages[i]  # [1]

            # 计算新策略的对数概率

            new_log_probs = torch.log(net(states).gather(1, chosen_actions.unsqueeze(1)))  # [T, 1]

            # 计算概率比（重要性采样比率）

            ratio = torch.exp(new_log_probs - log_probs.unsqueeze(1))  # [T, 1]

            # 计算替代损失

            surr1 = ratio * advantage  # 未截断项

            surr2 = torch.clamp(ratio, 1 - eps, 1 + eps) * advantage  # 截断项

            trajectory_loss = torch.mean(-torch.min(surr1, surr2))  # 取最小值

            loss += trajectory_loss  # 累计损失

        # 计算平均损失

        loss /= batch_size

        # 反向传播更新参数

        optimizer.zero_grad()

        loss.backward()

        optimizer.step()

    return loss.item()

if __name__ == '__main__':

    """主训练程序"""

    # [1] 环境与网络初始化

    group_size = 10  # 并行环境数量

    env_name = 'CartPole-v1'  # 环境名称

    envs = gym.vector.make(env_name, num_envs=group_size)  # 创建并行环境

    # 获取环境参数

    state_dim = envs.single_observation_space.shape[0]  # 状态维度=4

    n_actions = envs.single_action_space.n  # 动作数量=2

    # 设备配置

    device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")

    # 初始化策略网络和优化器

    policy = PolicyNet(state_dim, n_actions).to(device)

    optimizer = torch.optim.Adam(policy.parameters(), lr=0.02)  # 学习率0.02

    # 训练参数

    episode_num = 50  # 训练轮数

    trajectory_max_steps = 500  # 单轨迹最大步长

    return_list = []  # 奖励记录

    # [2] 训练主循环

    start = time.time()

    for i_episode in tqdm(range(episode_num)):

        # [3] 收集轨迹数据

        trajectories, episode_rewards = collect_trajectory_vectorized(

            envs, policy, trajectory_max_steps, device=device

        )

        # [4] 策略更新

        loss = grpo_update(trajectories, policy, optimizer)

        # [5] 记录性能指标

        avg_reward = sum(episode_rewards) / len(episode_rewards)

        return_list.append(avg_reward.cpu().numpy())

        # 打印训练信息

        print(f'第 {i_episode} 次试验, 平均奖励: {avg_reward:.2f}')

    # [6] 训练后处理

    print("总耗时(s): ", time.time() - start)

    # 保存模型

    save_path = "./grpo_cartpole_policy_update_final.pth"

    torch.save(policy.state_dict(), save_path)

    print(f"模型已保存至: {save_path}")

    # 绘制训练曲线

    plt.figure(figsize=(10, 6))

    plt.plot(return_list)

    plt.xlabel('train epochs')

    plt.ylabel('avg reward')

    plt.title('GRPO on CartPole-v1')

    plt.grid(True)

    plt.show()

    # 关闭环境

    envs.close()

在上面代码中，我们首先创建一个PolicyNet用以对模型的训练，之后的GPRO在过程中学习操作，并根据奖励完成项目既定目标，并将结果进行存储。训练过程请读者自行尝试。

14.1.3  基于GRPO后的CartPole模型演示

模型训练完毕后，为了验证我们的训练任务，需要对基于GRPO后的CartPole模型进行演示，代码如下所示：

# test_cartpole.py

import gym

import torch

import numpy as np

import matplotlib.pyplot as plt

import matplotlib.animation as animation

from argparse import ArgumentParser

# 定义策略网络（必须与训练代码完全一致）

class PolicyNet(torch.nn.Module):

    def __init__(self, state_dim, action_dim):

        super().__init__()

        self.fc1 = torch.nn.Linear(state_dim, 128)

        self.fc2 = torch.nn.Linear(128, action_dim)

    def forward(self, state):

        x = torch.relu(self.fc1(state))

        return torch.softmax(self.fc2(x), dim=1)

def load_model(model_path, device='cpu'):

    """加载训练好的模型"""

    # 初始化网络结构

    model = PolicyNet(state_dim=4, action_dim=2)

    try:

        # 加载训练权重

        model.load_state_dict(torch.load(model_path, map_location=device))

        model.eval()

        print(f"成功加载模型：{model_path}")

        return model

    except Exception as e:

        print(f"模型加载失败：{str(e)}")

        exit(1)

def run_episode(env, model, max_steps=500, render=True):

    """运行单个测试回合"""

    state, _ = env.reset()

    total_reward = 0

    frames = []

    for step in range(max_steps):

        if render:

            frame = env.render()

            if env.render_mode == 'rgb_array':

                frames.append(frame)

        # 使用模型选择动作

        with torch.no_grad():

            state_tensor = torch.FloatTensor(state).unsqueeze(0)

            action_probs = model(state_tensor)

        action = torch.argmax(action_probs).item()

        # 执行动作

        next_state, reward, terminated, truncated, _ = env.step(action)

        total_reward += reward

        state = next_state

        if terminated or truncated:

            print(f"回合结束，步数：{step + 1}，总奖励：{total_reward:.1f}")

            break

    return total_reward, frames

def save_gif(frames, filename, fps=30):

    """保存为GIF动画"""

    plt.figure(figsize=(6, 4))

    plt.axis('off')

    ims = [[plt.imshow(frame, animated=True)] for frame in frames]

    ani = animation.ArtistAnimation(plt.gcf(), ims, interval=50, blit=True)

    ani.save(filename, writer='pillow', fps=fps)

    print(f"动画已保存至：{filename}")

def main():

    # 命令行参数解析

    parser = ArgumentParser(description='CartPole测试程序')

    parser.add_argument('--model', type=str, default='./grpo_cartpole_policy_update_final.pth',

                        help='模型文件路径（默认：./grpo_cartpole_policy_update_final.pth）')

    parser.add_argument('--episodes', type=int, default=5,

                        help='测试回合数（默认：5）')

    parser.add_argument('--render', type=str, choices=['human', 'rgb_array'], default='human',

                        help='渲染模式：human（窗口显示）或 rgb_array（生成帧）')

    parser.add_argument('--save_gif', action='store_true',

                        help='保存为GIF动画（仅在rgb_array模式有效）')

    args = parser.parse_args()

    # 设备设置

    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    # 创建环境

    try:

        env = gym.make('CartPole-v1', render_mode=args.render)

    except gym.error.Error as e:

        print(f"环境创建失败：{str(e)}")

        print("请确保：1.已安装最新gym库 2.确认环境名称正确")

        exit(1)

    # 加载模型

    model = load_model(args.model, device)

    # 运行测试

    total_rewards = []

    best_frames = []

    max_reward = 0

    for ep in range(args.episodes):

        print(f"\n=== 第 {ep + 1}/{args.episodes} 测试回合 ===")

        reward, frames = run_episode(env, model)

        total_rewards.append(reward)

        # 记录最佳表现

        if reward > max_reward and args.render == 'rgb_array':

            max_reward = reward

            best_frames = frames

    # 输出统计信息

    print("\n=== 测试结果 ===")

    print(f"平均奖励：{np.mean(total_rewards):.1f} ± {np.std(total_rewards):.1f}")

    print(f"最佳奖励：{max(total_rewards)}")

    print(f"最差奖励：{min(total_rewards)}")

    # 保存最佳表现动画

    if args.save_gif and args.render == 'rgb_array' and len(best_frames) > 0:

        save_gif(best_frames, "cartpole_demo.gif")

    elif args.save_gif and args.render != 'rgb_array':

        print("警告：--save_gif 仅在rgb_array渲染模式下有效")

    env.close()

if __name__ == '__main__':

    main()

通过对训练好的模型进行演示，我们打印了演示步数及其获取的奖励，如下所示：

回合结束，步数：500，总奖励：500.0

=== 第 2/5 测试回合 ===

回合结束，步数：500，总奖励：500.0

=== 第 3/5 测试回合 ===

回合结束，步数：500，总奖励：500.0

=== 第 4/5 测试回合 ===

回合结束，步数：500，总奖励：500.0

=== 第 5/5 测试回合 ===

回合结束，步数：500，总奖励：500.0

=== 测试结果 ===

平均奖励：500.0 ± 0.0

最佳奖励：500.0

最低奖励：500.0

在这个过程中，我们可以通过注释模型载入参数来观察不同状态下的模型对CartPole的操作，其图像如图14-3所示。读者可以自行尝试运行代码。

图14-3  CartPole的操作

14.2  GRPO算法详解

传统的策略优化方法，比如PPO（Proximal Policy Optimization，近端策略优化），通常会用一个单独的价值模型来估算某个状态的价值。接着，它会利用广义优势估计（GAE）来计算优势值，并基于这些优势来逐步更新策略模型。在这个过程中，策略模型和价值模型是同步进行迭代的，这样做的目的是不断提升价值模型的估算准确度，让策略优化更加有效。

不过，GRPO（这里可以理解为一种改进或变体的策略优化方法）就采取了不一样的做法。它不再依赖单独的价值模型，而是直接根据群体策略所产生的奖励，来计算群体奖励优势。下面是GRPO实现的一份伪代码，如下所示：

# 注意：这不是实际公式。

# 这是一个高度简化的预期目标版本

def grae_advantages(rewards):

    """概念性组相对优势估计（结果监督）。"""

    mean_reward = np.mean(rewards)

    std_reward = np.std(rewards)

    normalized_rewards = (rewards - mean_reward) / (std_reward + 1e-8)

    advantages = normalized_rewards  # 对于结果监督，优势 = 归一化奖励

    return advantages

def grpo_loss(old_policy_logprobs_group, new_policy_logprobs_group, group_advantages, kl_penalty_coef, clip_epsilon):

    """概念性 GRPO 损失函数（对一组响应取平均）。"""

    group_loss = 0

    for i in range(len(group_advantages)):  # 遍历组内的每个相应

        advantage = group_advantages[i]

        new_policy_logprob = new_policy_logprobs_group[i]

        old_policy_logprob = old_policy_logprobs_group[i]

        ratio = np.exp(new_policy_logprob - old_policy_logprob)

        clipped_ratio = np.clip(ratio, 1 - clip_epsilon, 1 + clip_epsilon)

        surrogate_objective = np.minimum(ratio * advantage, clipped_ratio * advantage)

        policy_loss = -surrogate_objective

        kl_divergence = new_policy_logprob - old_policy_logprob

        kl_penalty = kl_penalty_coef * kl_divergence

        group_loss += (policy_loss + kl_penalty)  # 累加组内每个响应的损失

    return group_loss / len(group_advantages)  # 对组内损失取平均

具体来说，GRPO会运用统计手段，在群体策略中找出那些表现优秀、奖励高的策略，以及表现不佳、奖励低的策略。然后，它会调整这些策略的概率，增加高奖励策略被选中的机会，同时减少低奖励策略的使用概率。通过这样的方式，GRPO能够不断地迭代策略模型，直到达到一个稳定、收敛的状态。这种方法简化了模型结构，同时也为策略优化提供了新的思路。

14.2.1  从PPO对比GRPO

从PPO算法实现火箭回收案例（参见配套资源中的代码）的过程来看，PPO就像是教你的LLM一步步走路，确保它在每次更新时不会摔倒。它对LLM的“走路方式”（策略）进行温和的调整。

1. PPO算法回顾与总结

PPO的关键角色：

• 策略（LLM）：我们正在训练的LLM，用于生成更好的文本。
• 奖励模型：根据人类偏好对文本进行打分的AI裁判。
• 价值函数（辅助教练）：另一个AI模型，充当“辅助教练”。它估计每个状态的“好坏”（当前文本生成的前景如何）。这有助于PPO进行更智能的更新。

PPO的训练步骤：

1. 生成文本（Rollout）：LLM（策略）为不同的提示生成大量文本样本。
2. 获取分数（奖励模型）：奖励模型对每个文本样本进行打分。
3. 计算优势（GAE ——“好多少”分数）：这就是GAE的作用！它是一种巧妙的方法，用于计算每个单词选择的优劣，考虑奖励和价值函数的预测。（关于GAE的更多内容见下文！）
4. 优化LLM（策略更新）：我们更新LLM的策略，以最大化一个特殊的PPO目标函数。这个目标函数现在有三个关键部分：

• 鼓励更高奖励：它推动LLM生成能够获得更高分数的文本。
• 限制策略变化（剪切代理目标）：它预防策略在一次更新中变化过大，确保稳定性。
• KL散度惩罚：如果新策略与旧策略偏离太远，它会增加惩罚，进一步增强稳定性。

1. 熵奖励：它还包括一个熵奖励。简单来说，熵衡量LLM文本生成的“随机性”或“多样性”。增加熵奖励可以鼓励LLM更多地探索，而不是总是生成相同、可预测的响应。它有助于防止LLM过早变得“过于确定”，从而错过可能更好的策略。
2. 更新价值函数（辅助教练更新）：训练价值函数成为一个更好的“辅助教练”——更准确地预测不同文本生成的“好坏”。

为什么选择GAE（Generalized Advantage Estimation，广义优势估计），GAE是一种在策略梯度方法中广泛使用的优势函数估计方法，它结合了蒙特卡洛方法和时序差分方法的优点，以达到低方差和低偏差的估计效果。

• 蒙特卡洛（MC）：高方差，低偏差。想象一下等到整个文本生成后再获得奖励，然后将该奖励分配给文本中的每一个单词。就像只有在小狗完成整个“坐下、待命、取回”动作序列后才给予奖励。对整个序列的奖励是准确的，但对单个动作（“坐下”与“待命”与“取回”）的信号非常嘈杂。高方差，所以学习速度慢。
• 时间差分（TD）：低方差，高偏差。想象一下在每个单词生成后给予奖励。“好单词！”“普通单词！”“很棒的单词！”信号不那么嘈杂，学习速度更快。但是，我们只是局部地判断单词，没有考虑整个文本的长期质量。可能会有偏差，可能会错过“大局”。
• GAE ：平衡。广义优势估计（GAE）就像“多步TD”。它考虑了多个步骤（单词）上的奖励，平衡了方差（MC）与偏差（TD）之间的权衡。就像不仅在结束时给予奖励，还在价值函数预测的指导下，为沿途的“小步骤”给予奖励。

下面我们讲解一下GRPO算法。

2. GRPO算法讲解

GRPO是DeepSeek中对PPO的一种聪明的改进，旨在更加高效，尤其是在复杂的推理任务中。GRPO 就像是PPO的精简版。它保留了PPO的核心思想，但去掉了独立的价值函数（辅助教练），使其更轻量、更快速。PPO算法与GRPO示意如图14-4所示。

图14-4  PPO算法与GRPO示意图

1）GRPO的改进

基于组的优势估计（GRAE）。GRPO的魔法成分在于它如何估计优势。它不是使用辅助教练，而是使用一组由LLM生成的相同提示的响应来估计每个响应相对于组内其他响应的“好坏”。

2）GRPO训练流程（简化版）

1. 生成一组响应：对于每个提示，从LLM中生成多个响应的一组。
2. 对组进行打分（奖励模型）：获取组内所有响应的奖励分数。
3. 计算组内相对优势（GRAE组内比较）：通过比较每个响应的奖励与组内平均奖励来计算优势。在组内对奖励进行归一化以得到优势。
4. 优化策略（使用GRAE的PPO风格目标函数）：使用一个PPO风格的目标函数更新LLM的策略，但使用这些组内相对优势。

14.2.2  GRPO核心原理与案例演示

本小节将详解GRPO核心原理，有些暂时无法理解的内容，读者可以参考14.2.3节的内容一并上网查阅。GRPO的核心思想是通过比较同一组内不同策略或动作的相对表现来优化学习过程，而不是依赖传统的价值模型（Critic Model）来评估每个动作的价值。其主要机制‌‌包括分组机制，即对同一提示生成K个响应构成一个组：

其中

为

的奖励值。需要注意，这里的

为每个序列生成的最终结果，即代表模型对输入的序列生成了一个对应当前状态的完整序列输出。而奖励值

则是针对整个序列

进行计算，也就是说优化目标是基于完整序列的奖励值进行策略更新。

在GRPO进行更新时，我们用到组内标准化优势函数，即针对同一个提示生成的一组结果，我们需要计算其内部的优势，如下所示：

其中

为组内平均奖励，而

则为组内标准差。

因此通过这种方法，可以依次计算出在组内所有生成的K个结果的优势分值，并获得其相对优势排名

，其为

在组内的排名，我们用K来表示（从1到K，1为最低，K为最高）。更进一步，我们需要了解相对优势排名：

这样经过变换，将原始的排名改为相对优势得分。我们将优势函数计算为标准化优势与排名优势的加权和：

最终我们获得GRPO目标（损失）函数：

；当

可以看到GRPO目标函数设计通过相对优势降低方差，结合数值差异与排序信息，同时以组均值作为天然基线，无须额外网络。

下面通过一个实际应用案例来理解GRPO。假设一个文本生成任务，同一提示生成K=4个结果：

1. 奖励值：

   

   。
2. 组内均值：

   

   ，标准差

   

   。
3. 排名优势：

   

   第1名（1.0）→

   

   =0.5，此时第4名（0.3）→

   

   =－0.5。

此时对于目标函数，可以根据组内标准化优势函数和相对排名优势函数计算得到：

，

之后再根据GRPO优势函数加权计算可得到：

带入

中，我们假设原有的策略优势得分

，则损失项取得

。

此过程确保策略更新既关注高奖励样本，又避免过度偏离旧策略。

14.2.3  GRPO原理的补充问答

1. 为何不针对每个生成步骤（token）优化

在强化学习生成任务中，直接为每个生成步骤（token）分配奖励存在挑战：

• 延迟奖励（Delayed Reward）：生成任务的质量通常只能在完整序列生成后才能评估（例如，一首诗的整体意境无法通过单个词判断）。
• 稀疏奖励（Sparse Reward）：若仅对部分token分配奖励（如关键词匹配），可能导致优化方向不稳定。
• 计算复杂性：为每个token设计独立的奖励函数会增加计算成本和标注难度。

因此可以说，GRPO方法倾向于使用完整序列的奖励，而非局部token奖励。

2. 理解内部优势和内部相对优势

我们在前面讲到内部相对优势排名，并引用公式如下：

其中

为奖励值

在组内的排名，从1到K，1为最低，K为最高。

这样做的好处在于如下三点：

1. 排名中心化：通过减去

   

   ，将排名中心化到零均值（例如当K=5时，中位数为3）。
2. 归一化范围：除以

   

   后，结果范围被限制在 [−1,1]附近，具体取决于K的奇偶性。
3. 强化相对顺序：模型更关注样本在组内的相对排名，而非绝对奖励值，增强对奖励噪声的鲁棒性。

下面是一个示例。

若K=5，排名为1到5，则：

此时经过此种计算，则我们将得名更改为：

第 1 名：

第 3 名：

第 5 名：

我们对内部优势和内部相对优势做了一个总结对比，如表14-3所示。

表14-3  内部优势和内部相对优势的对比

指    标	计算目标	特    点	适用场景
内部优势	标准化绝对奖励差异	消除组间差异，关注组内相对强度	奖励值稳定且分布均匀时
内部相对优势	强化排名优先级	对噪声鲁棒，直接反映相对顺序	奖励存在偏差或噪声时

通过结合两者，GRPO能更灵活地优化策略，即标准化优势提供数值稳定性，相对优势排名增强对排序的敏感性。

3. 将优势函数计算为标准化优势与排名优势的加权和的好处

我们在计算标准化优势时，其依赖奖励值的绝对数值可能受奖励模型的偏差或噪声影响（例如，奖励模型对某些结果打分偏高或偏低）。同时当组内奖励分布不均匀（如存在极端值）时，标准化后的优势可能不稳定。

而排名优势仅关注排名顺序，忽略奖励值的具体差异（例如，排名相邻的两个结果可能奖励差距极大或极小）。

下面我们来完成一个极端示例，若组内奖励为[1,2,100]（存在极端值），标准化优势可能因标准差

过大而弱化差异，但排名优势仍能明确区分优劣（第3名显著优于前两名）。

通过结合标准化优势和排名优势，GRPO实现了以下目标：

• 信息互补：数值差异与排序信号的双重利用，避免单一指标的局限性。
• 鲁棒性增强：对奖励模型的噪声、偏差和极端值更具容错性。
• 灵活优化：通过调整加权值，适应不同任务场景的需求。
• 稳定训练：控制优势值的分布范围，提升收敛效率。

这种设计在复杂生成任务（如对话、文本生成）中尤为重要，因为奖励模型往往难以完美校准，且生成结果的优劣可能同时依赖数值差异和相对排序。

4. 冷启动时候第一次损失函数的计算

在我们进行第一次损失函数计算时，由于没有“上一次”的优势值存在，因此计算时，我们一般认为的设置

，那么此时的损失函数被简化为：

14.2.4  平衡车中的GRPO控制详解

我们在14.1节完成了使用平衡车控制GRPO，从这个例子可以看到我们通过设定的相同初始状态一次生成多条轨迹，然后统计这些群体轨迹的平均奖励来计算群体优势，提供更新方向，从而实现策略优化。

我们知道Group Computation就是GRPO的核心改进方法。它通过计算n个平均奖励的群体优势来提供更新方向，计算方法如下公式所示：

其中

表示的这是N个平均奖励的均值，而

则为N个平均奖励的标准差，mean是求平均值，有了这个对优势的计算方法后，则采用和PPO相同的更新方法一样。

1. 数据采集部分的讲解

前面14.1节的示例代码中我们对各个步骤进行了定义，首先定义了PolicyNet用于完成对平衡车的控制，而collect_trajectory_vectorized的作用是并行化获取训练数据，根据设定的并行数与训练次数获取多组完整的操作数据。下面我们分别对其进行讲解：

group_size = envs.num_envs # 获取并行环境数量

seed_num = np.random.randint(0, 1000) # 生成随机种子

states, _ = envs.reset(seed=[seed_num] * group_size) # 重置所有环境

参数和方法的作用：

1. group_size：获取并行环境的数量（例如10个并行CartPole环境）。
2. seed_num：生成一个随机种子，用于确保每次重置环境时初始状态的多样性。
3. envs.reset：重置所有环境到初始状态，并为每个环境分配相同的随机种子。虽然种子相同，但每个环境的初始化可能因内部随机数生成机制而不同。

all_states = []          # 存储所有时间步的状态 [T, group_size, state_dim]

all_actions = []         # 存储所有时间步的动作 [T, group_size]

all_log_probs = []       # 存储所有时间步的对数概率 [T, group_size]

all_rewards = torch.zeros(group_size)  # 累计奖励 [group_size]

参数的作用：

1. all_states：记录每个时间步所有环境的观测状态。
2. all_actions：记录每个时间步策略网络输出的动作。
3. all_log_probs：记录每个动作的对数概率（用于后续计算优势值）。
4. all_rewards：记录每个环境的累计奖励。

接下来就是使用for循环对每个环境执行运行过程并采样结果。

    # 轨迹收集循环

    for t in range(trajectory_max_steps):

        # 状态转张量

        states_tensor = torch.tensor(states, dtype=torch.float32, device=device)

        # 计算动作概率

        probs = policy_net(states_tensor)  # [group_size, action_dim]

        dist = Categorical(probs)  # 创建分类分布

        actions = dist.sample()  # 采样动作 [group_size]

        log_probs = dist.log_prob(actions).detach()  # 对数概率 [group_size]

        # 环境交互

        next_states, rewards, terminated, truncated, infos = envs.step(actions.cpu().numpy())

而在环境交互中，我们采用如下代码实现：

next_states, rewards, terminated, truncated, infos = envs.step(actions.cpu().numpy())

上述代码用于执行动作，并获取下一步状态、奖励、终止标记等信息，其参数的作用如下：

1. terminated：环境自然终止（比如杆子倾倒）。
2. truncated：达到最大步长强制终止。

下面是对奖励进行处理：

rewards[all_dones] = 0  # 终止环境的奖励置零

rewards += -abs(next_states[:, 0])  # 添加水平位置惩罚

all_rewards += rewards  # 累计奖励

作用：

1. 终止环境奖励置零：如果环境已终止，后续奖励不再计入。
2. 水平位置惩罚：CartPole的观测状态中，第0维是小车水平位置。添加惩罚项（如-abs(position)）鼓励小车保持在中心位置，以提高稳定性。
3. 累计奖励：将当前步奖励累加到总奖励中。

最后对数据进行归一化处理，如下所示：

normalized_rewards = (all_rewards / trajectory_max_steps).to(device)  # 奖励归一化

all_states = torch.tensor(all_states).permute(1, 0, 2).to(device)     # 调整维度

all_log_probs = torch.stack(all_log_probs).permute(1, 0).to(device)

all_actions = torch.stack(all_actions).permute(1, 0).to(device)

作用：

1. 奖励归一化：将总奖励除以最大步长，使不同长度的轨迹奖励具有可比性。
2. 维度调整：

1. all_states：从[T, group_size, state_dim]调整为[group_size, T, state_dim]，便于按环境索引。
2. all_log_probs和all_actions：从[T, group_size]调整为[group_size, T]，与状态对齐。

2. GRPO策略更新部分的讲解

grpo_update的作用是实现GRPO的策略更新。calc_advantages_with_grpo计算每个轨迹的标准化优势值，用于衡量当前动作相对于平均表现的好坏。

轨迹数据 → 计算优势值 → 遍历轨迹 → 计算新策略概率 → 概率比 → 替代损失 → 反向传播 → 更新策略

         ↑

         |

         +-- 分组标准化优势值

其中的核心是多轮策略优化，代码如下所示：

for i_iter in range(n_iterations):

    loss = 0.0

    # 遍历每个轨迹

    for i in range(batch_size):

        # 提取单条轨迹数据

        states = all_states[i]    # 形状: [T, state_dim]

        log_probs = all_log_probs[i]  # 形状: [T]

        chosen_actions = all_chosen_actions[i]  # 形状: [T]

        advantage = advantages[i]  # 形状: [1]

        # 计算新策略的对数概率

        new_log_probs = torch.log(net(states).gather(1, chosen_actions.unsqueeze(1)))  # 形状: [T, 1]

       

        # 计算概率比（重要性采样比率）

        ratio = torch.exp(new_log_probs - log_probs.unsqueeze(1))  # 形状: [T, 1]

        # 计算替代损失（PPO-Clip）

        surr1 = ratio * advantage    # 未截断项

        surr2 = torch.clamp(ratio, 1 - eps, 1 + eps) * advantage  # 截断项

        trajectory_loss = torch.mean(-torch.min(surr1, surr2))  # 取最小值

        loss += trajectory_loss  # 累计损失

    # 计算平均损失并更新参数

    loss /= batch_size

    optimizer.zero_grad()

    loss.backward()

    optimizer.step()

从上面代码可以看到，我们首先遍历轨迹，对每个轨迹计算损失。之后通过策略网络net计算新动作概率的对数概率，并计算新旧策略的概率比（ratio），衡量策略更新幅度。在细节上我们直接使用概率比作为损失计算，同时截断项（surr2）将概率比限制在[1-eps, 1+eps]之间。最终避免了单步更新过大，保证了策略的稳定。

14.3  本 章 小 结

本章内容在强化学习的探索之旅中更进一步，深入剖析了一种颇具创新性的强化学习算法GRPO。通过将其与广为人知的PPO算法进行细致对比，我们能够清晰地洞察到GRPO算法的独特优势与革新之处。

PPO算法作为强化学习领域的经典之作，以其稳定的策略更新和出色的性能表现而备受赞誉。然而，它也存在一定的局限性，其中比较突出的一点就是需要额外构建一个评价模型来对策略执行的结果进行精准评价。这一额外的模型不仅增加了算法的复杂度，还在一定程度上提升了计算成本，对算法的实时性和可扩展性造成了一定的影响。

而GRPO算法则巧妙地突破了这一限制，它摒弃了PPO算法中额外评价模型的依赖，转而仅依靠对不同输出结果的直接比较来实现策略的优化。具体而言，GRPO算法通过精心设计的梯度更新机制，直接基于策略在不同状态下产生的输出动作及其对应的反馈信号，对策略参数进行精细调整。这种简化的优化方式，不仅显著降低了算法的复杂度，减少了计算资源的消耗，还使得算法能够更加高效地适应不同的环境和任务需求。

在实际应用中，GRPO算法展现出了强大的适应性和高效性。例如，在机器人控制领域，机器人需要在复杂多变的环境中快速做出决策并执行动作，GRPO算法能够凭借其简洁高效的优化机制，使机器人更快地学习到最优策略，提高任务完成的效率和准确性。又如在游戏AI领域，游戏场景瞬息万变，GRPO算法能够实时根据游戏状态调整策略，让游戏AI具备更强的对战能力和决策智慧。

此外，GRPO算法的这一特性还为强化学习的研究和应用开辟了新的思路。它启示我们，在追求算法性能提升的同时，也可以通过简化算法结构和优化计算流程来实现更高效的学习。未来，我们可以进一步探索GRPO算法在不同领域的应用潜力，结合具体场景的特点对其进行改进和拓展，推动强化学习技术在更多领域的落地和发展。同时，也可以借鉴GRPO算法的设计理念，开发更多具有创新性和实用性的强化学习算法，为解决复杂的实际问题提供更有力的支持。