• RLHF（Reinforcement Learning fromHuman Feedback，人类反馈强化学习）

目的：为了让大模型的输出更贴合人类的偏好，拟合有用真实无害的结果。

思维导图

方法对比

发布时间：最初是采用PPO，但是后来采用DPO简化计算，GRPO是最近基于PPO提出。

PPO、DPO、GRPO宽松来说都属于RLHF，但是DPO和GRPO没有显式的使用RL，所以有时候也认为不属于RLHF。

Chat版LLM训练必备环节(Pretrain-SFT-RLHF)

1. Pretrain（预训练）：样本是大规模的无监督数据（如网页、书籍、论文等），目标是让模型学会语言的基本规律（语法、语义、世界知识）。
2. SFT：样本是有标注的问答对数据，通常包括问答、对话、等，目的是让模型按人类想要的方式回答问题。
3. RLHF：样本是问题和好、坏回答，目标让模型按照人类偏好优化回答质量。

SFT不够吗，为什么要做RLHF？

1. SFT无法提供负反馈信号：（用“token1最不喜欢的是token2”SFT后，预测“token1最喜欢的是”可能得到是token2）
2. 正反馈信号粒度太粗：由于SFT的样本是问答对，你想要的只是这个回答，但模型不知道你把这个当做回答主要是因为回答中的哪些点，具体应该落到哪些词上（sft 的 loss 是平均 loss， rlhf 的 loss 是加权 loss。至于怎么加权，去问 reward_model 和 critic_model）。
3. RLHF的核心在于让模型能够“拒绝”错误的输出，从而提升其输出的安全性和稳定性。

背景知识

名词

• prm：progress reward model  过程奖励模型。
• MCTS：蒙特卡洛搜索树
• MC Rollout：蒙特卡洛展开，即在某个状态s，基于某种策略进行模拟（Rollout）直到重终点。
• Close-form的数学问题：有闭式解的数学题，即可以算出具体值的而不是一个接近的解。
• ICL：in context Learning
• 强化学习库有：trl和verl（字节）两种库。
• ray：分布式深度学习框架。
• OpenRLHF：为了解决RLHF单张GPU无法实现4个大模型的训练。它利用Ray将模型分布到多个GPU上，并结合vLLM优化效率，支持超过70B参数模型的全面RLHF训练。

 backbone

文本分类模型一般结构是文本模型+MLP，我们把其中用来提取文本特征的基座叫做backbone。

class AutoModelForSequenceClassification(nn.Module):
    def __init__(self, backbone, num_labels):
        self.backbone = backbone  # 例如 BertModel、LlamaModel
        self.score = nn.Linear(hidden_size, num_labels)  # 分类头

梯度计算

• x-x的梯度是0，x-x.detach()的梯度是1。原因x-x相当于两个变量抵消了，所以梯度优化不需要再优化了。
• x-x.detach()中，由于x.detach()是一个切断梯度的变量，也就相当于给x增加一个写死的值，这个是不影响x的梯度，所以梯度是1。

SFT

缺点：

1. 模型效果严重依赖训练数据质量。
2. 过拟合风险导致重复、知识遗忘、没有创造性。

强化学习

不同算法中优势的计算方式

概念对齐

重要性采样

①为什么要做重要性采样？

利用历史数据： 在强化学习中，训练一个有效的策略通常需要大量的交互数据。为了节省计算成本，我们可能希望使用以前策略（旧策略）生成的数据来改进当前策略，而不必重新收集新数据。直接使用旧数据会有偏差，因为旧策略和新策略可能不同。为了修正这种偏差，我们可以使用重要性采样。这样我们就可以把on-policy（每次更新后，用更新模型重新生成样本）改成off-policy（一次性生成样本，多次更新模型。）

②如何实现的

每个类型发生的概率增加了一个概率比当做权重。（可以参考下面的视频）

第10课-Actor-Critic方法（Part3-重要性采样和Off-Policy Actor-Critic）【强化学习的数学原理】_哔哩哔哩_bilibili

蒸馏（Distillation）

概念：蒸馏是一种模型压缩技术，让一个小模型（学生模型）模仿一个大模型（教师模型）的行为或输出，将大模型的知识“蒸馏”到小模型中，从而在保持性能的同时减少模型大小。

理解：小模型（student-model）不仅学习label而且拟合label在大模型（teacher-model）中的概率分布。

举例：假设teacher模型对一张猫的图片输出概率分布为 [0.7,0.1,0.2]（分别对应猫、狗、鸟），而真实标签是 [1, 0, 0]。学生模型不仅学习 [1, 0, 0]，还学习[0.7,0.1,0.2]这个分布，从而知道“猫”和“狗”在某些特征上可能更相似。

损失函数：

• 软标签蒸馏损失函数（Soft Label Loss）：teacher和student的softmax 输出做 KL 散度（让student学习teacher最后一层的softmax输出）
• 硬标签蒸馏损失函数（Hard Label Loss）：student 对 ground truth 的交叉熵损失，同时使用温度t调节teacher的softmax输出，不同温度时不同label之间的概率会变化

问题影响公式：

• 当 T=1 时，softmax 的行为与普通分类任务一致。
• 当 T>1（温度升高），输出分布变得更加平缓，概率差距缩小，各类别的概率更接近。
• 当 T<1（温度降低），softmax 输出更接近 one-hot 形式，高概率类别的值更大，低概率类别的值更小。

其他：

• 通常 softmax 会带一个温度T>1放大小概率项（常见的设定是 T=2,5,10），信息更丰富
• 除此之外，也可以学习teacher的中间输出层，这样让teacher的中间态的表示和student的中间态分布有一定的相关性，可能效果会更好。

on-policy与off-policy区别

①宽泛理解

On-Policy（一边学一遍执行任务）：这就像一个人一边学习一边执行任务。你在执行任务的同时，也在尝试改进你当前正在使用的方法。如果你发现了一个更好的方法，你会立即尝试在下一次任务中使用它。这种方法更像是“试错”的过程，你不断地改进自己的策略，但你的学习和执行是紧密相关的。
Off-Policy（先学习任务样例，再执行）：这就像你在一边学习，一边观察其他人如何执行任务，然后在自己执行任务时应用你所学到的。你可以从其他人的经验中学习，而不必亲自尝试每个可能的方法。这种方法更加灵活，你可以在学习过程中积累经验，然后在以后的任务中应用这些经验，即使你的策略发生了变化。

• On-Policy 方法严格依赖当前策略生成的数据，数据效率较低，但训练过程更稳定。

• Off-Policy 方法可以利用历史数据，数据效率更高，但可能需要额外的技巧（如目标网络）来稳定训练。

翻遍整个B站！这绝对是2023年讲的最好的强化学习零基础入门到精通完整版教程（含实战源码）_哔哩哔哩_bilibili

②从采样的时间线来理解

状态s时，我们根据Q函数得到at，执行at得到s(t+1)，此时我们会更新Q函数（也就是Q函数中的w），如果下一步动作的a_t+1是始终是根据原始的Q函数计算得出那就是off-policy，如果下一步的动作a_t+1是根据最近一次更新的Q函数计算得出那就是on-policy。（是否是on-policy是看是否用的是最新的Q函数）或者说“使用的策略与实际执行的策略是否相同”

③从实际场景来理解

在强化学习中所用到的样本，是提前一次性生成好（off-policy），还是一边训练模型一边用梯度更新后的模型再产生新样本（on-policy）。

