Skip to content

10. 后训练实战:用真实 Qwen3.5 权重跑后训练 ​

本章说是实战,但实际上依旧是体验为主;真跑后训练完全不可行。

概述 ​

上一章用 TinyLM demo 讲清了 SFT/DPO/GRPO/PPO/Thinking/Agentic 的算法原理。本章把这些算法跑在真实 Qwen3.5 权重 + 真实数据集 + 真实训练循环上——也就是 posttrain/ 目录里的实际代码。

posttrain/
├── common.py           # 教学 TinyLM 后端(第 9 章 demo 用)
├── real_common.py      # 真实训练公共工具(所有阶段的地基)
├── adapter.py          # 后训练产物 Qwen3.5-0.8B → MiniQwen 适配(第 8 章)
├── sft/                # data.py + train_sft.py      (真实 SFT)
├── dpo/                # data.py + dpo_train.py      (真实 DPO)
├── grpo/               # data.py + grpo_train.py     (真实 GRPO)
├── ppo/                # data.py + ppo_train.py      (真实 PPO)
├── thinking/           # data.py + thinking_train.py (真实 Thinking)
├── agentic_rl/         # data.py + agentic_rl_train.py(真实 Agentic RL)
└── demo/               # 六个教学 demo(第 9 章)

真实阶段和 demo 的算法完全一致,差异只在:模型(Qwen3.5-0.8B vs TinyLM)、tokenizer(BPE vs 字符级)、数据(真实数据集 vs 手写几条)、工程设施(bf16/checkpoint/梯度累积 vs 无)。

10.1 真实训练的公共地基:real_common.py ​

代码文件

所有真实阶段都依赖 real_common.py。它提供:种子、设备、tokenizer、模型加载、chat 模板、logprob 计算、checkpoint、导出。

模型加载:只从真权重出发 ​

python
def load_model(from_pretrained, device):
    """从 checkpoint 目录加载 MiniQwenForCausalLM"""
    # 配置读取顺序: config.json > config.pt > miniqwen_config.pt > 默认
    # 权重优先 model.safetensors,否则 model.pt
    if not os.path.isdir(from_pretrained):
        raise FileNotFoundError(f"模型目录不存在: {from_pretrained}")
    ...
    return model, config

真实训练必须从真权重出发(FileNotFoundError 提示),不会像 demo 那样随机初始化。

chat 模板:RL 阶段的 prompt 前缀 ​

python
def build_user_context(tokenizer, user_text, system_prompt=None):
    """构造 system(如有)+ user 的聊天上下文文本(不含 assistant 头)"""
    ...

def build_prompt_prefix(tokenizer, user_text, system_prompt=None):
    """RL/偏好数据的 prompt 前缀:build_user_context + '<|im_start|>assistant\n'"""
    ...

注意 build_prompt_prefix 不用 add_generation_prompt=True——Qwen3.5 模板会在 assistant 头后塞一个空 think 块,非 thinking 模型不需要。DPO/GRPO/PPO/Agentic 数据层共用这个前缀。

生成 + 旧策略 logp(RL 的核心) ​

python
def generate_with_logprobs(model, tokenizer, prompt_ids, max_new_tokens=64,
                           temperature=1.0, top_p=0.9, eos_ids=None, device=None):
    """自回归生成,记录采样分布下每生成 token 的 logp(旧策略 logp)
    返回 (ids, old_logps, stop_id)
    - ids: prompt + 生成完整列表
    - old_logps: 每生成 token 的 logπ_old
    - stop_id: 未到 max_new 就停则为该 eos id,跑满为 None"""
    ...

def batch_sequence_logprobs(model, ids, attention_mask=None, temperature=1.0):
    """对整条 batch 序列算每位置预测 token 的 logp(带梯度,当前策略重算用)
    logp_shift[b,t] = log π_θ(ids[b,t+1] | ids[b,:t+1]),与 SFT loss 的 shift 语义一致"""
    ...

这两个函数是 GRPO/PPO 的关键:rollout 时缓存旧策略 logp,重算时得到当前策略 logp,两者之差就是概率比 ρ = πθ/πold。

checkpoint 与导出 ​

python
def save_checkpoint(path, model, optimizer, step, loss, extra=None):
    """保存 {step, loss, model_state_dict, optimizer_state_dict, *extra}"""
    ...

def export_hf_model(model, tokenizer, output_dir, export_name=None):
    """导出 HF 可读格式(config.pt + config.json + model.pt + tokenizer),供下一阶段 --from_pretrained"""
    ...

