9. 后训练理论：SFT 到 RL 的算法原理与教学 demo

概述

预训练（第 7 章）让模型学会了"说人话"，权重适配（第 8 章）给了我们一个可用的语言基座。但基座模型只会续写，不会"按照指令办事"。后训练（Post-training） 就是把这个基座变成"听话、有用、对齐人类偏好"的模型的过程。

本章简要讲解后训练各阶段的算法原理，并用 posttrain/demo/ 下的教学 demo 跑通每个算法——demo 用纯 PyTorch 手写的 TinyLM（约 1M 参数 + 字符级 tokenizer），CPU 秒级可跑，是算法的"流程标本"。

后训练路线图（Qwen3.5 官方流程的简化）:
  基座模型
    │
    ▼
  SFT ──── 让模型学会"按指令回答"（监督微调）
    │
    ▼
  DPO ──── 用"偏好对"直接对齐人类偏好（无需奖励模型）
    │
    ▼
  GRPO / PPO ──── 强化学习，最大化可验证奖励（数学/代码）
    │
    ├── Thinking ──── 教模型先思考再作答（推理时引导）
    └── Agentic RL ──── 教模型调用外部工具（计算器）

9.1 教学 demo 的基础：TinyLM

在讲各阶段算法前，先认识 demo 用的后端。posttrain/common.py 提供了一个极简的字符级语言模型，让算法本身成为主角，不被模型规模淹没。

代码文件

TinyLM: 约 1M 参数, 2 层 Transformer, 字符级 tokenizer
  CharTokenizer: 字符 → id（特殊 token 单独编 id）
  TinyLM:  Embedding → Block×2 → RMSNorm → LM Head（tie）

demo 的所有阶段共用同一套接口：

model(input_ids, labels=None) → {"loss": Tensor, "logits": Tensor, "hidden": Tensor}

CharTokenizer

字符级 tokenizer 把每个字符映射成一个 id，特殊 token（<|im_start|>/<|im_end|>/<|think|> 等）单独编 id：

class CharTokenizer:
    """微型字符级 tokenizer，特殊 token 单独编 id"""
    def __init__(self, corpus=""):
        self.c2i = {ch: i for i, ch in enumerate(sorted(set(_BASE_ALPHABET) | set(corpus)))}
        # 特殊 token 从 base 之后续编
        self.s2i = {tok: len(self.c2i) + i for i, tok in enumerate(SPECIALS)}
        ...
    def encode(self, text): ...   # 按特殊 token 正则切分，其余逐字符编码
    def decode(self, ids): ...    # 逐 id 映射回字符串
    def im_end_id(self): ...      # <|im_end|> 的 id（对应官方 248046）

TinyLM

class TinyLM(nn.Module):
    """约 1M 参数的字符级语言模型"""
    def __init__(self, n_vocab, n_ctx=128, d=64, n_layer=2, n_head=4, d_inter=128, seed=0):
        self.embed = nn.Embedding(n_vocab, d)
        self.blocks = nn.ModuleList([Block(d, n_head, d_inter, n_ctx) for _ in range(n_layer)])
        self.norm = RMSNorm(d)
        self.lm_head = nn.Linear(d, n_vocab, bias=False)
        self.lm_head.weight = self.embed.weight   # tie_word_embeddings，对齐 Qwen3.5

_init_weights() 有个细节：embed 和所有 Linear 权重初始化为 ~N(0, 0.02) 的小标准差。默认的大 std 会让 tied lm_head 给出 ~√d 尺度 logits，softmax 饱和导致初始交叉熵远大于 ln(vocab)，模型学不动。

forward 内部做自回归右移：logits[:, :-1] 预测 labels[:, 1:]，ignore_index=-100 跳过 padding。

chat_format：对齐 Qwen 的聊天模板

demo 用 chat_format 拼接 Qwen 风格的对话：

def chat_format(messages):
    """messages: [(role, content), ...] → Qwen chat 模板字符串"""
    return "".join(f"<|im_start|>{role}\n{content}<|im_end|>\n" for role, content in messages)

9.2 SFT：让模型学会"按指令回答"

原理

预训练是"下一个 token 预测"，模型会续写任何文本。SFT（Supervised Fine-Tuning）喂给模型的是 指令-回复对，并且只在 assistant 回复上算 loss（label masking），让模型学会"看到指令 → 输出该回复"。

	预训练	SFT
Loss 计算	所有 token 都算	只在 assistant 回复上算
数据格式	纯文本拼接	对话格式（system + user + assistant）
学习率	较大（3e-4）	较小（1e-5），避免灾难性遗忘

Label Masking：