蒙特卡洛

1.蒙特卡洛方法（Monte Carlo，MC）

介绍：指使⽤随机数来解决很多计算问题的⽅法总称。也叫作模拟法、统计试验法。
应用1-用随机打点，统计落在圆内的点数占比来求圆周率。
应用2-求定积分：画一个矩阵，随机打点，统计"函数下方的点数:总点数"来计算积分

2.蒙特卡洛树搜索（MCTS）

介绍：一种用来选出“最优的下一步策略”的算法。
原理：重复多次自博弈，每次选择UCT值最高的策略，进行下一步操作（避免因为执行次数低导致的胜率高胜率高引起的偏差）。最后访问次数最多的节点就是最佳策略节点。
流程：选择子节点->展开子节点->自博弈->记录更新数据。
应用：解决强化学习中的博弈问题

3.UCT：（Upper Confidence Bound Apply to Tree）

介绍：一种树搜索算法，可以解决尝试那些胜率高但是执行度低的策略。
公式：UCT=Q_i/N_i + 常数C*根号(ln(T)/N_i)     Q_i是i节点赢的次数，N_i是i节点访问次数，C是常数，而T是总访问次数。
理解：例如统计ctr的场景，uct_ctr=ctr加上一个权重，这个权重是是一个随着全局曝光数减小的一个。最终按照uct_ctr降序起到的作用就是优先“item曝光次数少但是ctr高的item，检验他们是不是真的高质量，如果后面曝光了几次ctr降了，那就减少曝光。如果ctr没降多少那就继续曝光。”。
意义：解决访问次数小时，样本不置信的问题。
应用：item的冷启曝光、蒙特卡洛树搜索。

蒙特卡洛方法，用最粗暴的方式解最难的题_哔哩哔哩_bilibili

4.蒙特卡洛估计

想求所有序列的最优解计算梯度，只有将所有路径都遍历一遍才能计算出来，但是这往往不可能。因此我们采用了从所有可能的路径中抽样n条来计算梯度，从采样的序列中抽样来模拟全局成为蒙特卡洛估计。但是这样计算来的梯度就会存在方差。

PPO（Proximal Policy Optimization）近端策略优化

PPO通过优势函数A=Q-V，能更准确地评估动作的价值，减少策略更新的方差。

注意：loss的中的某一项r(t)恒为1，和它的梯度是否为零无关，也不能因为这项值固定为 1 就直接从 loss 中剔除。

 数据：

prompt：'史蒂夫低头看去，立刻看到了他们。他也注意到他们正直直地看着他。一个男人满脸胡须，另一个男人拄着拐杖，还有一只外形奇特、毛茸茸的白棕相间的大狗，体型巨大无比。'
chosen："[{'content': '史蒂夫低头看去，立刻看到了他们。他也注意到他们正直直地看着他。一个男人满脸胡须，另一个男人拄着拐杖，还有一只外形奇特、毛茸茸的白棕相间的大狗，体型巨大无比。', 'role': 'user'}, {'content': '那只狗有着长而浓密的卷曲毛发，宽大、鼓胀的嘴巴。', 'role': 'assistant'}]"
rejected："[{'content': '史蒂夫低头看去，立刻看到了他们。他也注意到他们正直直地看着他。一个男人满脸胡须，另一个男人拄着拐杖，还有一只外形奇特、毛茸茸的白棕相间的大狗，体型巨大无比。', 'role': 'user'}, {'content': '那位胡须男人看起来有些眼熟。史蒂夫感到心跳加速。', 'role': 'assistant'}]"

训练流程

样本为question_喜欢的answer_不喜欢的answer和大模型。用样本训练reward模型，然后再用于优化llm。

调试代码

import shutil
import torch
from accelerate import PartialState
from datasets import load_dataset
from transformers import (
    AutoModelForCausalLM,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    HfArgumentParser,
)
from trl_main.trl.trainer import PPOTrainer, PPOConfig,ModelConfig
from trl_main.trl.models import AutoModelForCausalLMWithValueHead
from trl_main.trl.scripts import ScriptArguments
from trl_main.trl.trainer.utils import get_kbit_device_map,get_peft_config,get_quantization_config,SIMPLE_CHAT_TEMPLATE

"""
python -i examples/scripts/ppo/ppo.py \
    --dataset_name trl-internal-testing/descriptiveness-sentiment-trl-style \
    --dataset_train_split descriptiveness \
    --learning_rate 3e-6 \
    --output_dir models/minimal/ppo \
    --per_device_train_batch_size 64 \
    --gradient_accumulation_steps 1 \
    --total_episodes 10000 \
    --model_name_or_path EleutherAI/pythia-1b-deduped \
    --missing_eos_penalty 1.0
"""

if __name__ == "__main__":

    # 解析命令行参数
    # parser = argparse.ArgmentParser(description="PPO Preference Alignment Training")
    parser = HfArgumentParser((ScriptArguments, PPOConfig, ModelConfig))
    script_args, training_args, model_args = parser.parse_args_into_dataclasses()
    # remove output_dir if exists
    shutil.rmtree(training_args.output_dir, ignore_errors=True)
    ################
    # Model & Tokenizer
    ################
    torch_dtype = (
        model_args.torch_dtype if model_args.torch_dtype in ["auto", None] else getattr(torch, model_args.torch_dtype)
    )
    quantization_config = get_quantization_config(model_args)
    model_kwargs = dict(
        revision=model_args.model_revision,
        attn_implementation=model_args.attn_implementation,
        torch_dtype=torch_dtype,
        device_map=get_kbit_device_map() if quantization_config is not None else None,
        quantization_config=quantization_config,
    )

    tokenizer = AutoTokenizer.from_pretrained(
        model_args.model_name_or_path, padding_side="left", trust_remote_code=model_args.trust_remote_code
    )
    tokenizer.add_special_tokens({"pad_token": "[PAD]"})
    if tokenizer.chat_template is None:
        tokenizer.chat_template = SIMPLE_CHAT_TEMPLATE
    value_model = AutoModelForSequenceClassification.from_pretrained(
        training_args.reward_model_path, trust_remote_code=model_args.trust_remote_code, num_labels=1
    ).cpu()
    reward_model = AutoModelForSequenceClassification.from_pretrained(
        training_args.reward_model_path, trust_remote_code=model_args.trust_remote_code, num_labels=1
    ).cpu()
    policy = AutoModelForCausalLM.from_pretrained(
        training_args.sft_model_path, trust_remote_code=model_args.trust_remote_code
    ).cpu()

    peft_config = get_peft_config(model_args)
    if peft_config is None:
        ref_policy = AutoModelForCausalLM.from_pretrained(
            training_args.sft_model_path, trust_remote_code=model_args.trust_remote_code
        ).cpu()
    else:
        ref_policy = None

    ################
    # Dataset
    ################
    dataset = load_dataset(
        script_args.dataset_name, name=script_args.dataset_config, split=script_args.dataset_train_split
    )
    eval_samples = 100
    train_dataset = dataset.select(range(len(dataset) - eval_samples))
    eval_dataset = dataset.select(range(len(dataset) - eval_samples, len(dataset)))
    dataset_text_field = "prompt"

    def prepare_dataset(dataset, tokenizer):
        """pre-tokenize the dataset before training; only collate during training"""

        def tokenize(element):
            outputs = tokenizer(
                element[dataset_text_field],
                padding=False,
            )
            return {"input_ids": outputs["input_ids"]}

        return dataset.map(
            tokenize,
            batched=True,
            remove_columns=dataset.column_names,
            num_proc=training_args.dataset_num_proc,
        )

    # Compute that only on the main process for faster data processing.
    # see: https://github.com/huggingface/trl/pull/1255
    with PartialState().local_main_process_first():
        train_dataset = prepare_dataset(train_dataset, tokenizer)
        eval_dataset = prepare_dataset(eval_dataset, tokenizer)

    ################
    # Training
    ################
    trainer = PPOTrainer(
        args=training_args,
        processing_class=tokenizer,
        model=policy,
        ref_model=ref_policy,
        reward_model=reward_model,
        value_model=value_model,
        train_dataset=train_dataset,
        eval_dataset=eval_dataset,
        peft_config=peft_config
    )
    trainer.train()

    # Save and push to hub
    trainer.save_model(training_args.output_dir)
    if training_args.push_to_hub:
        trainer.push_to_hub(dataset_name=script_args.dataset_name)

    trainer.generate_completions()