每个阶段结束用 export_hf_model 导出,下一阶段的 --from_pretrained 直接指到这个目录。

10.2 阶段总览:数据与默认链路 ​

| 阶段 | 数据文件 | 训练脚本 | 默认 base 权重 | 关键超参 | |------|----------|----------|------|----------------|----------| | SFT | data/sft_2048.jsonl | sft/train_sft.py | 预训练产物 | seq_len=1024, lr=1e-5, epochs=1 | | DPO | data/dpo_truthy.jsonl | dpo/dpo_train.py | SFT 导出 | steps=100, lr=1e-6, beta=0.1 | | GRPO | data/grpo_math.jsonl | grpo/grpo_train.py | DPO 导出 | steps=60, g=4, lr=1e-6 | | PPO | data/grpo_math.jsonl(复用) | ppo/ppo_train.py | DPO 导出 | steps=60, lr=1e-6 | | Thinking | data/thinking_gsm8k.jsonl | thinking/thinking_train.py | DPO 导出 | steps=150, lr=1e-6 | | Agentic RL | data/agentic_math.jsonl | agentic_rl/agentic_rl_train.py | DPO 导出 | steps=120, g=6, lr=1e-6 |

真实训练链路:

预训练产物 (checkpoints/miniqwen_pretrain/miniqwen_final)
    │  bash run.sh sft
    ▼
SFT (checkpoints/miniqwen_sft)
    │  bash run.sh dpo
    ▼
DPO (checkpoints/qwen35_dpo)
    │  bash run.sh grpo / ppo / thinking / agentic
    ▼
GRPO / PPO / Thinking / Agentic

run.sh 的 pick_model 实现"上一阶段导出存在则用,否则回退后训练 adapter":

10.3 SFT:真实对话数据微调 ​

数据:data.py ​

代码文件

python
IM_START_ID = 248045        # <|im_start|> 单 token
ASSISTANT_HEADER = [74455, 198]   # 'assistant\n' 紧跟 im_start 后两 token
SYSTEM_PROMPT = "You are a helpful assistant."

class SFTDataset(torch.utils.data.Dataset):
    def __init__(self, tokenizer, data_path, seq_len=2048, max_samples=None,
                 mask_history=False, strip_think=True):
        ...  # 预 tokenize 全部样本
    def _tokenize_one(self, sample):
        # 1. 解析对话格式 → apply_chat_template(tokenize=False, add_generation_prompt=False)
        # 2. 可选 strip_empty_think:剥掉 chat 模板自动插入的空 think 块(\n\n\n\n)
        # 3. encode;超长直接跳过(不硬截断,避免丢 <|im_end|>)
        # 4. find_assistant_segments 定位每个 assistant 回复段
        # 5. labels 默认全 -100,把 assistant 段 [start:end] 填为 ids

关键点:

  • find_assistant_segments:在 token 序列里定位每个 IM_START_ID,若紧跟 ASSISTANT_HEADER,则该段是 assistant 轮。多轮对话每一轮 assistant 都参与 loss(对齐 LLaMA-Factory 的做法)。
  • --mask_history:只保留最后一段 assistant 参与 loss(只训最后一轮)。
  • --strip_think(默认开):剥掉 apply_chat_template 自动插入的空 think 块(\n\n\n\n),非 thinking 模型需要。
  • 超长直接跳过而非截断:截断会丢掉 <|im_end|>,破坏对话结构。

训练:train_sft.py ​

代码文件

训练循环:

python
for epoch in range(start_epoch, args.epochs):
    for step, (input_ids, labels) in enumerate(train_loader):
        with AmpContext(args.bf16, device):
            loss = model(input_ids=input_ids, labels=labels)["loss"] / grad_accum_steps
        loss.backward()
        if should_update:                     # 梯度累积到步
            lr = get_lr(global_step, total_steps, args.lr)
            for g in optimizer.param_groups: g["lr"] = lr
            clip_grad_norm_(model.parameters(), args.grad_clip)
            optimizer.step(); optimizer.zero_grad(set_to_none=True)
            global_step += 1
        ...

结束导出:save_checkpoint(final.pt) + export_sft_model(config.pt + config.json + model.pt + tokenizer)。

运行:

bash
bash run.sh sft              # 单卡,默认 from PRETRAIN_FINAL
bash run.sh sft_2gpu         # 2 卡 DDP
# 或直接:
python posttrain/sft/train_sft.py \
    --data_path data/sft_2048.jsonl \
    --from_pretrained checkpoints/miniqwen_pretrain/miniqwen_final \
    --output_dir checkpoints/miniqwen_sft \
    --seq_len 1024 --batch_size 1 --epochs 1 --lr 1e-5 --bf16

