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、导出。
模型加载:只从真权重出发
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 前缀
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 的核心)
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 与导出
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
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
训练循环:
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 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 --bf1610.4 DPO:偏好数据直接对齐
数据:data.py
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):
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 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 --bf1610.5 GRPO:可验证任务强化学习
数据:data.py
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):
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 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 --bf1610.6 PPO:Actor-Critic 强化学习
数据:复用 GRPO
# 文件头注释:
# 默认数据 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):
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 的 TinyLMforward直接返回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 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 --bf1610.7 Thinking:推理微调
数据:data.py
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|> 开始:
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 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 --bf1610.8 Agentic RL:工具调用的强化学习
数据:data.py
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
两阶段训练:
SFT 热身(
warmup_trajectory):教模型完整的多步轨迹(调工具 → 读 env 结果 → 作答),到 loss≈0.05 就停。只教工具调用不够——模型看到 env 结果后只会重复调用,必须把"作答"也教了。RL(
agentic_step_group):组级 REINFORCE,组内均值当 baseline。
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 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 --bf1610.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_sft | DPO 的 base |
| DPO | SFT 产物 + 偏好数据 | qwen35_dpo | GRPO/PPO/Thinking/Agentic 的 base |
| GRPO/PPO/Thinking/Agentic | DPO 产物 + 各自数据 | 各阶段 checkpoint | — |
方法论:先 demo 验证算法(第 9 章),再真实跑(本章);算法一致,规模不同。这符合"从基础开始、逐步加复杂度"的学习路线。
上一章:9. 后训练理论 | 下一章:11. vLLM 推理框架