Python技术迷

Python 实现 GRPO 简版

前几天加班到快十二点,组里那谁问我:“哥,你懂那个 GRPO 不?我看 OpenAI 在论文里提过,好像挺牛的样子。” 我一听这个名字就笑了,其实 GRPO(Group Relative Policy Optimization)真没想象中那么玄乎,本质上就是对 PPO(Proximal Policy Optimization)的一种改良版,优化的重点在于分组归一化奖励,让策略更新更稳定。

咱们今天就用 Python 写个小 demo,看懂原理就行,别太在意细节。

1. 先说点人话:GRPO 到底干嘛的

简单说吧,PPO 是强化学习里一个很火的算法,它的思路是“不要一步走太远”,每次更新策略的时候,对新旧策略的比例加个限制,不让更新过猛。

那 GRPO 呢?它在 PPO 的基础上,把样本分组后再标准化奖励。 这听着像废话,其实是关键:因为有时候一批样本的奖励分布不稳定,比如有的任务奖励特别高,有的特别低,这会让梯度更新方向乱飘。分组标准化之后,相当于每个小组自己比,不被其他组干扰。

2. 环境准备

咱不搞太复杂的环境,就拿 Gym 里的 CartPole-v1 小车平衡杆来练手。

import gym
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from collections import deque

3. 策略网络:别整花活

模型别搞复杂,直接用两层全连接层,输出动作分布的概率。

classPolicyNet(nn.Module):
def__init__(self, obs_dim, act_dim):
        super(PolicyNet, self).__init__()
        self.fc1 = nn.Linear(obs_dim, 64)
        self.fc2 = nn.Linear(64, act_dim)

defforward(self, x):
        x = torch.relu(self.fc1(x))
return torch.softmax(self.fc2(x), dim=-1)

4. 和 PPO 不一样的地方:分组归一化

重点来了。GRPO 在计算优势函数(advantage)的时候,不是直接把所有样本混在一起归一化,而是先分组,比如每 32 个样本为一组,在组内标准化。

这个操作很小,但非常关键,尤其在任务分布不均的时候。

defgroup_normalize(advantages, group_size=32):
    n = len(advantages)
    groups = [advantages[i:i+group_size] for i in range(0, n, group_size)]
    norm_adv = []
for g in groups:
        g = (g - np.mean(g)) / (np.std(g) + 1e-8)
        norm_adv.extend(g)
return np.array(norm_adv)

5. 训练循环:像打游戏一样调

接着就是标准的强化学习套路,收集一批数据 → 计算奖励 → 更新策略。 PPO 的核心在于 “clip”,防止过度优化,GRPO 一样有,只是多了上面那个分组步骤。

deftrain_grpo(env_name="CartPole-v1", epochs=1000, gamma=0.99, clip=0.2, lr=3e-4):
    env = gym.make(env_name)
    obs_dim = env.observation_space.shape[0]
    act_dim = env.action_space.n
    policy = PolicyNet(obs_dim, act_dim)
    optimizer = optim.Adam(policy.parameters(), lr=lr)

for epoch in range(epochs):
        obs_list, act_list, rew_list, logp_old_list = [], [], [], []
        obs = env.reset()[0]
        done = False
        ep_rew = 0
whilenot done:
            obs_t = torch.tensor(obs, dtype=torch.float32)
            probs = policy(obs_t)
            dist = torch.distributions.Categorical(probs)
            action = dist.sample()
            next_obs, reward, done, _, _ = env.step(action.item())

            obs_list.append(obs)
            act_list.append(action.item())
            rew_list.append(reward)
            logp_old_list.append(dist.log_prob(action).item())

            obs = next_obs
            ep_rew += reward

# 计算折扣奖励
        returns = []
        G = 0
for r in reversed(rew_list):
            G = r + gamma * G
            returns.insert(0, G)
        returns = np.array(returns)
        advantages = returns - np.mean(returns)
        advantages = group_normalize(advantages, group_size=16)

# 更新策略
        obs_t = torch.tensor(np.array(obs_list), dtype=torch.float32)
        act_t = torch.tensor(act_list)
        old_logp_t = torch.tensor(logp_old_list)
        adv_t = torch.tensor(advantages, dtype=torch.float32)

        probs = policy(obs_t)
        dist = torch.distributions.Categorical(probs)
        logp = dist.log_prob(act_t)
        ratio = torch.exp(logp - old_logp_t)

        loss = -torch.mean(torch.min(
            ratio * adv_t,
            torch.clamp(ratio, 1 - clip, 1 + clip) * adv_t
        ))

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

if epoch % 50 == 0:
            print(f"Epoch {epoch} | Reward: {ep_rew:.1f}")

6. 跑起来试试

直接跑 train_grpo() 就能看到小车在晃晃悠悠地学会平衡。 如果你把 group_normalize() 改成全局归一化,会发现奖励曲线波动更大,也更容易不稳定。

7. 再说点体会

有时候你以为是算法问题,其实只是数据分布不均。 GRPO 这个改进,表面看只是“多加了个分组归一化”,但它背后反映的是一种训练稳定性的思路 —— 别让极端样本主导梯度方向。

强化学习的很多优化其实都是围绕这个原则在打转,比如 reward scaling、entropy bonus、clip ratio,这个 GRPO 也一样。

结尾

总结一句话吧: GRPO 是 PPO 的一个小进化版,核心在“分组归一化优势”。在 Python 里实现其实不难,难的是理解“为什么这么做”。

下次有人问你“GRPO 是什么鬼”,你就能笑着回一句: “哦,就是个带分组的 PPO。” 😏

-END-

我为大家打造了一份RPA教程,完全免费:songshuhezi.com/rpa.html

🔥虎哥私藏精品🔥

虎哥作为一名老码农,整理了全网最全《python高级架构师资料合集》,总量高达650GB,点击下方公众号回复关键字 python 全部免费领