10.4 DPO:偏好数据直接对齐 ​

数据:data.py ​

代码文件

python
class DPOPreferenceDataset(torch.utils.data.Dataset):
    def __init__(self, tokenizer, data_path, max_samples=None, system_prompt=None, max_seq_len=2048):
        ...  # 预 tokenize 全部偏好对
    def _tokenize_one(self, t):
        prompt_text = build_dpo_prompt(tokenizer, system, question)   # build_prompt_prefix
        p_ids = tokenizer.encode(prompt_text)
        c_ids = p_ids + tokenizer.encode(t["chosen"])     # chosen 完整
        r_ids = p_ids + tokenizer.encode(t["rejected"])   # rejected 完整
        start = len(p_ids)
        # 空回复或超长跳过

chosen/rejected 共享同一个 prompt 前缀(含 system + <|im_start|>assistant\n),各自独立编码。

训练:dpo_train.py ​

代码文件

算法核心(dpo_loss):

python
def dpo_loss(policy, ref, device, beta, c_ids, c_mask, r_ids, r_mask, starts, c_ends, r_ends):
    # 当前策略(带梯度)+ 参考模型(冻结,no_grad)
    lp_theta_c = batch_sequence_logprobs(policy, c_ids, c_mask)
    lp_theta_r = batch_sequence_logprobs(policy, r_ids, r_mask)
    with torch.no_grad():
        lp_ref_c = batch_sequence_logprobs(ref, c_ids, c_mask)
        lp_ref_r = batch_sequence_logprobs(ref, r_ids, r_mask)

    # 只取回复段 logp 均值(归一化,prompt 不进 margin)
    r_theta_c = masked_response_mean(lp_theta_c, starts, c_ends)
    r_theta_r = masked_response_mean(lp_theta_r, starts, r_ends)
    r_ref_c = masked_response_mean(lp_ref_c, starts, c_ends)
    r_ref_r = masked_response_mean(lp_ref_r, starts, r_ends)

    # DPO loss:σ 括号里是 "chosen 优于 rejected 的对数比 margin"
    margin = (r_theta_c - r_ref_c) - (r_theta_r - r_ref_r)
    loss = -F.logsigmoid(beta * margin).mean()
    return loss, margin.mean().item()

与第 9 章 demo 的区别:demo 用整条序列 logp 之和,真实版用 masked_response_mean 只取回复段 logp 均值——这是真实工程里更稳妥的做法(prompt 长度不影响 margin,长回复不被不公平均压低)。

argparse:--from_pretrained(默认 POSTTRAIN_ADAPTER_CKPT)、--ref_model(默认深拷贝冻结)、--data_path(默认 DPO_DATA)、--steps 100、--batch_size 2、--lr 1e-6、--beta 0.1。

运行:

bash
bash run.sh dpo
# 或:
python posttrain/dpo/dpo_train.py \
    --from_pretrained checkpoints/miniqwen_sft/miniqwen_sft \
    --data_path data/dpo_truthy.jsonl \
    --output_dir checkpoints/qwen35_dpo \
    --steps 100 --batch_size 2 --lr 1e-6 --beta 0.1 --bf16

10.5 GRPO:可验证任务强化学习 ​

数据:data.py ​

代码文件

python
def extract_final_answer(answer):
    """GSM8K 的 answer 是 '推理\n#### 42',只取 #### 之后"""
    return answer.split("####")[-1].strip() if "####" in answer else answer.strip()

class GRPOTaskDataset(torch.utils.data.Dataset):
    def __init__(self, tokenizer, data_path, max_samples=None, system_prompt=None, max_seq_len=2048):
        ...  # 预 tokenize 全部 prompt
    def _tokenize_prompt(self, t):
        prompt_ids = tokenizer.encode(build_prompt_prefix(tokenizer, t["prompt"], system))
        ...

GRPO 是 on-policy:__getitem__ 只返回 {prompt_ids, answer, prompt},回复不缓存——每步现场采样 rollout。

训练:grpo_train.py ​

代码文件

算法核心(grpo_step):