源码实现-trl0.15.2

• reward部分

# 4. compute rewards （由KL和score计算rewards，对应架构图中的r）
kl = logprobs - ref_logprobs #[batch,completion_length] policy和reference的KL
non_score_reward = -args.kl_coef * kl
rewards = non_score_reward.clone() # [batch,completion_length] 每个token都有一个reward
actual_start = torch.arange(rewards.size(0), device=rewards.device) #[2]行坐标
actual_end = torch.where(sequence_lengths_p1 < rewards.size(1), sequence_lengths_p1, sequence_lengths) #[2]纵坐标 回答最后的位置(pad_token_id)
rewards[[actual_start, actual_end]] += scores #[2, 53] score是reward模型输出，给每行最后一个位置加上score（最后一token位是score+0.05*KL，其他token位置是0.05KL）

• Advantage部分（GAE-使用value和reward计算Advantage）

# 6. compute advantages and returns GAE方式（使用reward和value计算优势）
lastgaelam = 0
advantages_reversed = []
gen_length = responses.shape[1]
for t in reversed(range(gen_length)): # token维度，从尾部开始遍历
    nextvalues = values[:, t + 1] if t < gen_length - 1 else 0.0  # 第一次先用0初始化v
    delta = rewards[:, t] + args.gamma * nextvalues - values[:, t]  # 使用reward和value计算TD误差 δ=r+γ*v(s_t+1)-v(s) 
    lastgaelam = delta + args.gamma * args.lam * lastgaelam # 基于t+1优势和TD误差,计算t时刻优势，递推公式 tA_t=δ+γ*λ*A_t+1
    advantages_reversed.append(lastgaelam) #得到 A_t
advantages = torch.stack(advantages_reversed[::-1], axis=1)
returns = advantages + values # 计算return也就是Q。 因为A=Q-V所以Q=A+V  

• value_model和policy_model封装到一起

class PolicyAndValueWrapper(nn.Module):
    def __init__(self, policy, value_model) -> None:
        super().__init__()
        self.policy = policy
        self.value_model = value_model
        self.critic_backbone = getattr(value_model, value_model.base_model_prefix)
    #就是 把输入用policy和value_model预测。value_model在policy基础上+一层MLP（映射到维度1）
    def forward(self, **kwargs):
        output = self.critic_backbone(**kwargs)
        logits = self.value_model.score(output.hidden_states[-1])
        return self.policy(**kwargs), logits

• loss部分

with accelerator.accumulate(model):
    micro_batch_end = micro_batch_start + args.per_device_train_batch_size
    micro_batch_inds = mini_batch_inds[micro_batch_start:micro_batch_end]
    mb_advantage = advantages[micro_batch_inds]
    mb_responses = responses[micro_batch_inds]
    mb_query_responses = query_responses[micro_batch_inds]
    mb_logprobs = logprobs[micro_batch_inds] #π_old
    mb_return = returns[micro_batch_inds]
    mb_values = values[micro_batch_inds]
    # model是在PolicyAndValueWrapper，forward返回的logits和value
    output, vpred_temp = forward(model, mb_query_responses, processing_class.pad_token_id) #获取输出logits和价值预测vpred_temp
    logits = output.logits[:, context_length - 1 : -1] #截取回答 [2, 53, 151936]
    logits /= args.temperature + 1e-7  
    new_logprobs = selective_log_softmax(logits, mb_responses) #选出回答上那个token对应的logtis [2, 53]
    new_logprobs = torch.masked_fill(
        new_logprobs, padding_mask[micro_batch_inds], INVALID_LOGPROB
    )
    vpred = vpred_temp[:, context_length - 1 : -1].squeeze(-1) # [2,53] # 到每个token的value
    vpred = torch.masked_fill(vpred, padding_mask_p1[micro_batch_inds], 0)
    vpredclipped = torch.clamp(
        vpred,
        mb_values - args.cliprange_value,
        mb_values + args.cliprange_value,
    )
    vf_losses1 = torch.square(vpred - mb_return) # value模型的输出和return 计算MSE
    vf_losses2 = torch.square(vpredclipped - mb_return)
    vf_loss_max = torch.max(vf_losses1, vf_losses2) #clip  因为他和policy是相反的所以他用的max
    vf_loss = 0.5 * masked_mean(vf_loss_max, ~padding_mask_p1[micro_batch_inds])
    vf_clipfrac = masked_mean(
        (vf_losses2 > vf_losses1).float(), ~padding_mask_p1[micro_batch_inds]
    )
    logprobs_diff = new_logprobs - mb_logprobs #π/π_old  每次只有π是实时查询得到的，π_old是之前保存的
    ratio = torch.exp(logprobs_diff)
    pg_losses = -mb_advantage * ratio
    pg_losses2 = -mb_advantage * torch.clamp(ratio, 1.0 - args.cliprange, 1.0 + args.cliprange)
    pg_loss_max = torch.max(pg_losses, pg_losses2)
    pg_loss = masked_mean(pg_loss_max, ~padding_mask[micro_batch_inds])
    loss = pg_loss + args.vf_coef * vf_loss
    accelerator.backward(loss)
    optimizer.step()
    optimizer.zero_grad()

DPO（Direct Preference Optimization）直接偏好优化

无需环境交互（无需强化学习），通过对比损失+KL散度，让模型在有限空间学会生成更符合人类偏好的回答。

deepseek R1与kimi1.5 硬核解读_哔哩哔哩_bilibili

GRPO（Group Relative Policy Optimization）

算法流程

损失函数

调试代码

from datasets import load_dataset
from peft import LoraConfig
import torch
dataset = load_dataset("data/trl-lib/tldr", split="train")
training_args = GRPOConfig(
    output_dir="output/Qwen2-0.5B-GRPO",
    learning_rate=1e-4,
    logging_steps=10,
    gradient_accumulation_steps=2,
    max_completion_length=128,
    per_device_train_batch_size=4,
    per_device_eval_batch_size=2,
    model_init_kwargs={"torch_dtype": torch.bfloat16},
    num_train_epochs=1,
    save_steps=1000,
)
trainer = GRPOTrainer(
    model="model/Qwen2.5-0.5B-Instruct",
    reward_funcs="model/weqweasdas/RM-Gemma-2B",
    args=training_args,
    train_dataset=dataset,
    peft_config=LoraConfig(task_type="CAUSAL_LM"),
)
trainer.train()
trainer.save_model(training_args.output_dir)

源码实现-trl0.15.2

