# 从零构建GPT-1：用PyTorch实现一个能“思考”的文本生成器 几年前，当我第一次读到那篇名为《通过生成式预训练提升语言理解能力》的论文时，内心是有些震撼的。它不像当时许多研究那样，执着于为每个特定任务设计精巧的模型结构，而是提出了一种近乎“蛮横”的思路：先用海量无标签文本，让一个庞大的模型学会语言的通用规律，然后再用少量标注数据，像“微调”乐器一样，让它适应具体的任务。这个模型，后来被大家亲切地称为GPT-1。今天，我们不再满足于仅仅理解它的原理图。对于一名渴望深入AI腹地的开发者而言，真正的乐趣在于亲手将它从论文中“复活”，看着一行行代码逐渐汇聚成一个能够生成连贯文本的智能体。本文将带你踏上一段完整的实践之旅，使用PyTorch框架，从最基础的张量操作开始，一步步搭建、训练并微调你自己的GPT-1模型。无论你是想深入理解Transformer的运作机制，还是为后续探索更大型语言模型打下坚实基础，这次动手实践都将是一次宝贵的经验。 ## 1. 环境准备与核心概念再审视 在动手写代码之前，我们需要确保开发环境就绪，并对GPT-1的几个核心设计选择有清晰的认识。这不仅仅是安装几个库那么简单，更是为了在后续编码时，能理解每一个模块存在的意义。 首先，创建一个干净的Python虚拟环境总是个好习惯。这能避免包版本冲突带来的各种诡异问题。 ```bash conda create -n gpt1_env python=3.9 conda activate gpt1_env pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 根据你的CUDA版本调整 pip install transformers datasets tqdm numpy ``` 接下来，我们快速回顾一下GPT-1的三大支柱，这决定了我们代码库的顶层结构： 1. **纯解码器架构**：GPT-1只使用了Transformer架构中的解码器部分，并且是带掩码（Masked）的自注意力机制。这意味着在预测序列中下一个词时，模型只能“看到”它之前的词，无法“偷看”未来的信息，这完美契合了语言生成任务的自回归特性。 2. **生成式预训练**：其预训练目标极其简洁——给定前文，预测下一个词。这种任务被称为“语言建模”。通过在海量文本上完成这个任务，模型能内化语法、事实知识甚至一定的推理能力。 3. **任务无关的微调**：预训练后的模型就像一个掌握了通用语言知识的学生。微调阶段，我们不是改变它的“大脑结构”，而是通过一个简单的线性分类头，并继续在带标签的特定任务数据上训练，教会它如何运用已有知识解决具体问题（如分类、问答）。 理解这三点后，我们的代码蓝图也就清晰了：实现一个多层Transformer解码器块，构建语言模型预训练损失，最后嫁接上任务特定的微调接口。 ## 2. 搭建模型核心：Transformer解码器块 GPT-1的“心脏”是堆叠的Transformer解码器层。我们将从最基础的自注意力机制开始，逐层向上构建。 ### 2.1 掩码自注意力机制 自注意力机制允许序列中的每个位置与其他所有位置交互。但对于解码器，我们需要一个“掩码”来防止当前位置关注到未来的位置。在PyTorch中，我们可以利用 `torch.tril`（下三角矩阵）来轻松实现。 ```python import torch import torch.nn as nn import torch.nn.functional as F import math class CausalSelfAttention(nn.Module): def __init__(self, embed_dim, num_heads, dropout=0.1): super().__init__() assert embed_dim % num_heads == 0, "embed_dim 必须能被 num_heads 整除" self.num_heads = num_heads self.head_dim = embed_dim // num_heads self.scale = self.head_dim ** -0.5 # 缩放因子，稳定训练 # 将Q, K, V的投影合并到一个线性层中，提升效率 self.qkv_proj = nn.Linear(embed_dim, 3 * embed_dim) self.out_proj = nn.Linear(embed_dim, embed_dim) self.dropout = nn.Dropout(dropout) def forward(self, x, mask=None): """ x: 输入张量，形状为 (batch_size, seq_len, embed_dim) mask: 可选的注意力掩码 """ batch_size, seq_len, embed_dim = x.shape # 1. 生成Q, K, V qkv = self.qkv_proj(x) # (batch_size, seq_len, 3*embed_dim) qkv = qkv.reshape(batch_size, seq_len, 3, self.num_heads, self.head_dim) qkv = qkv.permute(2, 0, 3, 1, 4) # (3, batch_size, num_heads, seq_len, head_dim) q, k, v = qkv[0], qkv[1], qkv[2] # 2. 计算缩放点积注意力 attn_scores = torch.matmul(q, k.transpose(-2, -1)) * self.scale # (batch_size, num_heads, seq_len, seq_len) # 3. 应用因果掩码（防止信息泄露） causal_mask = torch.tril(torch.ones(seq_len, seq_len, device=x.device)).view(1, 1, seq_len, seq_len) attn_scores = attn_scores.masked_fill(causal_mask == 0, float('-inf')) # 4. 可选：添加其他掩码（如padding mask） if mask is not None: attn_scores = attn_scores.masked_fill(mask.unsqueeze(1).unsqueeze(2) == 0, float('-inf')) # 5. 计算注意力权重和输出 attn_weights = F.softmax(attn_scores, dim=-1) attn_weights = self.dropout(attn_weights) attn_output = torch.matmul(attn_weights, v) # (batch_size, num_heads, seq_len, head_dim) # 6. 合并多头，投影输出 attn_output = attn_output.transpose(1, 2).contiguous().view(batch_size, seq_len, embed_dim) output = self.out_proj(attn_output) return output ``` > 注意：这里的 `causal_mask` 是关键。它确保在计算位置 `i` 的注意力时，只考虑位置 `0` 到 `i` 的键（Key），未来位置的信息被掩码为负无穷（`-inf`），经过softmax后权重为0。 ### 2.2 前馈网络与层归一化 每个Transformer解码器块在自注意力层之后，都跟着一个前馈神经网络（FFN）和残差连接。GPT-1使用的FFN是一个两层MLP，中间有一个GELU激活函数。 ```python class FeedForward(nn.Module): def __init__(self, embed_dim, ff_dim, dropout=0.1): super().__init__() self.net = nn.Sequential( nn.Linear(embed_dim, ff_dim), nn.GELU(), # GPT-1/2使用GELU，比ReLU更平滑 nn.Linear(ff_dim, embed_dim), nn.Dropout(dropout) ) def forward(self, x): return self.net(x) class TransformerBlock(nn.Module): """一个完整的Transformer解码器块""" def __init__(self, embed_dim, num_heads, ff_dim, dropout=0.1): super().__init__() self.ln1 = nn.LayerNorm(embed_dim) self.attn = CausalSelfAttention(embed_dim, num_heads, dropout) self.ln2 = nn.LayerNorm(embed_dim) self.ff = FeedForward(embed_dim, ff_dim, dropout) self.dropout = nn.Dropout(dropout) def forward(self, x, mask=None): # 带残差连接和层归一化的自注意力 attn_output = self.attn(self.ln1(x), mask) x = x + self.dropout(attn_output) # 残差连接1 # 带残差连接和层归一化的前馈网络 ff_output = self.ff(self.ln2(x)) x = x + self.dropout(ff_output) # 残差连接2 return x ``` 现在，我们已经有了构建整个GPT-1模型的基石。接下来，我们需要将这些块组合起来，并添加上下文嵌入层。 ## 3. 组装GPT-1模型与预训练目标 一个完整的GPT-1模型由词嵌入层、位置编码层以及N个堆叠的Transformer解码器块组成，最后接一个用于预测下一个词的线性输出层。 ### 3.1 构建完整模型 ```python class GPT1(nn.Module): def __init__(self, vocab_size, max_seq_len, embed_dim, num_layers, num_heads, ff_dim, dropout=0.1): super().__init__() self.max_seq_len = max_seq_len self.token_embed = nn.Embedding(vocab_size, embed_dim) self.pos_embed = nn.Embedding(max_seq_len, embed_dim) # 构建Transformer解码器层堆栈 self.blocks = nn.ModuleList([ TransformerBlock(embed_dim, num_heads, ff_dim, dropout) for _ in range(num_layers) ]) self.ln_f = nn.LayerNorm(embed_dim) # 最后的层归一化 self.lm_head = nn.Linear(embed_dim, vocab_size, bias=False) # 语言模型头 # 权重初始化（对稳定训练很重要） self.apply(self._init_weights) def _init_weights(self, module): if isinstance(module, nn.Linear): torch.nn.init.normal_(module.weight, mean=0.0, std=0.02) if module.bias is not None: torch.nn.init.zeros_(module.bias) elif isinstance(module, nn.Embedding): torch.nn.init.normal_(module.weight, mean=0.0, std=0.02) def forward(self, idx, targets=None): """ idx: 输入token索引，形状为 (batch_size, seq_len) targets: 目标token索引，用于计算损失，形状同idx 返回：logits (batch_size, seq_len, vocab_size)， 可选损失值 """ device = idx.device batch_size, seq_len = idx.shape assert seq_len <= self.max_seq_len, f"序列长度{seq_len}超过最大长度{self.max_seq_len}" # 1. 创建位置索引并获取嵌入 pos = torch.arange(0, seq_len, dtype=torch.long, device=device).unsqueeze(0) # (1, seq_len) tok_emb = self.token_embed(idx) # (batch_size, seq_len, embed_dim) pos_emb = self.pos_embed(pos) # (1, seq_len, embed_dim) x = tok_emb + pos_emb # 相加融合 # 2. 通过所有Transformer块 for block in self.blocks: x = block(x) x = self.ln_f(x) logits = self.lm_head(x) # (batch_size, seq_len, vocab_size) # 3. 计算损失（如果提供了targets） loss = None if targets is not None: # 将logits和targets展平以计算交叉熵损失 loss = F.cross_entropy(logits.view(-1, logits.size(-1)), targets.view(-1)) return logits, loss ``` ### 3.2 预训练：语言建模任务 预训练的目标是标准的语言建模：给定一个token序列，预测下一个token。我们的模型已经为此设计好了。我们只需要准备一个合适的数据集，并设置优化循环。 假设我们有一个简单的文本数据集，我们可以按如下方式创建一个数据加载器： ```python from torch.utils.data import Dataset, DataLoader class TextDataset(Dataset): def __init__(self, text_path, tokenizer, block_size): with open(text_path, 'r', encoding='utf-8') as f: text = f.read() tokens = tokenizer.encode(text) # 假设tokenizer返回一个整数列表 self.data = torch.tensor(tokens, dtype=torch.long) self.block_size = block_size def __len__(self): return len(self.data) - self.block_size def __getitem__(self, idx): # 取一段连续的token作为输入，下一个token作为目标 chunk = self.data[idx: idx + self.block_size + 1] x = chunk[:-1] y = chunk[1:] return x, y # 示例：使用简单的字符级tokenizer（实际应用应使用BPE等） class CharTokenizer: def __init__(self, text): chars = sorted(list(set(text))) self.stoi = {ch: i for i, ch in enumerate(chars)} self.itos = {i: ch for i, ch in enumerate(chars)} self.vocab_size = len(chars) def encode(self, s): return [self.stoi.get(c, 0) for c in s] def decode(self, l): return ''.join([self.itos.get(i, '') for i in l]) ``` 有了数据和模型，预训练循环的核心代码如下： ```python def train_epoch(model, dataloader, optimizer, device): model.train() total_loss = 0 for batch_idx, (x, y) in enumerate(dataloader): x, y = x.to(device), y.to(device) optimizer.zero_grad() _, loss = model(x, targets=y) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 梯度裁剪，防止爆炸 optimizer.step() total_loss += loss.item() return total_loss / len(dataloader) ``` ## 4. 微调策略与多任务适配 预训练后的模型已经具备了强大的语言表示能力。微调阶段，我们需要根据下游任务调整模型。GPT-1论文的精妙之处在于，它通过**输入变换**将不同的任务（如分类、蕴含、相似度）都统一成了类似“文本续写”的格式，从而无需改变模型主体结构，只需在顶部添加一个线性分类器。 ### 4.1 输入变换与任务适配 对于不同的NLP任务，我们需要在输入序列中插入特殊的标记来指示任务类型和结构。常见的标记包括： * `[SOS]`：序列开始 * `[EOS]`：序列结束 * `[DELIM]`：分隔符，用于分隔多个句子 下表展示了如何将不同任务格式化为模型可接受的输入： | 任务类型 | 输入示例 (格式化前) | 格式化后输入序列 | 预测目标 | | :--- | :--- | :--- | :--- | | **文本分类** | 句子: `"这部电影很棒。"` | `[SOS] 这部电影很棒。 [EOS]` | 类别标签 (如 `positive`) | | **文本蕴含** | 前提: `"猫在垫子上"`, 假设: `"垫子上有动物"` | `[SOS] 猫在垫子上 [DELIM] 垫子上有动物 [EOS]` | 蕴含关系标签 (如 `entailment`) | | **语义相似度** | 句子A: `"今天天气好"`, 句子B: `"阳光明媚"` | `[SOS] 今天天气好 [DELIM] 阳光明媚 [EOS]` | 相似度分数 (如 `1`) | | **多项选择QA** | 文档: `...`, 问题: `Q?`, 选项: `A1; A2; ...` | `[SOS] 文档... [DELIM] Q? [DELIM] A1 [EOS]` | 正确选项的索引 | 在代码中，我们需要扩展词汇表以包含这些特殊标记，并在数据预处理阶段完成格式转换。 ### 4.2 微调模型架构与损失函数 微调时，我们在预训练好的GPT-1模型上添加一个任务特定的线性分类头。损失函数是预训练语言模型损失和任务监督损失的加权和，这有助于防止在少量标注数据上“遗忘”预训练获得的知识（灾难性遗忘）。 ```python class GPT1ForFineTuning(nn.Module): def __init__(self, base_gpt, num_classes, lambda_lm=0.5): """ base_gpt: 预训练好的GPT1模型 num_classes: 下游任务的类别数（如情感分类的2类） lambda_lm: 语言模型损失项的权重系数 """ super().__init__() self.gpt = base_gpt self.lambda_lm = lambda_lm # 任务分类头：取最后一个token（通常是[EOS]）的表示进行分类 self.task_head = nn.Linear(self.gpt.token_embed.embedding_dim, num_classes) def forward(self, input_ids, attention_mask=None, labels=None): """ input_ids: 格式化后的token ID序列 labels: 任务标签 """ # 获取GPT的最后一层隐藏状态 hidden_states, lm_loss = self.gpt(input_ids, targets=input_ids) # 这里targets用input_ids自身计算LM损失 # 取序列最后一个有效token的表示（对应[EOS]） last_token_states = hidden_states[:, -1, :] # (batch_size, embed_dim) # 通过任务头得到logits task_logits = self.task_head(last_token_states) # (batch_size, num_classes) total_loss = None if labels is not None: # 计算任务分类损失 task_loss = F.cross_entropy(task_logits, labels) # 总损失 = 任务损失 + λ * 语言模型损失 total_loss = task_loss + self.lambda_lm * lm_loss return task_logits, total_loss ``` 在微调训练循环中，我们同时更新`task_head`和`base_gpt`的所有参数。`lambda_lm`是一个重要的超参数，通常设置为0.1到0.5之间，用于平衡对新任务的学习和对原有语言知识的保留。 ## 5. 实战演练：训练一个微型诗歌生成器 理论说得再多，不如跑通一个完整的例子。让我们用一个小型中文诗歌数据集，训练一个能生成五言绝句的微型GPT-1。 ### 5.1 数据准备与训练 我们使用一个包含数万首五言诗的数据集。首先进行数据预处理： ```python # 假设 poems.txt 中每行是一首五言绝句（20个字） with open('poems.txt', 'r', encoding='utf-8') as f: poems = [line.strip() for line in f if len(line.strip()) == 20] # 构建词汇表（字级别） all_chars = set(''.join(poems)) tokenizer = CharTokenizer(''.join(all_chars)) # 使用之前定义的简单Tokenizer vocab_size = tokenizer.vocab_size # 创建数据集和数据加载器 block_size = 19 # 预测下一个字，输入长度为19 dataset = TextDataset('poems.txt', tokenizer, block_size) # 需要适配TextDataset以读取列表 dataloader = DataLoader(dataset, batch_size=32, shuffle=True) # 初始化一个微型GPT-1模型 model_config = { 'vocab_size': vocab_size, 'max_seq_len': 20, 'embed_dim': 128, # 嵌入维度 'num_layers': 4, # Transformer层数 'num_heads': 4, # 注意力头数 'ff_dim': 512, # 前馈网络隐藏层维度 'dropout': 0.1 } model = GPT1(**model_config).to(device) optimizer = torch.optim.AdamW(model.parameters(), lr=6e-4) # 训练循环 num_epochs = 10 for epoch in range(num_epochs): avg_loss = train_epoch(model, dataloader, optimizer, device) print(f"Epoch {epoch+1}, Loss: {avg_loss:.4f}") # 可以在这里保存模型检查点 ``` ### 5.2 文本生成与推理 训练完成后，最激动人心的部分来了：让模型创作诗歌。我们使用**自回归**的方式，每次生成一个字。 ```python def generate_poem(model, tokenizer, start_char='春', max_len=20, temperature=0.8): """ 使用训练好的模型生成诗歌。 temperature: 控制随机性，越高越随机，越低越确定。 """ model.eval() with torch.no_grad(): # 初始化输入为开始字符 input_ids = torch.tensor([[tokenizer.stoi[start_char]]], device=device) generated = [start_char] for _ in range(max_len - 1): # 前向传播，获取下一个字的logits logits, _ = model(input_ids) # 取最后一个时间步的logits，并应用temperature next_token_logits = logits[:, -1, :] / temperature # 通过softmax得到概率分布 probs = F.softmax(next_token_logits, dim=-1) # 根据概率分布采样下一个字（而非总是选概率最高的） next_token_id = torch.multinomial(probs, num_samples=1) # 将新生成的token加入输入序列 input_ids = torch.cat([input_ids, next_token_id], dim=1) # 解码并记录 next_char = tokenizer.itos[next_token_id.item()] generated.append(next_char) # 如果生成了结束符或达到长度，则停止 if next_char == '[EOS]' or len(generated) >= max_len: break return ''.join(generated[:max_len]) # 尝试生成几首 for start in ['春', '月', '山', '孤']: poem = generate_poem(model, tokenizer, start_char=start, temperature=0.7) print(f"首字 '{start}': {poem}") ``` 你可能会得到像“春风吹又生，花落知多少”这样虽然简单但已具雏形的句子。通过调整`temperature`参数，你可以在“保守遵循训练数据”和“天马行空自由发挥”之间找到平衡。 ### 5.3 常见陷阱与调优技巧 在实践过程中，你几乎一定会遇到模型不收敛、生成结果毫无意义或者过拟合等问题。下面是一些关键的检查点和调优经验： - **损失不下降**：首先检查梯度。在训练循环中加入梯度范数打印，如果梯度接近0，可能是初始化问题或激活函数饱和；如果梯度爆炸，需要减小学习率或增加梯度裁剪的阈值。 - **生成重复或无意义内容**：这通常是**曝光偏差**的早期表现。模型在训练时总是看到真实的“上文”，但在生成时，它用的是自己之前生成的、可能有错误的“上文”。除了调整`temperature`，更高级的解决方法是使用**束搜索**或**核采样**。 - **过拟合**：在小数据集上微调时尤其明显。除了使用Dropout，**权重衰减**和**早停**是必须的。监控验证集上的损失，当它连续几个epoch不再下降时，就停止训练。 - **内存溢出**：Transformer的内存消耗与序列长度的平方成正比。对于长文本，需要关注`max_seq_len`的设置，或者研究**内存高效的注意力**实现。 最后，别忘了保存和加载你的模型成果： ```python # 保存 torch.save({ 'model_state_dict': model.state_dict(), 'config': model_config, 'tokenizer': tokenizer.stoi, # 保存词汇表映射 }, 'gpt1_poetry.pth') # 加载 checkpoint = torch.load('gpt1_poetry.pth') loaded_model = GPT1(**checkpoint['config']).to(device) loaded_model.load_state_dict(checkpoint['model_state_dict']) ``` 从零开始实现GPT-1的过程，就像在亲手组装一台精密的钟表。每一个注意力头，每一层归一化，都对应着论文中一个抽象的数学公式。当代码成功运行，屏幕上开始跳出由模型生成的、略显生涩却结构完整的句子时，那种成就感是单纯阅读论文无法比拟的。这次实践不仅让你掌握了Transformer解码器的实现细节，更重要的是，你建立起了对大语言模型如何“工作”的直观感受。这为后续理解更复杂的模型，如GPT-2/3的上下文学习，甚至尝试修改架构进行创新，打下了最坚实的地基。记住，在AI的世界里，理解最深的方式永远是亲手构建它。