python
def grpo_step(policy, ref, tokenizer, device, prompt_ids, answer, g, max_new,
              temperature, top_p, beta, eps):
    # 1. rollout:同一 prompt 采 g 条回复,缓存旧策略逐 token logp
    rollouts = [generate_with_logprobs(policy, tokenizer, prompt_ids, max_new,
                                       temperature, top_p, device=device) for _ in range(g)]
    prompt_len = len(prompt_ids)
    rewards = [rule_based_reward(tokenizer.decode(ids[prompt_len:]), answer) for ids, _lp, _s in rollouts]
    rewards_t = torch.tensor(rewards, dtype=torch.float)

    # 2. 组内归一化优势
    adv = (rewards_t - rewards_t.mean()) / (rewards_t.std() + 1e-4)

    # 3. clip 比率 loss + KL(每个样本 backward 累积梯度)
    policy.train()
    total = 0.0
    for (ids, old_lp, _s), a in zip(rollouts, adv):
        cur = sequence_logprobs(policy, ids, prompt_len, temperature=temperature, device=device)
        old = torch.tensor(old_lp, dtype=torch.float, device=device)
        ratio = (cur - old).exp()                     # ρ = πθ/πold
        surr = torch.min(ratio * a, torch.clamp(ratio, 1 - eps, 1 + eps) * a)
        with torch.no_grad():
            ref_lp = sequence_logprobs(ref, ids, prompt_len, temperature=temperature, device=device)
        kl = (cur - ref_lp).mean()                    # 近似 KL(πθ‖πref)
        loss = -surr.mean() + beta * kl
        loss.backward()
        total += loss.item()
    return total / len(rollouts), rewards, sum(1 for r in rewards if r > 0)

argparse:--steps 60、--batch_size 2、--g 4、--lr 1e-6、--beta 0.1(KL)、--eps 0.2(clip)、--max_new 64、--temperature 1.0、--top_p 0.9。

运行:

bash
bash run.sh grpo
# 或:
python posttrain/grpo/grpo_train.py \
    --from_pretrained checkpoints/qwen35_dpo/qwen35_dpo \
    --data_path data/grpo_math.jsonl \
    --output_dir checkpoints/qwen35_grpo \
    --steps 60 --batch_size 2 --g 4 --lr 1e-6 --bf16

10.6 PPO:Actor-Critic 强化学习 ​

数据:复用 GRPO ​

代码文件

python
# 文件头注释:
# 默认数据 data/grpo_math.jsonl(GSM8K train 前 1000 题)与 GRPO 共用——
# PPO 的奖励也是"回复里出现最终答案 → 1 分",所以用同一份可验证任务。

PPO 的 PPOTaskDataset 结构和 GRPO 几乎相同,但数据来源是 GRPO_DATA(data/grpo_math.jsonl),不是单独的 ppo 数据。

训练:ppo_train.py ​

代码文件

算法核心(ppo_step):

python
class ValueHead(nn.Module):
    """Critic:从 hidden state 预测回报"""
    def __init__(self, d):
        super().__init__()
        self.net = nn.Linear(d, 1)
    def forward(self, hidden):      # (T, D) → (T,)
        return self.net(hidden).squeeze(-1)

def ppo_step(policy, value, tokenizer, device, prompt_ids, keyword, max_new,
             temperature, top_p, clip_eps, ent_coef):
    prompt_len = len(prompt_ids)
    # 1. rollout 一条回复,缓存旧策略 logp
    ids, old_logps, _stop = generate_with_logprobs(policy, tokenizer, prompt_ids, max_new,
                                                   temperature, top_p, device=device)
    # 2. 奖励(末尾给分)
    R = 1.0 if keyword in tokenizer.decode(ids[prompt_len:]) else 0.0

    # 3. 一次 backbone 前向拿 hidden,再推 logits(带梯度,只用跑一次)
    policy.train()
    inp = torch.tensor([ids], dtype=torch.long, device=device)
    hidden = policy.model(inp)                                # MiniQwenModel 返回 hidden
    logits = policy.lm_head(hidden) / temperature             # (1, T, V)
    logp = F.log_softmax(logits, dim=-1)
    cur = logp[0, prompt_len - 1:-1].gather(-1,
          torch.tensor(ids[prompt_len:], device=device).unsqueeze(-1)).squeeze(-1)
    ratio = (cur - torch.tensor(old_logps, device=device)).exp()

    # 4. 价值预测 + 优势 A = R - V
    v_t = value(hidden[0, prompt_len - 1:-1])                 # (L,)
    adv = torch.tensor(float(R), device=device) - v_t.detach() # 蒙特卡洛优势

    # 5. 三部分 loss
    pol_loss = -(torch.min(ratio * adv, torch.clamp(ratio, 1 - clip_eps, 1 + clip_eps) * adv)).mean()
    val_loss = F.mse_loss(v_t, torch.full_like(v_t, R))
    ent = -(logp.exp() * logp).sum(-1)[0, prompt_len - 1:-1].mean()
    loss = pol_loss + 0.5 * val_loss - ent_coef * ent
    return loss, R, ent.item()