• _prepare_inputs

    def _prepare_inputs(self, inputs: dict[str, Union[torch.Tensor, Any]]) -> dict[str, Union[torch.Tensor, Any]]:
        mode = "eval" if self.control.should_evaluate else "train"
        if mode == "train": #缓存中有  old_per_token_logps、ref_per_token_logps 等信息
            if self.state.global_step % self.num_iterations == 0:  # 当num_iterations=1的时候，只写不读缓存
                inputs = self._generate_and_score_completions(inputs)
                self._buffered_inputs[self._step % self.args.gradient_accumulation_steps] = inputs 
            else: # 当num_iterations>1的时候，第一个迭代先写缓存，后续2,3,4..都读缓存，因为同一句prompt的后续的π_old和π_ref都和第一次迭代相同。
                inputs = self._buffered_inputs[self._step % self.args.gradient_accumulation_steps]
            self._step += 1
        else:
            # In evaluation, we don't reuse completions across multiple updates, so we don't need to buffer inputs.
            inputs = self._generate_and_score_completions(inputs)
        return inputs

（train每次调用_prepare_inputs获取输入，_prepare_inputs根据情况决定是否调用_generate_and_score_completions）

• _generate_and_score_completions

    def _generate_and_score_completions(
        self, inputs: dict[str, Union[torch.Tensor, Any]]
    ) -> dict[str, Union[torch.Tensor, Any]]:
        device = self.accelerator.device
        prompts = [x["prompt"] for x in inputs] #4个相同问题
        prompts_text = [maybe_apply_chat_template(example, self.processing_class)["prompt"] for example in inputs]
        prompt_inputs = self.processing_class(  #使用tokenizer类，编码，产出input_ids和attention_mask
            prompts_text, return_tensors="pt", padding=True, padding_side="left", add_special_tokens=False
        )
        prompt_inputs = super()._prepare_inputs(prompt_inputs)
        prompt_ids, prompt_mask = prompt_inputs["input_ids"], prompt_inputs["attention_mask"]

        if self.max_prompt_length is not None:
            prompt_ids = prompt_ids[:, -self.max_prompt_length :]
            prompt_mask = prompt_mask[:, -self.max_prompt_length :]

        # Generate completions using either vLLM or regular generation
        if self.args.use_vllm:
            # 如果使用vllm进行推理加速
        else:
            # 普通的推理方式
            # Regular generation path
            with unwrap_model_for_generation(self.model_wrapped, self.accelerator) as unwrapped_model: # self.model_wrapped 是输入模型增加了lora r=8
                prompt_completion_ids = unwrapped_model.generate(                                      # 使用self.model_wrapped，对问题生成batch_size个回答,每个回答不一样
                    prompt_ids, attention_mask=prompt_mask, generation_config=self.generation_config   
                )
            # prompt_completion_ids中包含原有的问题和新增的回答，所以进行切分。得到回答。
            # Compute prompt length and extract completion ids
            prompt_length = prompt_ids.size(1)
            prompt_ids = prompt_completion_ids[:, :prompt_length]
            completion_ids = prompt_completion_ids[:, prompt_length:]

        # Mask everything after the first EOS token  计算eos后的为mask
        is_eos = completion_ids == self.processing_class.eos_token_id
        eos_idx = torch.full((is_eos.size(0),), is_eos.size(1), dtype=torch.long, device=device)
        eos_idx[is_eos.any(dim=1)] = is_eos.int().argmax(dim=1)[is_eos.any(dim=1)]
        sequence_indices = torch.arange(is_eos.size(1), device=device).expand(is_eos.size(0), -1)
        completion_mask = (sequence_indices <= eos_idx.unsqueeze(1)).int()

        # Concatenate prompt_mask with completion_mask for logit computation
        attention_mask = torch.cat([prompt_mask, completion_mask], dim=1)  # (B, P+C)  #合并问题和回答的mask

        logits_to_keep = completion_ids.size(1)  # we only need to compute the logits for the completion tokens

        with torch.no_grad():
            # When using num_iterations == 1, old_per_token_logps == per_token_logps, so we can skip it's
            # computation here, and use per_token_logps.detach() instead.
            if self.num_iterations > 1:
                old_per_token_logps = self._get_per_token_logps(
                    self.model, prompt_completion_ids, attention_mask, logits_to_keep
                )
            else:
                old_per_token_logps = None

            if self.beta == 0.0:
                ref_per_token_logps = None
            elif self.ref_model is not None: 
                ref_per_token_logps = self._get_per_token_logps(
                    self.ref_model, prompt_completion_ids, attention_mask, logits_to_keep
                )
            else: #ref_model可以单独配，如果不配置，就用关闭self.model关闭lora权重来预测
                with self.accelerator.unwrap_model(self.model).disable_adapter():
                    ref_per_token_logps = self._get_per_token_logps(
                        self.model, prompt_completion_ids, attention_mask, logits_to_keep
                    )

        # Decode the generated completions
        completions_text = self.processing_class.batch_decode(completion_ids, skip_special_tokens=True)
        if is_conversational(inputs[0]):#输入是否是对话类型的（list类型）
            completions = []
            for prompt, completion in zip(prompts, completions_text):
                bootstrap = prompt.pop()["content"] if prompt[-1]["role"] == "assistant" else ""
                completions.append([{"role": "assistant", "content": bootstrap + completion}])
        else:#输入是上下文格式的
            completions = completions_text

        rewards_per_func = torch.zeros(len(prompts), len(self.reward_funcs), device=device)
        for i, (reward_func, reward_processing_class) in enumerate(
            zip(self.reward_funcs, self.reward_processing_classes)
        ):
            if isinstance(reward_func, nn.Module):  # Module instead of PretrainedModel for compat with compiled models   #reward_model是NN模型
                reward_func_name = f"reward {reward_func.config._name_or_path.split('/')[-1]}"
            else: #reward_func是函数、或者其他，甚至http接口都可以
                reward_func_name = reward_func.__name__
            with profiling_context(self, reward_func_name): #记录profiling_context中矩阵的耗时
                if isinstance(
                    reward_func, nn.Module
                ):  # Module instead of PretrainedModel for compat with compiled models
                    if is_conversational(inputs[0]):
                        messages = [{"messages": p + c} for p, c in zip(prompts, completions)]
                        texts = [apply_chat_template(x, reward_processing_class)["text"] for x in messages]
                    else:
                        texts = [p + c for p, c in zip(prompts, completions)] #原始 问答文本
                    reward_inputs = reward_processing_class(
                        texts, return_tensors="pt", padding=True, padding_side="right", add_special_tokens=False
                    )
                    reward_inputs = super()._prepare_inputs(reward_inputs) # 编码后input_ids和attention_mask
                    with torch.inference_mode():
                        rewards_per_func[:, i] = reward_func(**reward_inputs).logits[:, 0]  # Shape (B*G,)  输出[4, 1] 应该是batch_size*group   batch_zide是回答数 group是问题数
                else:
                    # Repeat all input columns (but "prompt" and "completion") to match the number of generations
                    keys = [key for key in inputs[0] if key not in ["prompt", "completion"]]
                    reward_kwargs = {key: [example[key] for example in inputs] for key in keys}
                    output_reward_func = reward_func(prompts=prompts, completions=completions, **reward_kwargs)
                    rewards_per_func[:, i] = torch.tensor(output_reward_func, dtype=torch.float32, device=device)

        # Gather the reward per function: this part is crucial, because the rewards are normalized per group and the
        # completions may be distributed across processes
        rewards_per_func = gather(rewards_per_func)  #适配在多进程或分布式训练环境下：  torch.Size([4, 1])

        # Apply weights to each reward function's output and sum
        rewards = (rewards_per_func * self.reward_weights.to(device).unsqueeze(0)).sum(dim=1)

        # Compute grouped-wise rewards
        mean_grouped_rewards = rewards.view(-1, self.num_generations).mean(dim=1)
        std_grouped_rewards = rewards.view(-1, self.num_generations).std(dim=1)

        # Normalize the rewards to compute the advantages
        mean_grouped_rewards = mean_grouped_rewards.repeat_interleave(self.num_generations, dim=0)
        std_grouped_rewards = std_grouped_rewards.repeat_interleave(self.num_generations, dim=0)
        advantages = (rewards - mean_grouped_rewards) / (std_grouped_rewards + 1e-4)  #减均值除方差 从reward得到advantage

        # Slice to keep only the local part of the data
        process_slice = slice(
            self.accelerator.process_index * len(prompts),
            (self.accelerator.process_index + 1) * len(prompts),
        )
        advantages = advantages[process_slice]

        # Log the metrics
        mode = "eval" if self.control.should_evaluate else "train"

        completion_length = self.accelerator.gather_for_metrics(completion_mask.sum(1)).float().mean().item()
        self._metrics[mode]["completion_length"].append(completion_length)

        reward_per_func = rewards_per_func.mean(0)
        for i, reward_func in enumerate(self.reward_funcs):
            if isinstance(reward_func, nn.Module):  # Module instead of PretrainedModel for compat with compiled models
                reward_func_name = reward_func.config._name_or_path.split("/")[-1]
            else:
                reward_func_name = reward_func.__name__
            self._metrics[mode][f"rewards/{reward_func_name}"].append(reward_per_func[i].item())

        self._metrics[mode]["reward"].append(rewards.mean().item())
        self._metrics[mode]["reward_std"].append(std_grouped_rewards.mean().item())

        if self.log_completions and self.state.global_step % self.args.logging_steps == 0:
            prompts_to_log = gather_object(prompts_text)
            completions_to_log = gather_object(completions_text)
            rewards_to_log = rewards.tolist()

            if self.accelerator.is_main_process:
                if is_rich_available():
                    print_prompt_completions_sample(
                        prompts_to_log,
                        completions_to_log,
                        rewards_to_log,
                        self.state.global_step,
                    )
                if self.args.report_to and "wandb" in self.args.report_to and wandb.run is not None:
                    import pandas as pd

                    # For logging
                    table = {
                        "step": [str(self.state.global_step)] * len(rewards),
                        "prompt": prompts_to_log,
                        "completion": completions_to_log,
                        "reward": rewards.tolist(),
                    }
                    df = pd.DataFrame(table)
                    wandb.log({"completions": wandb.Table(dataframe=df)})

        return {
            "prompt_ids": prompt_ids,
            "prompt_mask": prompt_mask,
            "completion_ids": completion_ids,
            "completion_mask": completion_mask,
            "old_per_token_logps": old_per_token_logps,
            "ref_per_token_logps": ref_per_token_logps,
            "advantages": advantages,
        }

