从 PPO 到 GRPO 的演进 PPO(Proximal Policy Optimization)是 RLHF 的传统方法,但它有几个问题:需要一个 Critic 模型(额外显存)、训练不稳定、超参数敏感。DeepSeek 提出的 GRPO(Group Relative Policy Optimization)通过组内相对比较消除了 Critic 模型,简化了训练同时提升了稳定性。
方法 需要Critic 显存占用 训练稳定性 实现复杂度 PPO ✅ 高 中 高 DPO ❌ 低 高 低 GRPO ❌ 中 高 中 GRPO 核心思想 传统 PPO 用 Critic 网络估计 baseline,GRPO 直接用同一 prompt 的多个采样结果的平均值作为 baseline:
对每个 prompt p: 1. 采样 G 个回复 {r_1, r_2, ..., r_G} 2. 计算每个回复的奖励 {R_1, R_2, ..., R_G} 3. 组内归一化:advantage_i = (R_i - mean(R)) / std(R) 4. 用归一化后的 advantage 更新策略 算法对比 # PPO 的 Advantage 计算 def ppo_advantage(rewards, values, gamma=0.99, lam=0.95): """需要 Critic 网络预测 values""" advantages = [] returns = [] gae = 0 for t in reversed(range(len(rewards))): delta = rewards[t] + gamma * values[t + 1] - values[t] gae = delta + gamma * lam * gae advantages.insert(0, gae) return advantages # GRPO 的 Advantage 计算 def grpo_advantage(rewards_per_group: list): """不需要 Critic,直接用组内统计""" group_mean = np.mean(rewards_per_group) group_std = np.std(rewards_per_group) advantages = [(r - group_mean) / (group_std + 1e-8) for r in rewards_per_group] return advantages GRPO 训练流程 import torch import torch.nn.functional as F from transformers import AutoModelForCausalLM, AutoTokenizer class GRPOTrainer: def __init__(self, model, ref_model, reward_model, tokenizer, config): self.model = model # 策略模型(训练) self.ref_model = ref_model # 参考模型(冻结) self.reward_model = reward_model # 奖励模型 self.tokenizer = tokenizer self.config = config def train_step(self, prompts: list): batch_size = len(prompts) group_size = self.config.group_size # G=8 或 16 all_advantages = [] all_old_logps = [] all_responses = [] # 1. 对每个 prompt 采样 G 个回复 for prompt in prompts: responses = [] rewards = [] for _ in range(group_size): # 生成回复 response = self._generate(prompt, temperature=0.8) responses.append(response) # 计算奖励 reward = self.reward_model.score(prompt, response) rewards.append(reward) # 2. GRPO Advantage 计算 advantages = self._compute_grpo_advantage(rewards) all_advantages.extend(advantages) all_responses.extend(responses) # 3. 计算旧策略的 log prob old_logps = self._compute_logps(self.model, prompts_expanded, all_responses) all_old_logps = old_logps.detach() # 4. PPO-style 更新(多轮 epoch) for epoch in range(self.config.ppo_epochs): new_logps = self._compute_logps(self.model, prompts_expanded, all_responses) ref_logps = self._compute_logps(self.ref_model, prompts_expanded, all_responses) # 5. 计算 Loss loss = self._grpo_loss( new_logps=new_logps, old_logps=all_old_logps, ref_logps=ref_logps, advantages=torch.tensor(all_advantages), beta=self.config.kl_coef # KL 散度惩罚 ) loss.backward() self.optimizer.step() self.optimizer.zero_grad() return loss.item() def _grpo_loss(self, new_logps, old_logps, ref_logps, advantages, beta=0.04): # Ratio ratio = torch.exp(new_logps - old_logps) # Clipped ratio (PPO clip) clipped_ratio = torch.clamp(ratio, 1 - self.config.clip_range, 1 + self.config.clip_range) # Policy loss policy_loss = -torch.min(ratio * advantages, clipped_ratio * advantages).mean() # KL penalty (GRPO 使用 k3 估计器) kl = (torch.exp(ref_logps - new_logps) - (ref_logps - new_logps) - 1).mean() return policy_loss + beta * kl def _compute_grpo_advantage(self, rewards: list): """GRPO 核心创新:组内相对优势""" mean_r = np.mean(rewards) std_r = np.std(rewards) advantages = [(r - mean_r) / (std_r + 1e-8) for r in rewards] return advantages 奖励函数设计 GRPO 的效果很大程度上取决于奖励函数。DeepSeek R1 使用了多种奖励信号:
...