真实版 vs demo 的一个关键差异:真实 MiniQwenForCausalLM.forward 只返回 {"loss", "logits"}(不返回 hidden),所以 PPO 直接调 policy.model(inp)(MiniQwenModel 返回 hidden_states)再 policy.lm_head(hidden) 推 logits。demo 的 TinyLM forward 直接返回 hidden 字段。

value head 单独存 checkpoint:save_checkpoint(..., extra={"value_state_dict": value.state_dict()}),因为 value head 不是 policy 的一部分。

argparse:--steps 60、--batch_size 2、--lr 1e-6、--clip_eps 0.2、--ent_coef 0.01、--data_path(默认 GRPO_DATA)。

运行:

bash
bash run.sh ppo
# 或:
python posttrain/ppo/ppo_train.py \
    --from_pretrained checkpoints/qwen35_dpo/qwen35_dpo \
    --data_path data/grpo_math.jsonl \
    --output_dir checkpoints/qwen35_ppo \
    --steps 60 --batch_size 2 --lr 1e-6 --bf16

10.7 Thinking:推理微调 ​

数据:data.py ​

代码文件

python
class ThinkingDataset(torch.utils.data.Dataset):
    def __init__(self, tokenizer, data_path, max_samples=None, system_prompt=None, max_seq_len=2048):
        ...  # 动态取 <|im_start|> 的 token id
    def _tokenize_one(self, s):
        ctx = build_user_context(tokenizer, s["prompt"], system)   # system+user,不含 assistant 头
        if s["kind"] == "THINK":
            text = ctx + f"<|im_start|>think\n{s['reasoning']}<|im_end|>\n" \
                       + f"<|im_start|>assistant\n{s['answer']}<|im_end|>\n"
        else:  # DIRECT
            text = ctx + f"<|im_start|>assistant\n{s['answer']}<|im_end|>\n"
        second = _second_im_start(ids)   # 第二个 <|im_start|>(think/assistant 头)
        ...

def collate_thinking(items, pad_id):
    """批量 padding + label 掩码:labels[:, :second] = -100(system/user 不监督)"""

load_reasoning_samples 把数据分成两类:有 reasoning 列 → THINK(推理样本),只有 answer → DIRECT(直接回答样本)。

训练:thinking_train.py ​

代码文件

算法本质是 SFT(交叉熵),只是监督起点从第二个 <|im_start|> 开始:

python
labels = ids.clone()
labels[labels == pad_id] = -100
labels[:, :second] = -100        # 只监督 think/assistant 头之后

argparse:--steps 150、--batch_size 2、--lr 1e-6、--seq_len 2048、--log_interval 30。

运行:

bash
bash run.sh thinking
# 或:
python posttrain/thinking/thinking_train.py \
    --from_pretrained checkpoints/qwen35_dpo/qwen35_dpo \
    --data_path data/thinking_gsm8k.jsonl \
    --output_dir checkpoints/qwen35_thinking \
    --steps 150 --batch_size 2 --lr 1e-6 --bf16

10.8 Agentic RL:工具调用的强化学习 ​

数据:data.py ​

代码文件

python
TOK_TOOL = "<|tool|>"; TOK_TOOL_END = "<|tool_end|>"
TOK_ENV = "<|env|>";    TOK_ENV_END = "<|env_end|>"

def format_warmup_trajectory(tokenizer, expr, result):
    """一条完整期望轨迹的 token ids(不含 prompt 前缀):
    <|tool|>calc(expr)<|tool_end|><|env|>result<|env_end|>\n result<|im_end|>\n"""
    ...

class AgenticTaskDataset(torch.utils.data.Dataset):
    def __init__(self, tokenizer, data_path, max_samples=None, system_prompt=None, max_seq_len=2048):
        ...  # 预 tokenize 全部 prompt

训练:agentic_rl_train.py ​

代码文件

两阶段训练:

  1. SFT 热身(warmup_trajectory):教模型完整的多步轨迹(调工具 → 读 env 结果 → 作答),到 loss≈0.05 就停。只教工具调用不够——模型看到 env 结果后只会重复调用,必须把"作答"也教了。

  2. RL(agentic_step_group):组级 REINFORCE,组内均值当 baseline。