计算loss前的准备数据，生成了old_per_token_logs和ref_per_token_logs。

• compute_loss

    @profiling_decorator
    def compute_loss(self, model, inputs, return_outputs=False, num_items_in_batch=None): # 每个question都会调用一次compute_loss
        if return_outputs:
            raise ValueError("The GRPOTrainer does not support returning outputs")
        # Compute the per-token log probabilities for the model

        prompt_ids, prompt_mask = inputs["prompt_ids"], inputs["prompt_mask"]   # prompt_ids 是[4, 395]，每行都相同，把一个question复制了成4行
        completion_ids, completion_mask = inputs["completion_ids"], inputs["completion_mask"] # completion_ids 是[4, 128]，1个问题生成4个回答
        input_ids = torch.cat([prompt_ids, completion_ids], dim=1) #合并 问题-回答
        attention_mask = torch.cat([prompt_mask, completion_mask], dim=1)
        logits_to_keep = completion_ids.size(1)  # we only need to compute the logits for the completion tokens

        per_token_logps = self._get_per_token_logps(model, input_ids, attention_mask, logits_to_keep) # 计算回答序列上每个词，在model上的log概率

        # Compute the KL divergence between the model and the reference model
        if self.beta != 0.0:
            ref_per_token_logps = inputs["ref_per_token_logps"]  #计算回答序列上每个词，在ref_model上的log概率         ，ref_per_token_logps与per_token_logps计算KL散度
            per_token_kl = (                                    #shape不变[4, 128]
                torch.exp(ref_per_token_logps - per_token_logps) - (ref_per_token_logps - per_token_logps) - 1
            )

        # Compute the loss
        advantages = inputs["advantages"]  # [4],也就是优势是回答维度的，值有正有负
        # When using num_iterations == 1, old_per_token_logps == per_token_logps, so we can skip it's computation (see
        # _generate_and_score_completions) and use per_token_logps.detach() instead.
        old_per_token_logps = inputs["old_per_token_logps"] if self.num_iterations > 1 else per_token_logps.detach() #第一个epoch, inputs["old_per_token_logps"]都是none
        coef_1 = torch.exp(per_token_logps - old_per_token_logps) # πΘ 与πΘold 计算策略概率比（先logit相减再exp）  [4, 128]
        coef_2 = torch.clamp(coef_1, 1 - self.epsilon, 1 + self.epsilon)
        per_token_loss1 = coef_1 * advantages.unsqueeze(1)  # 乘优势 [4, 128]
        per_token_loss2 = coef_2 * advantages.unsqueeze(1)
        per_token_loss = -torch.min(per_token_loss1, per_token_loss2) #裁剪和不裁剪取最小
        if self.beta != 0.0:
            per_token_loss = per_token_loss + self.beta * per_token_kl #求和
        loss = (per_token_loss * completion_mask).sum() / completion_mask.sum() #token维度求和

        # Log the metrics
        mode = "eval" if self.control.should_evaluate else "train"

        if self.beta != 0.0:
            mean_kl = ((per_token_kl * completion_mask).sum(dim=1) / completion_mask.sum(dim=1)).mean()
            self._metrics[mode]["kl"].append(self.accelerator.gather_for_metrics(mean_kl).mean().item())

        is_clipped = (per_token_loss1 < per_token_loss2).float()
        clip_ratio = (is_clipped * completion_mask).sum() / completion_mask.sum() #记录有多少触发的clip的占比
        self._metrics[mode]["clip_ratio"].append(self.accelerator.gather_for_metrics(clip_ratio).mean().item())
        return loss

• 数据集在加载的时候就根据total_cnt,给每个question生成了下标，然后按照重复次数和gpu处理个数进行分组，也就是知道了哪个gpu要处理哪些下标。
• 变量中logps全称应该是log-probabilities，也就是对数概率，它是先softmax再取log对数。
• 因为，per_token_logs和old_per_token_logps都是softmax再取log后得到的，每个元素都是负数，现在相减再e的对数，也就是原来的值相除，对应公式中。
• 总结：如果一个模型要进行GRPO强化，只需要准备一个reward模型或者策略就行了+一些question即可。
• 当μ=1时，直接用π-π.detach()，此时得到梯度是1。
• ，所以是-torch.min()。
•  并行训练参数解释：