token:   <system>  user指令   <assistant>  这是一段回复
label:     -100      -100        -100       ✅ 计算loss

-100 是 PyTorch 交叉熵的忽略标记，该位置不参与 loss 计算。

demo 代码

代码文件

from common import CharTokenizer, TinyLM, load_model, make_sft_batch

# 6 条中文 (prompt, response) 对话作为教学数据
DEMO = [
    ("你是谁？", "我是一个助手。"),
    ("1+1=? ", "2"),
    ...
]

tok = CharTokenizer(corpus="".join(p + r for p, r in DEMO))   # 收集所有用到的字符
model = load_model("tiny", n_vocab=tok.n_vocab, seed=0)       # TinyLM
opt = torch.optim.AdamW(model.parameters(), lr=3e-3, weight_decay=0.1)

for step in range(300):
    input_ids, labels = make_sft_batch(tok, DEMO, n_ctx=64)   # 只监督回复段
    loss = model(input_ids, labels=labels)["loss"]
    opt.zero_grad()
    loss.backward()
    torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
    opt.step()
    ...

make_sft_batch 构造监督数据：

def make_sft_batch(tokenizer, samples, n_ctx):
    """samples: (prompt, response) 列表 → (input_ids, labels)
    prompt 段标 -100，只有 assistant 回复参与 loss"""
    for prompt, response in samples:
        prompt_ids = tokenizer.encode(chat_format([("user", prompt)])) \
                     + tokenizer.encode("<|im_start|>assistant\n")
        resp_ids = tokenizer.encode(response + "<|im_end|>\n")
        full = (prompt_ids + resp_ids)[:n_ctx - 1]           # 留一位给 shift
        labels = [-100] * len(prompt_ids) + full[len(prompt_ids):]
        ...

训练后对每个 prompt 自回归生成回复，验证模型学会了对话格式。

真实 SFT 还支持多轮对话（每一轮 assistant 都参与 loss）和 --mask_history（只训最后一轮）。

9.3 DPO：用偏好对直接对齐

原理

SFT 之后，模型学会了格式，但不知道"什么样的回答更好"。传统做法是训练一个奖励模型，再用 RL（PPO）优化。DPO（Direct Preference Optimization） 绕开了奖励模型——直接利用偏好对 (chosen, rejected) 更新策略。

DPO 的核心洞察：最优策略可以解析地写成"参考策略 + 隐式奖励"的形式，因此不需要显式奖励模型，只需让模型更喜欢 chosen、更不喜欢 rejected：

$${\mathcal{L}}_{DPO}=-\log \sigma (\beta \cdot [(\log {\pi }_{\theta }(y_w)-\log {\pi }_{ref}(y_w))-(\log {\pi }_{\theta }(y_l)-\log {\pi }_{ref}(y_l))])$$

其中：

• ${\pi }_{\theta }$：当前策略（带梯度）
• ${\pi }_{ref}$：参考模型（训练前深拷贝，冻结）
• $y_w$：chosen（更优回复），$y_l$：rejected（较差回复）
• $\beta$：温度系数，控制对齐强度（越大越激进）
• $\sigma$：sigmoid

直觉：让"chosen 相对参考模型的优势"远大于"rejected 相对参考模型的优势"。当 chosen 明显优于 rejected 时，括号内为正，-logσ(正数) 变小 → 梯度朝这个方向更新。

demo 代码

代码文件

def dpo_loss(policy, ref, chosen_ids, rejected_ids, beta=0.1):
    """DPO loss：chosen 相对参考的优势 - rejected 相对参考的优势，过 sigmoid"""
    logp_c_pol  = seq_logprob(policy, None, chosen_ids)          # 策略，带梯度
    logp_c_ref  = seq_logprob(ref, None, chosen_ids)             # 参考，no_grad
    logp_r_pol  = seq_logprob(policy, None, rejected_ids)
    logp_r_ref  = seq_logprob(ref, None, rejected_ids)
    r_w = logp_c_pol[1] - logp_c_ref[1]     # chosen 的相对优势
    r_l = logp_r_pol[1] - logp_r_ref[1]     # rejected 的相对优势
    margin = r_w - r_l                       # σ 括号里"chosen 优于 rejected"的对数比
    loss = -F.logsigmoid(beta * margin).mean()
    return loss, margin.detach()

主流程：建 policy，ref = copy.deepcopy(policy) 并冻结；AdamW(lr=3e-4)；每步构造 3 组 (prompt, chosen, rejected) 的 batch，算 DPO loss 更新。

训练后打印每条样本的 r_w / r_l / margin，直观看到"DPO 视角下相对参考模型的对数比"。

