核心

预训练从零实现 — 课程A W5

对应课程: course_A W5 (预训练实践)

难度: ★★★ (L3 进阶)

知识基础: Build LLM from Scratch Ch5 — 在未标注数据上预训练

> 训练数据的来源、版权、去重 — 这是合规审查的核心环节

0. 复用之前的代码

从 07_GPT_Architecture.ipynb 加载 GPT 模型结构。

代码

import torch
import torch.nn as nn
import tiktoken
from torch.utils.data import Dataset, DataLoader

# 复制 07 中的核心类 (此处省略, 实际使用时会从完整代码文件中导入)
# 这里用 nn.LayerNorm 代替自定义实现以便演示

class GELU(nn.Module):
    def forward(self, x):
        return 0.5 * x * (1 + torch.tanh(
            torch.sqrt(torch.tensor(2.0 / torch.pi)) * (x + 0.044715 * x**3)
        ))

class FeedForward(nn.Module):
    def __init__(self, emb_dim):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(emb_dim, 4 * emb_dim), GELU(), nn.Linear(4 * emb_dim, emb_dim))
    def forward(self, x): return self.layers(x)

print("✓ 基本组件加载完成")
print(f"PyTorch: {torch.__version__}, MPS: {torch.backends.mps.is_available()}")

1. 理解交叉熵损失

LLM 训练 = 下一个 token 预测 = 多分类问题。

对于每个位置: 模型输出一个 vocab_size 维的向量 → softmax → 取最大概率的 token

代码

# 交叉熵损失 可视化
vocab_size = 10
logits = torch.tensor([[2.0, 1.0, 0.1, -1.0, -2.0, 0.5, 0.3, -0.5, 0.8, 1.5]])
target = torch.tensor([0])  # 正确答案是 token 0

probs = torch.softmax(logits, dim=-1)
loss = nn.functional.cross_entropy(logits, target)

print(f"logits: {logits[0].tolist()}")
print(f"softmax 概率: {[f'{p:.3f}' for p in probs[0].tolist()]}")
print(f"真实 token: {target.item()}")
print(f"交叉熵损失: {loss.item():.4f}")
print(f"困惑度 (perplexity): {torch.exp(loss).item():.2f} (模型对下一个token的'惊讶程度')")

Perplexity (困惑度): 模型对下一个 token 的"惊讶程度"。

PPL=1: 完美预测
PPL=50: 模型等效于随机在 50 个 token 中猜
GPT-3 最终 PPL 约 10-20

2. 准备训练数据

用小数据集演示: "The Verdict" 短篇小说 (~5000 tokens)

代码

# 创建一个超小训练语料 (用 tiktoken 编码)
enc = tiktoken.get_encoding("gpt2")
train_text = "The quick brown fox jumps over the lazy dog. " * 50
train_text += "Machine learning is transforming the world of technology. " * 50
train_text += "Python is the most popular language for data science and AI. " * 50

token_ids = enc.encode(train_text)
print(f"训练数据: {len(train_text)} 字符, {len(token_ids)} tokens")
print(f"前 20 个 token ID: {token_ids[:20]}")

代码

class GPTDataset(Dataset):
    """滑动窗口数据集"""
    def __init__(self, token_ids, max_length, stride):
        self.inputs = []
        self.targets = []
        for i in range(0, len(token_ids) - max_length, stride):
            self.inputs.append(token_ids[i:i+max_length])
            self.targets.append(token_ids[i+1:i+max_length+1])
    
    def __len__(self):
        return len(self.inputs)
    
    def __getitem__(self, idx):
        return torch.tensor(self.inputs[idx]), torch.tensor(self.targets[idx])

max_length = 64
stride = 32
dataset = GPTDataset(token_ids, max_length, stride)
loader = DataLoader(dataset, batch_size=4, shuffle=True)
print(f"数据集大小: {len(dataset)} 个样本")
x, y = next(iter(loader))
print(f"批次形状: input {x.shape}, target {y.shape}")

3. 计算训练损失

对 LLM 来说: 一个批次的总损失 = 所有 token 位置的平均交叉熵损失

代码

def calc_loss_batch(logits, targets):
    """计算一个批次的损失"""
    # logits: (batch, seq_len, vocab_size)
    # targets: (batch, seq_len)
    logits = logits.view(-1, logits.size(-1))
    targets = targets.view(-1)
    return nn.functional.cross_entropy(logits, targets)

def calc_loss_loader(loader, model, device):
    """计算整个数据加载器的平均损失"""
    total_loss = 0.0
    num_batches = 0
    model.eval()
    with torch.no_grad():
        for x, y in loader:
            x, y = x.to(device), y.to(device)
            logits = model(x)
            loss = calc_loss_batch(logits, y)
            total_loss += loss.item()
            num_batches += 1
    return total_loss / max(num_batches, 1)

print("✓ 损失函数定义完成")

4. 训练循环

核心训练逻辑: 前向 → 算损失 → 反向传播 → 梯度裁剪 → 更新权重

代码

from torch.utils.tensorboard import SummaryWriter