reward

prompt：

采用system说明格式要求，user中放数据集的question：

SYSTEM_PROMPT = """
按照如下格式生成：
<think>
...
</think>
<answer>
...
</answer>
"""

奖励：

• 格式奖励：

  • 有<THINK> +0.05
  • 有<\THINK>+0.05
  • 有<ANSWER>+0.05
  • 有<\ANSWER>+0.05
  • 有$\boxed{}$ 框住+0.05
• 结果奖励：
  • 全匹配回答是否正确+2，经过化简、转化正确+1.5。
• 语言一致性。

  其他：

• ​​v3的时候，rewardmodel还是一个Learn出来的模型，到r1的时候就已经是一套规则了。
• 【Reward Hacking】：指的是在强化学习中，智能体利用奖励函数中的漏洞或缺陷，通过采取意料之外但能最大化奖励的行为，从而偏离预期目标。这种现象反映了奖励设计的不完善，使得智能体获得高奖励的同时，并未真正完成任务意图。（就是你loss函数不合理，然后模型确实实现loss最小化，但是并不是你想要的效果）

R1训练流程

sft一阶段（冷启动）

收集DeepSeek-R1-Zero的输出结果，人工筛选几k条高质量长推理链数据（待think标签）。

rl一阶段（推理导向的rl）

采用GRPO进行强化，奖励模型：格式规范性、答案正确性、语言一致性。

sft二阶段

60w由上一阶段由模型产生的推理数据（拒绝采样）；20w非推理数据。

rl二阶段（通用对齐的rl）

采用RLHF进行强化，确保安全性与实用性。

• 特定推理能力的SFT和RL，相较R1-Zero，实际上是为了引人人类先验，帮助模型在特定特例场景下控制输出格式提升可读性，SFT有prompt engineering产生的CoT数据、从R1-Zero采样并且人工处理、Markdown格式控制数据、摘要数据收集，RL增加一致性奖励等。
• DeepSeek-R1-Zero 的尝试或者存在的目的，是为了合成高质量的Long-CoT。
• R1-Zero 纯Reasoning RL训练过程中会采样出“Aha Moment”(顿悟)。
• R1论文中尝试了过程奖励，但是没有成功（将问题划分成步骤是困难的；有的中间结果好不好只有最后才知道）
• 使用「拒绝采样」来获取问题的最佳推理回答，即从模型生成的众多回答中选出最好的一个。

R1蒸馏

蒸馏方案：

• 方式一：找一批question，对齐teacher和student上的每个回答上的概率分布（KL散度）
• 方式二：找一批question，用teacher生成answer，使用次数据对student进行SFT。
• 区分：一般第一种效果应该更好，但是要求teacher和student的tokenizer必须相同。

R1蒸馏方案（SFT）：

• 数据准备：收集800k个推理样本。

• DeepSeek-R1 输出：对于每个样本，来自教师模型（DeepSeek-R1）的输出用作学生模型的目标。

• 监督微调（SFT）：学生模型（例如，Qwen-1.5B、Llama-14B）在这 800k 个样本上进行微调，以匹配 DeepSeek-R1 输出。

• 蒸馏模型：学生模型现在被蒸馏成更小的版本，但保留了 DeepSeek-R1 的大部分推理能力。

开放数据参考：https://huggingface.co/datasets/open-thoughts/OpenThoughts-114k

其他

• GRPO没有使用MCTS（蒙特卡洛搜索树）：句子生层的空间太大，难训练Critic模型估计Value。
• 使用蒙特卡洛搜索树可以不用穷举全部的轨迹可能就能优化loss函数，但是问题是可能会与最优路径产生方差。
• 使用蒙特卡洛搜索树得到的结果不如通过强化学习驱动的效果。
• PPO和GRPO都是是基于结果的奖励，让后将优势项分配到之前所有的token上。
• GRPO舍弃了mcts、value function、prm（过程奖励模型）
• veRL和openRLHF比trl更工程性能优化。
• openRLHF：https://github.com/OpenRLHF/OpenRLHF/tree/main

优化

Dr. GRPO（解决两种bias，长短回答bias，难易问题偏差）

GRPO->DAPO->GSPO

DeepSeek-GRM

• Training Scaling：模型越大、训练得越多 → 学得越好。
• Inference Scaling：模型越大 → 用起来效果越好。
• SPCT：DeepSeek团队，通过增加reward model的大小，来提升训练出模型的效果。
• Principle可以提升RM效果。（可以提前认为设定principle，也可以通过模型生成）
• 经过筛选的自我生成的评估准则可以提升 Reward-Model 的效果。
• 最终通过训练一个meta reward model来引导投票过程（具体就是训练一个二分类，表示当前的 principle 和 critique 是否要被用于投票）。对于可用的结果，进行投票（多个得分加和）。
• Inference-time scaling 指的是在推理阶段，不改变模型参数，而通过改变推理方式（比如增加计算量、使用多个提示、重复生成取平均等）来提升模型效果的一种策略。
• SPCT reward模型训练流程：
  • 训练①RFT（rejective fine-tuning）作为冷启动②rule-based online RL（reinforcement learning） 用于强化模型生成评估准则（principle）和推理批判（critique）的能力，后面都用 principle 和 critique 表示。
  • 推理：通过 inference-time scaling 的方式增加 RM 的最终能力。

GROP-微调qwen3

环境：

torch==2.6.0 transformers==4.51.3 vllm==0.8.2 trl==0.15.2 datasets==3.6.0 unsloth==2025.8.1 accelerate==1.1.0 deepspeed==0.15.1 peft==0.12.0 bitsandbytes gradio swanlab

代码：

from unsloth import FastLanguageModel
import torch

# 设置模型的最大序列长度。可以根据需要增加此值以支持更长的推理和上下文。
max_seq_length = 2048

# 设置LoRA的秩（rank）。秩越高，模型可能变得更“智能”，但训练和推理的速度会变慢，显存占用也会增加。
# LoRA是一种参数高效微调技术，通过训练小型的“适配器”矩阵来调整模型，而不是训练全部参数。
lora_rank = 32
model_name = "/code/liuhe/get/model/DeepSeek-R1-Distill-Qwen-7B"

# 从HuggingFace Hub加载预训练模型和分词器
model, tokenizer = FastLanguageModel.from_pretrained(
    # 指定要加载的预训练模型名称。可以是HuggingFace官方模型、本地模型或Unsloth优化后的模型。
    # model_name="Qwen/Qwen3-8B",
    model_name=model_name,
    # 设置模型的最大序列长度，与上面定义的变量一致。
    max_seq_length=max_seq_length,
    # 是否以4位精度加载模型。对于使用LoRA进行16位浮点数训练，此项应设置为False。
    load_in_4bit=False,
    # 是否启用vLLM进行快速推理。GRPO训练中生成多个响应时，此选项能显著提速。
    fast_inference=True,
    # 设置LoRA的最大秩，与上面定义的变量一致。
    max_lora_rank=lora_rank,
    # 设置GPU显存的使用率。如果遇到显存不足（OOM）的错误，可以适当降低此值。
    gpu_memory_utilization=0.7,
)