真实版用 batch_sequence_logprobs 批量算 logp，并只对回复段 [start, end) 取均值，避免长回复被不公平均压低。

9.4 GRPO：无批评家的强化学习

原理

GRPO（Group Relative Policy Optimization）是 DeepSeek-R1 使用的 RL 算法。它适用于奖励可验证的任务（数学答案、代码能否编译等）——不需要训练奖励模型，直接用规则判断对错。

核心思想：对同一个 prompt 采样 G 条回复，用组内相对优劣作为优势，省掉了 PPO 的 critic（价值网络）：

$$A_i=\frac{r_i-\text{mean}(r_1,...,r_G)}{\text{std}(r_1,...,r_G)+ϵ}$$

其中 $r_i$ 是规则奖励（如"答案出现给 1 分"）。组内归一化让优势零均值、有尺度，模型不会被整体压崩。

更新用 clip 比率 loss + KL 惩罚（和 PPO 策略项类似）：

$${\mathcal{L}}_{GRPO}=-\text{mean}[min({\rho }_{t}A,\text{clip}({\rho }_t,1-ϵ,1+ϵ)A)]+\beta \cdot \text{KL}({\pi }_{\theta }\parallel {\pi }_{ref})$$

其中 ${\rho }_t=\exp (\log {\pi }_{\theta }(y_t)-\log {\pi }_{old}(y_t))$ 是新旧策略概率比，$\text{clip}$ 防止更新过猛，KL 防止偏离参考模型太远。

demo 代码

代码文件

def rule_based_reward(text, keyword):
    """可验证奖励：回复里出现关键词给 1 分，否则 0"""
    return 1.0 if keyword in text else 0.0

def grpo_step(policy, ref, tok, prompt_ids, keyword, g=4, max_new=12, temperature=1.0,
              top_p=0.9, beta=0.1, eps=0.2):
    # 1. rollout：同一 prompt 采样 g 条回复，缓存旧策略逐 token logp
    samples = rollout(policy, tok, prompt_ids, g=g, max_new=max_new, temperature=temperature, top_p=top_p)
    prompt_len = len(prompt_ids)

    # 2. 规则奖励打分
    rewards = [rule_based_reward(tok.decode(ids[prompt_len:]).split(TOK_IM_END)[0], keyword)
               for ids, _old_lp, _tot in samples]
    rewards = torch.tensor(rewards, dtype=torch.float)

    # 3. 组内相对优势
    adv = (rewards - rewards.mean()) / (rewards.std() + 1e-4)

    # 4. clip 比率 loss + KL
    total = 0.0
    for (ids, old_logps, _tot), a in zip(samples, adv):
        cur = response_logprobs(policy, ids, prompt_len)      # 当前策略 logp（带梯度）
        old = torch.tensor(old_logps)                          # 旧策略 logp（无梯度）
        ratio = (cur - old).exp()                              # ρ = πθ/πold
        surr = torch.min(ratio * a, torch.clamp(ratio, 1 - eps, 1 + eps) * a)
        with torch.no_grad():
            ref_lp = response_logprobs(ref, ids, prompt_len)  # 参考模型 logp
        kl = (cur - ref_lp).mean()                             # 近似 KL(πθ‖πref)
        loss = -surr.mean() + beta * kl
        loss.backward()
        total += loss.item()
    return total / len(samples), rewards.tolist()

主流程：建 policy + 冻结 ref 深拷贝；AdamW(lr=1e-3)；每步对 3 道选择题每个 prompt 调 grpo_step 累积梯度后 opt.step()；定期打印 loss 和组内命中率。

9.5 PPO：经典 Actor-Critic RL

原理

PPO（Proximal Policy Optimization）是通用 RL 算法，适合奖励不可直接验证的场景（需要人类反馈、风格对齐等）。它比 GRPO 多一个 critic（价值网络），用学到的价值估计替代组内基线：

$$A_t=R-V(s_t)$$

三个损失相加：

1. 策略 loss（clip 比率）：同 GRPO 的 surr，L_pol = -mean(min(ρ·A, clip(ρ)·A))
2. 价值 loss：L_val = MSE(V(s_t), R)，让 critic 学会预测回报
3. 熵奖励：L_ent = -mean(entropy)，鼓励探索，避免过早收敛

$${\mathcal{L}}_{PPO}=L_{pol}+0.5\cdot L_{val}-\text{ent\_coef}\cdot L_{ent}$$

demo 代码

