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 全部免费领