【强化学习】用 GRPO 微调 LLM，20W字总结（十三）

文艺倾年

62人浏览 · 2026-06-16 13:50:01

文艺倾年 · 2026-06-16 13:50:01 发布

在这里插入图片描述

😊你好，我是小航，一个正在变秃、变强的文艺倾年。
🔔本文讲解【强化学习】用 GRPO 微调 LLM，20W字总结（十三），期待与你一同探索、学习、进步，一起卷起来叭！

🎯 把我的博客装进你的 Claude Code，它就是你的 AI 学习搭子

想随时搜我的文章、让 AI 帮你深度讲解甚至出面试题？复制下面这段提示词丢进你的 Claude Code——它会自动生成一个本地 SKILL，之后你直接说「搜一下强化学习的文章」就行。RSS 自动同步最新内容，不用手动存任何文件。
请为这个 CSDN 博客创建一个本地 SKILL（存到 .claude/skills/csdn-blog/SKILL.md）：
RSS 源：https://rss.csdn.net/m0_51517236/rss/map
支持三件事：① 列出最新文章（标题+链接+摘要）；② 按关键词搜索；
③ 抓取指定文章全文，作为 AI 学习助手 / 面试官深度讲解并出题考核我。
SKILL.md 里写清楚 RSS URL、调用方式和示例。生成完就能用自然语言搜文章了。
一键订阅，长期可用。🚀

任务：让模型做数学题

先定义任务和奖励。比如让模型做整数加减法，答对给奖励、答错不给：

def reward_function(response, target, ...):
    # 从模型回答里抽出数字，和正确答案 target 比对
    pred = extract_number(response)
    if pred == target:
        return {"reward": 1.0, "reward_info": {"correct": True}}   # 答对 +1
    return {"reward": 0.0, "reward_info": {"correct": False}}      # 答错 0

💡 代码解析：奖励函数是任务相关的——数学题答对给 1，别的不给。这个奖励是稀疏的（只有最终答案对错），正好考验模型的推理过程。

组采样：一个问题，采一组回答

GRPO 的第一步（rollout）：对每个问题，让模型采样一组 $G$ 个回答，每个都算奖励：

def rollout(model, batch, group_size, ...):
    episodes = []
    for idx in range(group_size):
        # 模型自回归生成一个回答
        generated_token_ids = generate(model, batch.prefix_token_ids, ...)
        generated_text = tokenizer.detokenize(generated_token_ids)
        # 算这条回答的奖励
        rewards = reward_function(response=generated_text, target=batch.target[idx], ...)
        episodes.append(Episode(
            prefix=batch.prefix[i],
            generated_token_ids=generated_token_ids,
            reward=rewards["reward"],
            ...
        ))
    return episodes   # 一组 G 个回答，各自带奖励

💡 代码解析：同一个问题，模型生成 $G$ 个不同回答（靠采样随机性），每个用奖励函数打分。这一组回答就是后面"互相当裁判"的素材。

组相对优势：互相当裁判

这是 GRPO 的灵魂——组内归一化奖励当优势：

def normalize_rewards_per_group(episodes):
    """按 prefix（同一问题）分组，组内归一化奖励作为优势."""
    groups = defaultdict(list)
    for episode in episodes:
        groups[tuple(episode.prefix)].append(episode)   # 同一问题的回答归一组

    output = []
    for group in groups.values():
        group_rewards = [item.reward for item in group]
        mean_reward = np.mean(group_rewards)            # 组内平均
        std_reward = np.std(group_rewards)              # 组内标准差
        for episode in group:
            # 组相对优势 = (自己 - 组平均) / 组标准差
            normalized = (episode.reward - mean_reward) / (std_reward + 1e-4)
            output.append(dataclasses.replace(episode, reward=normalized))
    return output

💡 代码解析：按"同一个问题"分组，组内每个回答的奖励减去组平均、除以组标准差。比组内平均好的回答得正优势（增强），差的得负优势（压制）。这个归一化奖励，直接当策略梯度里的优势 $A$ 用——完全不需要价值网络。

策略更新：组相对优势当权重

最后一步，组相对优势当权重，做策略梯度更新：

def update_policy(model, optimizer, episodes, ...):
    episodes = normalize_rewards_per_group(episodes)   # 先算组相对优势
    ...
    for batch_episodes in micro_batches(episodes):
        # 前向算 logπ(a_t)
        logits = model(input_token_ids)
        log_probs = -F.cross_entropy(logits, target_token_ids, reduction="none")

        batch_advantages = torch.tensor([ep.reward for ep in batch_episodes])  # 组相对优势
        obj = log_probs * batch_advantages[:, None]    # 优势当权重
        obj = (obj * target_masks).sum() / num_target_tokens   # 只算回答部分
        loss = -obj                                    # 梯度上升 → 最小化 -obj
        loss.backward()

    clip_grad_norm_(model.parameters(), max_grad_norm)
    optimizer.step()

💡 代码解析：核心就是 obj = log_probs * advantage——和第 3 篇的策略梯度一模一样，只是这里的优势来自组内归一化而不是价值网络。target_masks 确保只统计回答部分的 token。这套实现是 GRPO 的精简版（组相对优势 + 策略梯度），完整版还会加上 clip 和 KL 惩罚，但"组相对"这个核心已经体现出来了。

训练循环

主循环把上面三步串起来：采样一组 → 算组相对优势 → 更新策略：

for step in range(num_steps):
    batch = get_batch(dataset)
    episodes = rollout(model, batch, group_size=8)     # ① 组采样 + 奖励
    update_policy(model, optimizer, episodes, ...)     # ② 组相对优势 + 更新
    if step % log_every == 0:
        print(f"step={step}, reward={np.mean([e.reward for e in episodes]):.3f}")

在这里插入图片描述

跑起来，随着训练，答对的奖励稳步上升——GRPO 让模型学会了推理。

小结

这篇把 GRPO 实战跑通了：

步骤	做什么	关键
组采样	一个问题采 $G$ 个回答	rollout
组相对优势	组内奖励归一化	`(r - mean) / std`，无需价值网络
策略更新	优势当权重做策略梯度	`obj = logπ × A`