代码文件

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, opt, tok, prompt_ids, keyword, max_new=12,
             temperature=1.0, clip_eps=0.2, ent_coef=0.01):
    prompt_len = len(prompt_ids)
    # 1. rollout 一条回复（g=1），缓存旧策略 logp
    (ids, old_logps, _tot), = rollout(policy, tok, prompt_ids, g=1, max_new=max_new,
                                      temperature=temperature, top_p=0.9)
    # 2. 奖励（末尾给分）
    R = 1.0 if keyword in tok.decode(ids[prompt_len:]).split(TOK_IM_END)[0] else 0.0
    R_t = torch.tensor(float(R))

    # 3. 一次前向拿 logits + hidden
    out = policy(torch.tensor([ids]).long())
    logp = F.log_softmax(out["logits"][0], dim=-1)
    cur = logp[prompt_len - 1:-1].gather(-1, torch.tensor(ids[prompt_len:]).unsqueeze(-1)).squeeze(-1)
    ratio = (cur - torch.tensor(old_logps)).exp()

    # 4. 价值预测 + 优势 A = R - V
    v_t = value(out["hidden"][0, prompt_len - 1:-1])      # 每个 response 位置的状态
    adv = R_t - 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, R_t.expand_as(v_t))
    ent = -(logp[prompt_len - 1:-1].exp() * logp[prompt_len - 1:-1]).sum(-1).mean()
    loss = pol_loss + 0.5 * val_loss - ent_coef * ent
    opt.zero_grad()
    loss.backward()
    torch.nn.utils.clip_grad_norm_(list(policy.parameters()) + list(value.parameters()), 1.0)
    opt.step()
    return loss.item(), R

主流程：建 policy + ValueHead（critic 和 LM 分开优化）；AdamW(policy.parameters() + value.parameters(), lr=1e-3)；每步对每个 prompt 调 ppo_step。

PPO vs GRPO 怎么选：PPO 需要训练 critic，适合"奖励无法自动打分"（人类反馈、风格）；GRPO 用组内基线省掉 critic，适合"答案可验证"（数学、代码）。DeepSeek-R1 / Qwen3.5 走 GRPO。

9.6 Thinking：教模型先思考再作答

原理

Thinking（推理微调 / 推理 SFT）让模型学会在回答前先输出一段思考过程（CoT），再给最终答案。关键认知：它不是一套独立的权重，而是同一权重在训练数据里混入"带思考段"的样本，推理时用不同的引导起点来激发。

训练数据混合两类:
  THINK  样本:  user → <|im_start|>think\n 思考过程 <|im_end|> → <|im_start|>assistant\n 答案
  DIRECT 样本:  user → <|im_start|>assistant\n 答案        （无思考段，普通对话）

推理时引导:
  thinking 模式:      user 之后接 <|im_start|>think\n   ← 先思考再答
  non-thinking 模式:  user 之后接 <|im_start|>assistant\n  ← 直接答

loss 监督起点 = 第二个 <|im_start|>（think 样本从 think 头起算，direct 样本从 assistant 头起算），前面的 system/user 上下文不监督。

demo 代码

代码文件

def format_reasoning(prompt, coT, answer):
    """thinking 样本：user + think 段 + assistant 答案"""
    return (chat_format([("user", prompt)]) +
            f"<|im_start|>think\n{coT}<|im_end|>\n"
            f"<|im_start|>assistant\n{answer}<|im_end|>\n")

def format_direct(prompt, answer):
    """non-thinking 样本：无 think 段"""
    return (chat_format([("user", prompt)]) +
            f"<|im_start|>assistant\n{answer}<|im_end|>\n")

def build_batch(tok, samples, n_ctx):
    """混合 think + direct 样本，labels 从第二个 <|im_start|> 起监督"""
    ...
    second = _second_im_start(ids)   # 第二个 <|im_start|> 位置（think/assistant 头）
    labels = [-100] * second + ids[second:]
    ...

训练时同时喂 DEMO（think 样本）和 DEMO_DIRECT（direct 样本）；训练后演示两种生成模式：同一权重，thinking 接 <|im_start|>think\n，non-thinking 接 <|im_start|>assistant\n。

9.7 Agentic RL：教模型调用工具

原理

Agentic RL 让模型学会调用外部工具（这里是一个玩具 calc 计算器环境），先调工具拿结果，再作答。它是 REINFORCE with group baseline（GRPO 的思想雏形），核心是多步轨迹：

回合1: 生成工具调用  <|tool|>calc(expr)<|tool_end|>      （低温度 0.6，格式要准）
        ↓
环境:  执行并注入   <|env|>result<|env_end|>             （固定文本，模型不生成）
        ↓
回合2: 看到结果作答  <|im_start|>assistant ...答案        （高温度，探索）

奖励设计：

