| # 面试：训练系统深度 |
|
|
| > 本仓库 `src/trainers/` 的训练逻辑，覆盖 pretrain、SFT、DPO、PPO、GRPO、蒸馏 |
|
|
| ## 0. 训练流程概览 |
|
|
| ``` |
| 数据集 (dataset/) |
| │ |
| ▼ |
| SkipBatchSampler → DataLoader → 批次拼接 |
| │ |
| ▼ |
| Trainer (trainers/) |
| ├── 模型初始化（支持 checkpoint 恢复） |
| ├── 混合精度训练（bfloat16/float16） |
| ├── 梯度累积 + 梯度裁剪 |
| ├── 学习率调度（cosine） |
| └── 定期保存 checkpoint |
| ``` |
|
|
| --- |
|
|
| ## Q1. 损失里 label 为什么要平移？ |
|
|
| ### 标准 Next-Token Prediction |
|
|
| ```python |
| # src/models/lm/model.py:69-70 |
| shift_logits = logits[..., :-1, :].contiguous() |
| shift_labels = labels[..., 1:].contiguous() |
| loss = F.cross_entropy(shift_logits.view(-1, shift_logits.size(-1)), shift_labels.view(-1)) |
| ``` |
|
|
| ### 为什么平移？ |
|
|
| - `logits[:, :-1]` 预测的是 `labels[:, 1:]` 的内容 |
| - 标准 next-token 预测：给定前缀，预测下一个 token |
| - 不平移的话，模型会学到"预测自己"，没有意义 |
|
|
| > 面试点：如果 label 不平移会怎样？→ 模型会学到恒等映射，训练 loss 很低但生成质量很差 |
|
|
| --- |
|
|
| ## Q2. SFT 为什么只标 assistant 段？ |
|
|
| ### Loss Mask 机制 |
|
|
| ```python |
| # src/dataset/sft.py |
| def generate_labels(self, input_ids): |
| labels = input_ids.clone() |
| # prompt/system 段 label 置 -100 |
| labels[:prompt_end_pos] = -100 |
| # 只对 assistant 回复计算损失 |
| return labels |
| ``` |
|
|
| ### 为什么只标 assistant？ |
|
|
| 1. **避免学用户说话**：模型应该学习生成回复，不拟合用户输入 |
| 2. **提高效率**：只计算有意义部分的损失 |
| 3. **符合实际使用**：推理时只生成 assistant 回复 |
|
|
| > 面试点：如果标全部 token 会怎样？→ 模型会学用户输入的模式，生成时可能重复用户说话风格 |
|
|
| --- |
|
|
| ## Q3. 配置系统优先级？ |
|
|
| ### 三级优先级 |
|
|
| ``` |
| CLI 参数 > YAML 默认值 > 代码默认 |
| ``` |
|
|
| ### 实现（`src/utils/training.py:25-38`） |
|
|
| ```python |
| def apply_config(parser, default_config=None): |
| # 1. 先加载 YAML 默认值 |
| if default_config: |
| with open(default_config) as f: |
| config = yaml.safe_load(f) |
| # 2. 扁平化注入 argparse 默认值 |
| for key, value in config.items(): |
| if key in [a.dest for a in parser._actions]: |
| parser.set_defaults(**{key: value}) |
| # 3. CLI 参数覆盖 |
| return parser.parse_args() |
| ``` |
|
|
| ### 为什么这样设计？ |
|
|
| 1. **灵活性**：可以用 YAML 配置常用参数，CLI 临时覆盖 |
| 2. **可复现**：YAML 文件可以版本控制 |
| 3. **向后兼容**：代码默认值保证基本功能 |
|
|
| --- |
|
|
| ## Q4. 续训怎么保证精确？ |
|
|
| ### SkipBatchSampler（`src/utils/training.py:177-200`） |
|
|
| ```python |
| class SkipBatchSampler: |
| def __init__(self, dataset, batch_size, step): |
| self.step = step # 已完成的 step 数 |
| |
| def __iter__(self): |
| # 跳过前 step 个 batch |
| indices = list(range(len(self.dataset))) |
| indices = indices[self.step * self.batch_size:] |
| ... |
| ``` |
|
|
| ### Checkpoint 恢复 |
|
|
| ```python |
| # src/utils/training.py:105-158 |
| def lm_checkpoint(model, optimizer, scheduler, step, path): |
| # 原子保存 |
| tmp_path = path + ".tmp" |
| torch.save({ |
| 'step': step, |
| 'model_state_dict': model.state_dict(), |
| 'optimizer_state_dict': optimizer.state_dict(), |
| 'scheduler_state_dict': scheduler.state_dict(), |
| }, tmp_path) |
| os.replace(tmp_path, path) |
| ``` |
|
|
| ### 世界规模自适应（`src/utils/training.py:153-156`） |
|
|
| ```python |
| # GPU 数量变化时自动调整 step |
| if world_size != saved_world_size: |
| step = int(step * saved_world_size / world_size) |
| ``` |
|
|
| > 面试点：为什么需要世界规模自适应？→ 多机训练时 GPU 数量可能变化，需要按比例调整 step 保证训练进度一致 |
|
|
| --- |
|
|
| ## Q5. DPO Loss 实现 |
|
|
| ### 标准 DPO Loss（`src/trainers/lm/dpo.py:32-48`） |
|
|
| ```python |
| def dpo_loss(policy_logratios, reference_logratios, beta=0.15): |
| # DPO Loss = -logsigmoid(β * (π_logratios - ref_logratios)) |
| loss = -F.logsigmoid(beta * (policy_logratios - reference_logratios)) |
| return loss.mean() |
| ``` |
|
|
| ### 关键参数 |
|
|
| - `beta=0.15`：控制策略偏离参考模型的程度 |
| - `lr=4e-8`：极小学习率防止遗忘 |
|
|
| ### 参考模型（`src/trainers/lm/dpo.py:187-189`） |
|
|
| ```python |
| # 冻结的 ref_model |
| self.ref_model = init_model(args, config) |
| for param in self.ref_model.parameters(): |
| param.requires_grad = False |
| ``` |
|
|
| > 面试点：为什么 DPO 学习率这么小？→ DPO 直接优化策略，太大的学习率会导致策略偏离参考模型太远，生成质量下降 |
|
|
| --- |
|
|
| ## Q6. GRPO 实现（`src/trainers/lm/grpo.py:119-142`） |
|
|
| ### 核心思想 |
|
|
| Group Relative Policy Optimization：每个 prompt 生成多个候选，组内标准化优势。 |
|
|
| ### 实现细节 |
|
|
| ```python |
| def grpo_loss(self, logprobs, old_logprobs, advantages, clip_epsilon=0.2): |
| # 每个 prompt 生成 num_generations=6 个候选 |
| # 优势估计：组内标准化 |
| advantages = (reward - mean) / (std + 1e-4) |
| |
| # PPO Clip |
| ratio = torch.exp(logprobs - old_logprobs) |
| clipped = torch.clamp(ratio, 1 - clip_epsilon, 1 + clip_epsilon) |
| loss = -torch.min(ratio * advantages, clipped * advantages) |
| |
| # KL 惩罚 |
| kl = logprobs - old_logprobs |
| kl_penalty = torch.exp(kl) - kl - 1 |
| |
| return loss.mean() + self.kl_coef * kl_penalty.mean() |
| ``` |
|
|
| ### 两种 Loss 模式 |
|
|
| 1. **"cispo"**：高端 clamped ratio（`epsilon_high=5.0`） |
| 2. **"grpo"**：标准 PPO clip（`epsilon=0.2`） |
|
|
| ### 奖励函数（`src/trainers/lm/grpo.py:35-66`） |
|
|
| ```python |
| def reward_fn(self, text): |
| reward = 0 |
| |
| # 长度奖励：20-800 字符 +0.5 |
| if 20 < len(text) < 800: |
| reward += 0.5 |
| |
| # thinking 奖励：20-300 字符 +1.0 |
| if '<think>' in text and 20 < think_len < 300: |
| reward += 1.0 |
| |
| # thinking 次数奖励：恰好 1 次 +0.25 |
| if text.count('<think>') == 1: |
| reward += 0.25 |
| |
| # 重复惩罚：3-gram 重复度 |
| reward -= repetition_penalty(text) |
| |
| # Reward Model 分数 |
| rm_score = self.reward_model(text) |
| reward += rm_score |
| |
| return reward |
| ``` |
|
|
| > 面试点：GRPO 和 PPO 的区别？→ GRPO 不需要 critic model，直接用组内标准化估计优势，更简单高效 |
|
|
| --- |
|
|
| ## Q7. PPO 实现（`src/trainers/lm/ppo.py`） |
|
|
| ### CriticModel（`src/trainers/lm/ppo.py:35-47`） |
|
|
| ```python |
| class CriticModel(LMForCausalLM): |
| def __init__(self, config): |
| super().__init__(config) |
| # lm_head 替换为 value_head (hidden_size -> 1) |
| self.value_head = nn.Linear(config.hidden_size, 1, bias=False) |
| del self.lm_head |
| ``` |
|
|
| ### GAE（`src/trainers/lm/ppo.py:138-145`） |
|
|
| ```python |
| def compute_gae(rewards, values, gamma=1.0, lam=0.95): |
| advantages = [] |
| gae = 0 |
| for t in reversed(range(len(rewards))): |
| delta = rewards[t] + gamma * values[t + 1] - values[t] |
| gae = delta + gamma * lam * gae |
| advantages.insert(0, gae) |
| return advantages |
| ``` |
|
|
| ### 早停机制（`src/trainers/lm/ppo.py:181-188`） |
|
|
| ```python |
| def ppo_update(self, ...): |
| approx_kl = (logprobs - old_logprobs).mean() |
| if approx_kl > 0.25: |
| # 早停，但保持 DDP 通信闭环 |
| loss = loss * 0.0 |
| return loss |
| ``` |
|
|
| > 面试点：为什么早停时要 `loss * 0.0`？→ DDP 要求所有 rank 都参与前向/反向传播，`loss * 0.0` 保持计算图连通，避免死锁 |
|
|
| --- |
|
|
| ## Q8. 蒸馏实现（`src/trainers/lm/distillation.py`） |
|
|
| ### KL 散度蒸馏（`src/trainers/lm/distillation.py:23-34`） |
|
|
| ```python |
| def kl_divergence(teacher_logits, student_logits, temperature=1.5): |
| # KL(teacher || student) * temperature^2 |
| teacher_probs = F.softmax(teacher_logits / temperature, dim=-1) |
| student_log_probs = F.log_softmax(student_logits / temperature, dim=-1) |
| kl = F.kl_div(student_log_probs, teacher_probs, reduction='batchmean') |
| return kl * temperature ** 2 |
| ``` |
|
|
| ### 混合损失（`src/trainers/lm/distillation.py:91`） |
|
|
| ```python |
| loss = alpha * ce_loss + (1 - alpha) * kl_loss |
| # 默认 alpha=0.5, temperature=1.5 |
| ``` |
|
|
| ### Teacher/Student 独立配置 |
|
|
| ```python |
| # Teacher 和 Student 可以有不同的配置 |
| teacher_config = LMConfig(hidden_size=768, num_hidden_layers=8) |
| student_config = LMConfig(hidden_size=384, num_hidden_layers=4) |
| ``` |
|
|
| > 面试点：为什么蒸馏要乘以 temperature²？→ 保持梯度尺度一致，避免温度变化影响学习率 |
|
|
| --- |
|
|
| ## Q9. Rollout Engine 设计（`src/trainers/lm/rollout_engine.py`） |
| |
| ### 策略模式 |
| |
| ```python |
| class RolloutEngine(ABC): |
| @abstractmethod |
| def generate(self, prompts, **kwargs): |
| pass |
| |
| class TorchRolloutEngine(RolloutEngine): |
| def generate(self, prompts, **kwargs): |
| # 原生 PyTorch 推理 |
| ... |
| |
| class SGLangRolloutEngine(RolloutEngine): |
| def generate(self, prompts, **kwargs): |
| # 通过 HTTP API 调用 SGLang 服务 |
| ... |
| ``` |
| |
| ### 权重同步（`src/trainers/lm/rollout_engine.py:165-188`） |
| |
| ```python |
| def sync_weights(self, model): |
| # 仅 rank 0 执行保存 |
| if self.rank == 0: |
| # 保存到磁盘 |
| torch.save(model.state_dict(), 'tmp_model.pt') |
| # HTTP 请求 SGLang 热加载 |
| requests.post('http://localhost:8000/load_model', ...) |
| # 广播成功标志 |
| dist.broadcast(success_flag, src=0) |
| ``` |
| |
| > 面试点：为什么要权重同步？→ RL 训练需要最新的策略生成 rollout，必须确保推理服务使用最新权重 |
| |
| --- |
| |
| ## Q10. 混合精度训练 |
| |
| ### 本仓库的混合精度策略 |
| |
| ```python |
| # src/trainers/lm/pretrain.py |
| scaler = GradScaler(enabled=(args.dtype == 'float16')) |
| with autocast(device_type='cuda', dtype=dtype): |
| logits, _, aux_loss = model(input_ids, labels=labels) |
| loss = criterion(logits, labels) + aux_loss |
| |
| scaler.scale(loss).backward() |
| scaler.unscale_(optimizer) |
| torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) |
| scaler.step(optimizer) |
| scaler.update() |
| ``` |
| |
| ### 为什么用 bfloat16 而不是 float16？ |
| |
| - **bfloat16**：8 位指数，范围大，不易溢出 |
| - **float16**：5 位指数，范围小，容易溢出 |
| |
| > 面试点：GradScaler 的作用？→ 防止 float16 梯度下溢，动态调整 loss 缩放因子 |
| |
| --- |
| |
| ## Q11. 梯度累积 |
| |
| ### 为什么需要梯度累积？ |
| |
| 显存有限时，可以用小 batch 大 accumulation 模拟大 batch。 |
| |
| ### 实现（`src/trainers/lm/pretrain.py`） |
| |
| ```python |
| accumulation_steps = 8 # 默认 |
| for i, batch in enumerate(dataloader): |
| with autocast(device_type='cuda', dtype=dtype): |
| logits, _, aux_loss = model(batch) |
| loss = criterion(logits, labels) + aux_loss |
| loss = loss / accumulation_steps # 梯度累积需要除以步数 |
| |
| scaler.scale(loss).backward() |
| |
| if (i + 1) % accumulation_steps == 0: |
| scaler.unscale_(optimizer) |
| torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) |
| scaler.step(optimizer) |
| scaler.update() |
| optimizer.zero_grad() |
| ``` |
| |
| > 面试点：为什么 loss 要除以 accumulation_steps？→ 保持梯度尺度一致，避免累积步数影响学习率 |
| |
| --- |
| |
| ## Q12. 学习率调度（`src/utils/training.py:82-83`） |
| |
| ### Cosine 调度 |
| |
| ```python |
| def get_lr(step, lr, total): |
| # 最低衰减到 10% |
| return lr * (0.1 + 0.45 * (1 + math.cos(math.pi * step / total))) |
| ``` |
| |
| ### 为什么用 Cosine？ |
| |
| 1. **平滑衰减**：避免学习率骤降 |
| 2. **实验效果好**：比 step decay 和 linear decay 更稳定 |
| 3. **标准做法**：几乎所有大模型都用 cosine |
| |
| --- |
| |
| ## Q13. Checkpoint 原子写入（`src/utils/checkpoint.py:7-10`） |
| |
| ### 问题 |
| |
| 训练过程中保存 checkpoint 时，如果中途崩溃，可能导致 checkpoint 损坏。 |
| |
| ### 解决方案 |
| |
| ```python |
| def save_checkpoint(model, optimizer, scheduler, path): |
| # 先保存到临时文件 |
| tmp_path = path + ".tmp" |
| torch.save({...}, tmp_path) |
| # 原子替换 |
| os.replace(tmp_path, path) |
| ``` |
| |
| ### 为什么用 os.replace？ |
| |
| `os.replace` 是原子操作，要么完成要么不发生，不会出现部分写入的情况。 |
| |
| --- |
| |
| ## Q14. DDP 初始化（`src/utils/distributed.py`） |
| |
| ### 标准 DDP 初始化 |
| |
| ```python |
| def init_distributed(): |
| dist.init_process_group(backend='nccl') |
| local_rank = int(os.environ['LOCAL_RANK']) |
| torch.cuda.set_device(local_rank) |
| return local_rank |
| ``` |
| |
| ### DDP 死锁防护 |
| |
| ```python |
| # PPO 早停时保持通信闭环 |
| if early_stop: |
| loss = loss * 0.0 # 保持计算图连通 |
| ``` |
| |
| > 面试点：为什么 DDP 会死锁？→ 如果某个 rank 不参与前向/反向传播，其他 rank 会等待它，导致死锁 |
| |
| --- |
| |
| ## Q15. 世界规模自适应（`src/utils/training.py:153-156`） |
| |
| ### 问题 |
| |
| 多机训练时 GPU 数量可能变化，需要按比例调整 step。 |
| |
| ### 实现 |
| |
| ```python |
| def lm_checkpoint(model, optimizer, scheduler, step, path): |
| # 保存世界规模 |
| torch.save({ |
| 'step': step, |
| 'world_size': dist.get_world_size(), |
| }, path) |
| |
| # 恢复时检查世界规模 |
| saved_world_size = checkpoint['world_size'] |
| if world_size != saved_world_size: |
| step = int(step * saved_world_size / world_size) |
| ``` |
| |
| > 面试点：为什么需要这个？→ 保证训练进度一致，避免某些 rank 重复训练或跳过训练 |
| |