# 为模型添加PEFT（Parameter-Efficient Fine-Tuning，参数高效微调）配置，这里使用LoRA。
model = FastLanguageModel.get_peft_model(
    model,
    # LoRA的秩(r)，选择任何大于0的数字。建议值为 8, 16, 32, 64, 128。
    r=lora_rank,
    # target_modules 是一个列表，包含要应用LoRA技术的目标模块名称。
    # 通常我们会选择注意力机制中的投影层和前馈网络中的层。
    target_modules=[
        "q_proj", "k_proj", "v_proj", "o_proj",  # 注意力机制中的查询、键、值、输出投影
        "gate_proj", "up_proj", "down_proj",     # 前馈网络中的门控、上行和下行投影
    ],
    # LoRA的alpha参数，通常设置为秩(r)的2倍，这是一种常见的做法，有助于稳定训练。
    lora_alpha=lora_rank * 2,
    # 是否使用梯度检查点技术。'unsloth'表示使用Unsloth的优化版本，可以显著减少训练时的显存占用。
    use_gradient_checkpointing="unsloth",
    # 设置随机种子，以确保实验结果的可复现性。
    random_state=3407,
)
# 定义一些特殊的字符串标记，用于指导模型生成我们想要的格式。
# 这是一种“格式提示”或“模板化”的方法，让模型学会生成带有思考过程和最终答案的结构化输出。

# <start_working_out> 和 <end_working_out> 用于包裹模型的“思考过程”或“解题步骤”。
reasoning_start = "<start_working_out>"
reasoning_end   = "<end_working_out>"

# <SOLUTION> 和 </SOLUTION> 用于包裹模型给出的最终、简洁的答案。
solution_start  = "<SOLUTION>"
solution_end    = "</SOLUTION>"

# 定义系统提示（System Prompt）。这个提示会作为对话的初始指令，告诉模型它的角色和任务。
# 在这里，我们要求模型先进行思考，将过程放在<start_working_out>和<end_working_out>之间，
# 然后再将最终答案放在<SOLUTION>和</SOLUTION>之间。
system_prompt = \
f"""You are given a problem.
Think about the problem and provide your working out.
Place it between {reasoning_start} and {reasoning_end}.
Then, provide your solution between {solution_start}{solution_end}"""

# 打印系统提示，查看其内容。
system_prompt

# 创建一个自定义的聊天模板（Chat Template）。
# 聊天模板使用Jinja2语法，定义了如何将多轮对话（包含system, user, assistant等角色）格式化为单个字符串，
# 以便输入给模型进行训练或推理。
chat_template = \
    "{% if messages[0]['role'] == 'system' %}"\
        "{{ messages[0]['content'] + eos_token }}"\
        "{% set loop_messages = messages[1:] %}"\
    "{% else %}"\
        "{{ '{system_prompt}' + eos_token }}"\
        "{% set loop_messages = messages %}"\
    "{% endif %}"\
    "{% for message in loop_messages %}"\
        "{% if message['role'] == 'user' %}"\
            "{{ message['content'] }}"\
        "{% elif message['role'] == 'assistant' %}"\
            "{{ message['content'] + eos_token }}"\
        "{% endif %}"\
    "{% endfor %}"\
    "{% if add_generation_prompt %}{{ '{reasoning_start}' }}"\
    "{% endif %}"

# 将模板中的占位符替换为我们之前定义的特定字符串。
# 这样做可以使模板适应我们自定义的格式要求。
chat_template = chat_template\
    .replace("'{system_prompt}'",   f"'{system_prompt}'")\
    .replace("'{reasoning_start}'", f"'{reasoning_start}'")

# 将我们创建的自定义聊天模板赋值给分词器（tokenizer）的chat_template属性。
# 这样，之后调用tokenizer.apply_chat_template时，就会使用这个新模板。
tokenizer.chat_template = chat_template

# 使用分词器的apply_chat_template方法来测试我们自定义的聊天模板。
# 输入是一个包含多轮对话的列表，每轮对话是一个字典，包含'role'和'content'。
# tokenize = False 表示我们只想看到格式化后的字符串，而不是token ID。
# add_generation_prompt = True 会在末尾添加生成提示，这里是我们定义的'{reasoning_start}'，
# 引导模型从“思考过程”开始生成回答。
tokenizer.apply_chat_template([
    {"role" : "user", "content" : "What is 1+1?"},
    {"role" : "assistant", "content" : f"{reasoning_start}I think it's 2.{reasoning_end}{solution_start}2{solution_end}"},
    {"role" : "user", "content" : "What is 2+2?"},
], tokenize = False, add_generation_prompt = True)


# 为GRPO训练加载一个新的、更大的数学推理数据集
from datasets import load_dataset
dataset = load_dataset("/code/liuhe/dataset/DAPO-Math-17k-Processed", "default", split = "train")
dataset

# 定义一个函数来提取答案。在这个数据集中，答案直接就是solution字段，
# 但在其他数据集中（如GSM8K），答案可能被####标记包围，这个函数是为此类情况准备的。
def extract_hash_answer(text):
    # if "####" not in text: return None
    # return text.split("####")[1].strip()
    return text

extract_hash_answer(dataset[0]["solution"])

# 再次对新数据集进行格式化，以符合我们的对话模板。
# 这次我们只准备system和user角色的消息，assistant部分将由模型在GRPO训练中生成。
dataset = dataset.map(lambda x: {
    "prompt" : [
        {"role": "system", "content": system_prompt},
        {"role": "user",   "content": x["prompt"]},
    ],
    # 同时提取出标准答案，用于后续的奖励函数评估。
    "answer": extract_hash_answer(x["solution"]),
})

# 查看格式化后的第一条数据。
dataset[0]

# --- 定义GRPO的奖励函数 --- #
# 奖励函数是GRPO的核心，它评估模型生成的回答的质量，并返回一个分数。
# 分数越高，表示生成的回答越好。

import re

# 奖励函数1: 精确格式匹配
# 我们定义一个正则表达式来检查模型的输出是否严格遵循了 <end...><SOLUTION>...</SOLUTION> 的格式。

# 首先，创建一个正则表达式片段，用于匹配可能存在或不存在的EOS（end-of-sentence）令牌。
solution_end_regex = r"</SOLUTION>[\s]{0,}" + \
    "(?:" + re.escape(tokenizer.eos_token) + ")?"

# 编译完整的正则表达式
match_format = re.compile(
    rf"{reasoning_end}.*?"\
    rf"{solution_start}(.+?){solution_end_regex}"\
    rf"[\s]{{0,}}$",
    flags = re.MULTILINE | re.DOTALL
)
match_format


# 奖励函数2: 近似格式匹配
# 这个函数不检查答案是否正确，只检查格式。它计算输出中包含了多少个我们定义的特殊标签。
# 如果模型生成了所有必需的标签（每个一次），它会得到正分；否则会得到负分。
def match_format_approximately(completions, **kwargs):
    scores = []
    for completion in completions:
        score = 0
        response = completion[0]["content"]
        
        # 计算每个标签出现的次数，如果恰好是1次，则加分，否则扣分。
        # 无需奖励<start_working_out>，因为我们总是在提示中预先添加它。
        score += 0.5 if response.count(reasoning_end)   == 1 else -1.0
        score += 0.5 if response.count(solution_start)  == 1 else -1.0
        score += 0.5 if response.count(solution_end)    == 1 else -1.0
        scores.append(score)
    return scores