python
def rollout_trajectory(policy, tokenizer, device, prompt_ids, expr, result, max_new, temp_ans):
    start = len(prompt_ids)
    # 回合1:生成工具调用(低温度 0.6;tool_end 非单 token,文本层截断)
    gen1, _lp1, _stop = generate_with_logprobs(policy, tokenizer, prompt_ids, max_new,
                                               temperature=0.6, top_p=0.9, device=device)
    gen1 = cut_at_marker(tokenizer, gen1[start:], TOK_TOOL_END)
    ids = prompt_ids + gen1
    turn1_end = len(ids)
    ok, res = simulate_env(tokenizer.decode(ids[start:]))
    R = -1.0 if not ok else 1.0                    # 没调工具 → 大惩罚,防"闭嘴"坍缩
    if ok:
        ids = ids + tokenizer.encode(f"{TOK_ENV}{res}{TOK_ENV_END}\n")
    # 回合2:看到环境结果后作答(高温度探索)
    ans_start = len(ids)
    ids, _lp2, _stop = generate_with_logprobs(policy, tokenizer, ids, max_new,
                                              temperature=temp_ans, top_p=0.9, device=device)
    ans_text = tokenizer.decode(ids[ans_start:])
    if res is not None and res in ans_text: R += 1.0
    elif res is not None: R -= 0.3                  # 调了但没用上 → shaping 轻罚
    R -= 0.05 * len(ids[start:])                    # 长度惩罚
    return ids, start, turn1_end, ans_start, R, ok

def seg_logprobs(policy, ids, start, turn1_end, ans_start, device):
    """只取模型真正生成的段(回合1 + 回合2)的逐 token logp;
    环境注入的 <|env|>结果<|env_end|> 是固定文本,跳过不计"""
    lp = sequence_logprobs(policy, ids, start, temperature=1.0, device=device)
    return torch.cat([lp[:turn1_end - start], lp[ans_start - start:]])

def agentic_step_group(policy, tokenizer, device, prompt_ids, expr, result, max_new, temp_ans, g):
    trajs = [rollout_trajectory(...) for _ in range(g)]
    returns = torch.tensor([R for _, _, _, _, R, _ in trajs], dtype=torch.float)
    baseline = returns.mean()
    policy.train()
    total = 0.0
    for (ids, start, t1e, as_, R, _ok), adv in zip(trajs, returns - baseline):
        seg = seg_logprobs(policy, ids, start, t1e, as_, device)
        loss = -adv * seg.mean()
        loss.backward()
        total += loss.item()
    return total / g, returns.mean().item(), sum(1 for *_, ok in trajs if ok)

argparse:--steps 120、--batch_size 2、--g 6、--lr 1e-6、--max_new 12、--temperature 0.9(回合2)、--warmup_epochs 400、--warmup_target_loss 0.05。

运行:

bash
bash run.sh agentic
# 或:
python posttrain/agentic_rl/agentic_rl_train.py \
    --from_pretrained checkpoints/qwen35_dpo/qwen35_dpo \
    --data_path data/agentic_math.jsonl \
    --output_dir checkpoints/qwen35_agentic \
    --steps 120 --batch_size 2 --g 6 --lr 1e-6 --bf16

10.9 实战总结 ​

真实后训练的共同骨架(每个阶段都是这套):

load_tokenizer + load_model(from_pretrained)      # 真权重出发
    │
    ▼
构造 Dataset(预 tokenize)+ 可选 ref 模型(深拷贝冻结)
    │
    ▼
AdamW(lr, weight_decay) + 断点续训(可选)
    │
    ▼
训练循环: 采样 → 前向/loss → backward → 梯度累积 → clip → step → 日志
    │
    ▼
save_checkpoint(final.pt) + export_hf_model       # 供下一阶段 --from_pretrained

阶段间的数据流:

阶段需要的产出的下一阶段用它
SFT预训练产物miniqwen_sftDPO 的 base
DPOSFT 产物 + 偏好数据qwen35_dpoGRPO/PPO/Thinking/Agentic 的 base
GRPO/PPO/Thinking/AgenticDPO 产物 + 各自数据各阶段 checkpoint—

方法论:先 demo 验证算法(第 9 章),再真实跑(本章);算法一致,规模不同。这符合"从基础开始、逐步加复杂度"的学习路线。


上一章:9. 后训练理论 | 下一章:11. vLLM 推理框架