行为	奖励
工具调用格式正确	+1
模型自己答的部分包含环境结果	+1
没调工具（防"闭嘴"坍缩）	-1
调了但答案没用上（shaping）	-0.3
每生成一个 token	-0.05（长度惩罚）

优势用组内均值做基线：$A_i=R_i-\text{mean}(R_1..R_G)$（零均值，不会整体压崩）。

关键前提：RL 前必须先 SFT 热身。随机初始化的模型发不出结构化工具调用，永远拿不到工具奖励 → 所有轨迹负回报 → REINFORCE 均匀压低所有 token → 坍缩成"闭嘴"。

demo 代码

代码文件

def eval_expr(expr):
    """环境：AST 白名单求值（只放行四则运算），非法返回 '?'"""
    ...

def simulate_env(text):
    """扫描文本里的工具调用，返回 (是否调用成功, 环境结果)"""
    m = re.search(r'<\|tool\|>calc\(([^)]*)\)<\|tool_end\|>', text)
    return (True, eval_expr(m.group(1))) if m else (False, None)

def rollout_trajectory(policy, tok, prompt_ids, expr, result, max_new, temp_ans):
    start = len(prompt_ids)
    ids = list(prompt_ids)
    # 回合1：生成工具调用
    stop = {tok.s2i[TOK_TOOL_END], tok.im_end_id()}
    for _ in range(max_new):
        t, _ = _next_token(policy, ids, 0.6, 0.9)
        ids.append(t)
        if t in stop: break
    # 环境执行
    ok, res = simulate_env(tok.decode(ids[start:]))
    R = -1.0 if not ok else 1.0                     # 没调工具大惩罚
    if ok:
        ids += tok.encode(f"{TOK_ENV}{res}{TOK_ENV_END}\n")   # 注入环境结果
    # 回合2：看到结果作答
    ans_start = len(ids)
    for _ in range(max_new):
        t, _ = _next_token(policy, ids, temp_ans, 0.9)
        ids.append(t)
        if t == tok.im_end_id(): break
    ans_text = tok.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
    R -= 0.05 * len(ids[start:])                    # 长度惩罚
    return ids, start, R, ok

def agentic_step_group(policy, tok, prompt_ids, expr, result, max_new, temp_ans, g=6):
    """对一个 prompt 做一次组级 REINFORCE 更新（组内均值当 baseline）"""
    trajs = [rollout_trajectory(policy, tok, prompt_ids, expr, result, max_new, temp_ans)
             for _ in range(g)]
    returns = torch.tensor([R for _, _, R, _ in trajs], dtype=torch.float)
    baseline = returns.mean()
    total = 0.0
    for (ids, start, R, _ok), adv in zip(trajs, returns - baseline):
        lp = response_logprobs(policy, ids, start)  # 当前策略重算轨迹 logπ（带梯度）
        loss = -adv * lp.mean()
        loss.backward()
        total += loss.item()
    return total / g, returns.mean().item(), sum(1 for _, _, _, ok in trajs if ok)

主流程：建 TinyLM + AdamW(lr=1e-3)；先 warmup_trajectory 热身（SFT 教完整多步轨迹，到 loss≈0.05 停）；再 RL 循环每步对每个 prompt 调 agentic_step_group。

9.8 小总结

阶段	一句话	关键机制	demo 文件
SFT	让模型按指令回答	label masking（只训 assistant 段）	demo/sft_demo.py
DPO	用偏好对直接对齐	-logσ(β·(chosen优势 - rejected优势))	demo/dpo_demo.py
GRPO	可验证任务的 RL	组内归一化优势 + clip 比率 + KL	demo/grpo_demo.py
PPO	通用 Actor-Critic RL	critic 学价值 + clip 策略 + 熵	demo/ppo_demo.py
Thinking	先思考再作答	think 段 + 双模式引导	demo/thinking_demo.py
Agentic RL	调用外部工具	多步轨迹 + 组内 baseline + SFT 热身	demo/agentic_rl_demo.py

算法演进关系：

• SFT 是基础（所有后训练的前提）。
• DPO 是对齐偏好的轻量替代（不用奖励模型）。
• GRPO 和 PPO 都是 RL；GRPO 是 PPO 的"无 critic"变体，适合可验证奖励。
• Thinking 和 Agentic RL 是"任务形态"的扩展（思考/工具），底层算法仍回到 SFT 或 RL。

下一个问题：demo 用 TinyLM 验证了算法，但真正要得到可用的模型，需要用真实 Qwen3.5 权重、真实数据集、真实训练循环来跑。这就是下一章的内容——后训练实战。

---

上一章：8. 权重适配 | 下一章：10. 后训练实战