def train_model(model, train_loader, val_loader, n_epochs, lr=5e-4, device='cpu'):
    optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.1)
    writer = SummaryWriter(log_dir='runs/gpt-training')
    
    for epoch in range(n_epochs):
        model.train()
        total_loss = 0.0
        for batch_idx, (x, y) in enumerate(train_loader):
            x, y = x.to(device), y.to(device)
            
            optimizer.zero_grad()
            logits = model(x)
            loss = calc_loss_batch(logits, y)
            loss.backward()
            
            # 梯度裁剪: 防止梯度爆炸
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            
            optimizer.step()
            total_loss += loss.item()
        
        avg_train_loss = total_loss / len(train_loader)
        avg_val_loss = calc_loss_loader(val_loader, model, device)
        
        # 记录到 TensorBoard
        writer.add_scalar('Loss/train', avg_train_loss, epoch)
        writer.add_scalar('Loss/val', avg_val_loss, epoch)
        writer.add_scalar('Perplexity/train', torch.exp(torch.tensor(avg_train_loss)).item(), epoch)
        writer.add_scalar('Perplexity/val', torch.exp(torch.tensor(avg_val_loss)).item(), epoch)
        
        if (epoch + 1) % 5 == 0:
            print(f"Epoch {epoch+1}/{n_epochs} | Train Loss: {avg_train_loss:.4f} | Val Loss: {avg_val_loss:.4f} | "
                  f"PPL: {torch.exp(torch.tensor(avg_val_loss)).item():.2f}")
    
    writer.close()
    print(f"训练完成! 查看训练曲线: tensorboard --logdir=runs")
    return model

print("✓ 训练函数定义完成")

5. 开始训练（微型模型）

用小配置快速演示训练过程。

代码

# TODO: 这里导入 07_GPT_Architecture 中定义的 GPTModel
# 或者直接运行下面的简化训练

# 因为训练需要完整的 GPTModel 定义, 实际请在完整版 notebook 中执行
print("在完整版本中, 你需要:")
print("1. 导入 07 中的 GPTModel")
print("2. 创建小配置: emb_dim=128, n_layers=4, n_heads=4")
print("3. 设置 device='mps' (M2 Max)")
print("4. 运行 train_model(...)")
print("5. 观察损失下降和困惑度变化")

6. 高级技术

学习率调度 (附录D)

# 预热 + 余弦衰减
def lr_lambda(current_step, warmup_steps, total_steps):
    if current_step < warmup_steps:
        return current_step / max(warmup_steps, 1)  # 线性升温
    # 余弦衰减
    progress = (current_step - warmup_steps) / max(total_steps - warmup_steps, 1)
    return 0.5 * (1 + math.cos(math.pi * progress))

scheduler = torch.optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)

梯度裁剪

torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)

权重初始化

def init_weights(module):
    if isinstance(module, (nn.Linear, nn.Embedding)):
        nn.init.normal_(module.weight, mean=0.0, std=0.02)

> 更多细节: 原书 Appendix D "在训练循环中添加额外功能"

代码

# %%ai openai-chat-custom:Qwen3.5-9B-Q4_K_M.gguf --format markdown
# 为什么LLM预训练要使用 "下一个token预测" 而不是其他目标函数？
# 这个目标函数有什么局限？(提示: 联系 Gary Marcus 对 LLM 的批判)

检查清单

[x] 理解交叉熵损失 + Perplexity
[x] 滑动窗口数据加载器
[x] 完整训练循环 (含梯度裁剪)
[x] TensorBoard 记录
[x] 学习率预热 + 余弦衰减
[ ] 下一步: 09_FineTuning.ipynb — 微调模型做分类任务

🔍 审计角度 (Audit Lens)

1. 训练数据合规 — 预训练的核心审计点

预训练数据是审计的第一道关口：

数据来源: 是否使用版权受保护的数据？（中国《生成式AI管理办法》要求合法数据来源）数据去重: 重复数据会放大偏见 — 审计需检查去重策略数据过滤: 是否过滤了PII/敏感内容？过滤标准是什么？数据比例: 各语言/领域的数据比例是否合理？

2. Loss 曲线审计

训练损失和验证损失的曲线蕴含了合规信息：

训练Loss↓ + 验证Loss↓ = 正常学习 (合规)
训练Loss↓ + 验证Loss↑ = 过拟合 (可能记忆了训练数据中的PII)
两个Loss都震荡 = 训练不稳定 (需要检查数据质量)
Loss突然跳升 = 数据中毒或训练事故 (需要记录审计日志)

你需要在算法备案中提交训练过程的监控记录。

3. 困惑度(Perplexity)的合规含义

低困惑度 → 模型"自信" — 但可能过于自信甚至"死记硬背" 风险: 如果模型背诵了训练数据中的隐私信息, 低困惑度反而危险高困惑度 → 模型"困惑" — 输出质量可能不达标合规要求: 关键场景(金融/医疗)的困惑度需要低于阈值

4. 梯度裁剪的审计日志

梯度裁剪本意是稳定训练, 但:
如果梯度范数经常触达裁剪阈值 → 可能说明数据有异常
建议记录: 每次backward后的梯度范数, 裁剪频率
这些数据在模型事故追溯时非常有用