# 奖励函数3: 检查答案是否正确（字符串匹配）
# 这个函数使用我们之前定义的正则表达式提取模型给出的答案，并与标准答案进行比较。
def check_answer(prompts, completions, answer, **kwargs):
    # 获取用户的问题和模型生成的所有回答
    question = prompts[0][-1]["content"]
    responses = [completion[0]["content"] for completion in completions]

    # 使用正则表达式从每个回答中提取出<SOLUTION>标签内的内容
    extracted_responses = [
        guess.group(1)
        if (guess := match_format.search(r)) is not None else None \
        for r in responses
    ]

    scores = []
    # 遍历每个提取出的答案和对应的标准答案
    for guess, true_answer in zip(extracted_responses, answer):
        score = 0
        # 如果没有提取到答案（格式错误），给予重罚。
        if guess is None:
            scores.append(-2.0)
            continue
        # 如果答案完全正确，给予最高分！
        if guess == true_answer:
            score += 5.0
        # 如果去除空格后答案正确，也给予较高分数。
        elif guess.strip() == true_answer.strip():
            score += 3.5
        else:
            # 对于数值答案，我们也可以奖励近似正确的答案。
            # 如果答案在真实答案的±10%范围内，给予奖励。
            try:
                ratio = float(guess) / float(true_answer)
                if   ratio >= 0.9 and ratio <= 1.1: score += 2.0
                elif ratio >= 0.8 and ratio <= 1.2: score += 1.5
                else: score -= 2.5 # 错误答案给予惩罚
            except:
                # 如果无法转换为浮点数，给予重罚。
                score -= 4.5
        scores.append(score)
    return scores


# 奖励函数4: 检查答案是否为数值（更宽松的数值匹配）
# 定义一个新的正则表达式，专门用于从<SOLUTION>标签中提取数值。
match_numbers = re.compile(
    solution_start + r".*?[\s]{0,}([-]?[\d\.\,]{1,})",
    flags = re.MULTILINE | re.DOTALL
)


# 定义全局变量，用于控制日志打印的频率。
global PRINTED_TIMES
PRINTED_TIMES = 0
global PRINT_EVERY_STEPS
PRINT_EVERY_STEPS = 5

# 奖励函数4的实现
def check_numbers(prompts, completions, answer, **kwargs):
    question = prompts[0][-1]["content"]
    responses = [completion[0]["content"] for completion in completions]

    # 使用新的正则表达式提取数值
    extracted_responses = [
        guess.group(1)
        if (guess := match_numbers.search(r)) is not None else None \
        for r in responses
    ]

    scores = []
    # 为了便于调试，每隔几步打印一次问题、答案、模型响应和提取结果。
    global PRINTED_TIMES
    global PRINT_EVERY_STEPS
    if PRINTED_TIMES % PRINT_EVERY_STEPS == 0:
        print(
            '*'*20 + f"Question:\n{question}", f"\nAnswer:\n{answer[0]}", f"\nResponse:\n{responses[0]}", f"\nExtracted:\n{extracted_responses[0]}"
        )
    PRINTED_TIMES += 1

    for guess, true_answer in zip(extracted_responses, answer):
        # 如果没有提取到数值，给予重罚。
        if guess is None:
            scores.append(-2.5)
            continue
        # 尝试将提取的字符串和标准答案都转换为浮点数进行比较。
        try:
            true_answer = float(true_answer.strip())
            # 移除逗号，如 '123,456' -> '123456'
            guess       = float(guess.strip().replace(",", ""))
            # 如果数值完全相等，给予高分，否则给予惩罚。
            scores.append(3.5 if guess == true_answer else -1.5)
        except:
            # 如果转换失败，不给分也不扣分。
            scores.append(0)
            continue
    return scores


# 再次对数据集进行预处理，为GRPO训练做准备

# 将数据集的prompt部分（system和user消息）分词，并存储token
tokenized = dataset.map(
    lambda x: {"tokens" : tokenizer.apply_chat_template(x["prompt"], add_generation_prompt = True, tokenize = True)},
    batched = True,
)
# 打印第一条数据分词后解码的结果，进行验证
print(tokenizer.decode(tokenized[0]["tokens"]))
# 计算每条prompt的token长度
tokenized = tokenized.map(lambda x: {"L" : len(x["tokens"])})

import numpy as np
# 计算所有prompt长度的90%分位数，以此作为最大prompt长度的参考。
# 这样可以过滤掉极少数过长的prompt，使训练更稳定高效。
maximum_length = int(np.quantile(tokenized["L"], 0.9))
print("Max Length = ", maximum_length)

# 根据计算出的最大长度，过滤数据集。
dataset = dataset.select(np.where(np.array(tokenized["L"]) <= maximum_length)[0])
del tokenized


# 设置GRPO训练的参数

# prompt的最大长度，留一个token的余量
max_prompt_length = maximum_length + 1
# completion（模型生成部分）的最大长度
max_completion_length = max_seq_length - max_prompt_length

from vllm import SamplingParams
# 配置vLLM的采样参数，这些参数在GRPO训练中用于从模型生成多个不同的回答。
vllm_sampling_params = SamplingParams(
    min_p = 0.1, # 最小概率采样（Min-P），忽略概率低于此值的token
    top_p = 1.0, # Top-P（核）采样，从累积概率超过p的最小token集合中采样
    top_k = -1,  # Top-K采样，-1表示不启用
    seed = 3407, # 采样种子，保证可复现性
    stop = [tokenizer.eos_token], # 遇到EOS令牌时停止生成
    include_stop_str_in_output = True, # 在输出中包含停止符
)

from trl import GRPOConfig, GRPOTrainer
# 配置GRPO训练参数
training_args = GRPOConfig(
    vllm_sampling_params = vllm_sampling_params, # 传入vLLM采样参数
    temperature = 1.0, # 生成时的温度，值越高，随机性越强
    learning_rate = 5e-6, # GRPO的学习率通常比SFT小
    weight_decay = 0.01,
    warmup_ratio = 0.1, # 预热步数占总步数的比例
    lr_scheduler_type = "linear",
    optim = "adamw_8bit",
    logging_steps = 1, # 每一步都记录日志，便于观察
    per_device_train_batch_size = 1, # GRPO中，这个值通常会被num_generations覆盖
    gradient_accumulation_steps = 1, # 梯度累积，可以增加到4以获得更平滑的训练
    num_generations = 4, # 每个prompt生成4个不同的回答进行评估，如果显存不足可以减小此值
    max_prompt_length = max_prompt_length, # prompt最大长度
    max_completion_length = max_completion_length, # 生成内容最大长度
    # num_train_epochs = 1, # 训练总轮数
    max_steps = 100, # 为了快速演示，这里只训练100步
    save_steps = 100, # 每100步保存一次模型
    report_to = "swanlab", # 日志报告平台
    output_dir = "outputs", # 模型和日志输出目录
)

# 实例化GRPOTrainer
trainer = GRPOTrainer(
    model = model, # 我们的基础模型
    processing_class = tokenizer, # 分词器
    reward_funcs = [ # 传入我们定义的所有奖励函数
        # match_format_exactly, # 这个函数被注释掉了，因为它太严格了
        match_format_approximately,
        check_answer,
        check_numbers,
    ],
    args = training_args, # 传入训练配置
    train_dataset = dataset, # 训练数据集
)
# 开始GRPO训练
trainer.train()

相关资料

[LLM+RL] 理解 GRPO 公式原理及 TRL GrpoTrainer 代码实现（advantage 与 loss 计算）_哔哩哔哩_bilibili

GRPO进行微调：

https://github.com/aburkov/theLMbook/blob/main/GRPO_From_Scratch_Multi_GPU_DataParallel_Qwen_2_5_1_5B_Instruct.ipynb