目录
2、介绍GPT自监督预训练、有监督下游任务微调及预训练语言模型
2.1.2、 Masked 多头注意力：禁止 “偷看” 未来信息
三、预处理数据集：把 “课本” 翻译成 “机器人能懂的语言”
1、GPT的模型结构如图所示
它是由多层Transformer组陈的单向语言模型，主要分为输入层、编码层和输出层三个部分：
2、介绍GPT自监督预训练、有监督下游任务微调及预训练语言模型
2.1、GPT 自监督预训练
GPT 预训练的核心是基于 Transformer Decoder 的因果语言建模，其计算过程可通过具体示例拆解为 “输入编码 - 注意力计算 - 损失优化” 三步骤。
2.1.1、 输入编码：词向量与位置向量的融合
公式描述了输入编码过程，用示例说明：
- 假设输入序列为 “猫吃鱼”，分词后为 3 个 token：x' = [猫, 吃, 鱼]；
- 词向量查表：e_{猫}通过词向量矩阵
（假设维度为 3×5）映射为向量
，同理 “吃”“鱼” 分别映射为
和
；
- 位置向量叠加：位置 1（猫）的向量
，位置 2（吃）为
，叠加后
的第一个向量为
。
2.1.1.1、 输入序列与词表映射
假设我们有一个简单的词表，包含 3 个词：
词表 = {"猫": 0, "吃": 1, "鱼": 2}输入文本 "猫吃鱼" 被分词为 3 个 token，对应的词表索引为：
x' = [0, 1, 2]2.1.1.2、 词向量矩阵与查表操作
词向量矩阵
查表过程：
- 对于 token "猫"（索引 0），其词向量为
- 同理，"吃" 的词向量为
- "鱼" 的词向量为
3. 位置向量矩阵
位置向量用于表示 token 在序列中的位置信息。假设位置向量维度同样为 5，则 3 个位置的向量分别为：
4. 词向量与位置向量叠加
根据公式
，对每个 token 的词向量和对应位置向量进行叠加：
第一个 token "猫"（位置 1）：
第二个 token "吃"（位置 2）：
第三个 token "鱼"（位置 3）：
5. 最终输入向量
将上述三个叠加后的向量组合，得到最终输入到 Transformer 的向量\(h^{[0]}\)：
2.1.2、 Masked 多头注意力：禁止 “偷看” 未来信息
文章详细描述了掩码注意力的计算逻辑，用 “猫吃鱼” 示例：
步骤 1：线性变换拆分多头 假设多头注意力头数h=2，每个头维度
。输入\
（维度 3×5），通过
（5×4）拆分后，每个头的
、
、
维度为 3×2。
步骤 2：带掩码的缩放点积 计算注意力分数时，掩码矩阵M遮挡未来位置（如下表，× 表示被遮挡）：
关注对象→ 猫（位置 1） 吃（位置 2） 鱼（位置 3） 猫（预测吃） √ × × 吃（预测鱼） √ √ × 对应分数计算：
，被遮挡位置填充-∞，经 Softmax 后权重为 0。
步骤 3：残差连接与层归一化 注意力输出与原始输入
2.1.3、 损失函数：优化预测概率
公式的示例：
- 对于 “猫吃鱼”，模型需预测：
- 已知 “猫”，预测 “吃” 的概率P(吃|猫)；
- 已知 “猫、吃”，预测 “鱼” 的概率P(鱼|猫,吃)。
- 假设模型输出概率为P(吃|猫)=0.8，P(鱼|猫,吃)=0.7，则损失为：
，训练目标是最小化该值。
2.2 有监督下游任务微调
微调的核心是组合损失函数，以 “电影评论情感分类” 为例说明：
2.2.1、 任务适配：从文本到标签的映射
- 输入：“这部电影剧情很棒！”→ 经预训练模型编码后，取最后一层输出
；
- 输出映射：通过公式
，其中
为分类权重矩阵（假设情感分 2 类，维度 5×2），输出P(正面)=0.9，P(负面)=0.1。
2.2.2、组合损失：平衡任务与预训练知识
公式的示例：
- 下游任务损失
：该评论标签为 “正面”，计算交叉熵损失
；
- 预训练损失
：用 “这部电影剧情很棒” 预测下一个词（如 “！”），假设
，损失
；
- 组合损失：取
，则总损失=0.105 + 0.3×0.223≈0.172，既优化分类能力，又保留语言建模能力。
2.3 预训练语言模型
GPT 是 “简化的 Transformer Decoder”，通过与传统模型对比理解其优势：
2.3.1、 结构差异：从 “专用设计” 到 “通用基座”
- 传统模型：如情感分类器需手动设计特征（如 “包含‘棒’‘好’等词”），泛化能力弱；
- GPT（PLM）：通过 12 层 Transformer Decoder 堆叠（文章图 7.2），自动学习特征。例如同样处理情感分类，无需人工设计规则，直接通过微调适配。
2.3.2、 能力边界：生成 vs 理解
- GPT（自回归）：适合生成任务。输入 “周末我想去____”，模型续写 “爬山，呼吸新鲜空气”（符合文章所述 “Mask 机制确保单向生成”）；
- BERT（自编码）：适合理解任务。输入 “周末我想 [MASK] 爬山”，补全 “去”（利用双向注意力）。
通过具体示例可见，GPT 的预训练是 “从数据学规律”，微调是 “从规律到任务”，而预训练语言模型则是这一过程的 “通用载体”，三者通过数学公式和结构设计紧密衔接，实现从文本到能力的转化。
3、模型验证
3.1、 GPT 模型全流程
- 准备数据集：获取原始文本并划分训练 / 测试集；
- 训练词元分析器：分析词汇特征，构建子词词汇表；
- 预处理数据集：将文本转换为模型可接受的整数序列；
- 训练模型：基于 Transformer 架构训练自回归语言模型；
- 运用模型：通过提示文本生成连贯的新内容。
3.2、GPT 核心逻辑突出：
- 因果掩码确保自回归特性（只能看前文）；
- 词嵌入 + 位置嵌入融合语义和位置信息；
- 多头注意力捕捉不同维度的依赖关系；
- 文本生成采用迭代采样策略，支持温度调节。
4、完整实现
4.1、完整代码
"""
文件名: complete_gpt_pipeline.py
功能: GPT模型全流程实现（数据准备→词元分析→预处理→训练→生成）
"""
# 基础库导入
import os # 文件路径操作
import glob # 批量文件查找
import random # 随机采样
import re # 正则表达式（文本清洗）
import pickle # 保存/加载词元分析器
import torch # 深度学习框架
import torch.nn as nn # 神经网络模块
import torch.optim as optim # 优化器
import torch.nn.functional as F # 激活函数等
from torch.utils.data import Dataset, DataLoader # 数据集和数据加载器
from datasets import Dataset as HFDataset, concatenate_datasets # HuggingFace数据集工具
from tqdm import tqdm # 进度条
from collections import Counter # 计数工具
# ---------------------------一、准备数据集（数据来源与划分）---------------------------
def load_local_text_data(data_dir, file_pattern='*.txt', max_files=None, encoding='utf-8'):
"""
从本地目录加载文本文件并转换为数据集
参数:
data_dir: 文本文件所在目录
file_pattern: 文件名匹配模式（默认所有.txt文件）
max_files: 最大加载文件数（控制数据量）
encoding: 文件编码（默认utf-8）
返回:
HuggingFace Dataset对象，包含'text'列
"""
print(f"加载本地数据：{data_dir}")
# 查找目录中所有匹配的文件路径
file_paths = glob.glob(os.path.join(data_dir, file_pattern))
if not file_paths:
raise ValueError(f"未找到匹配文件：{data_dir}/{file_pattern}") # 无文件时报错
# 若文件过多，随机采样（控制内存占用）
if max_files and len(file_paths) > max_files:
random.seed(42) # 固定随机种子，确保结果可复现
file_paths = random.sample(file_paths, max_files)
texts = []
for path in file_paths:
try:
with open(path, 'r', encoding=encoding) as f:
text = f.read().strip() # 读取并去除首尾空白
if text: # 只保留非空文本（避免无效数据）
texts.append(text)
except Exception as e:
print(f"跳过无效文件 {path}：{e}") # 单个文件错误不中断整体流程
# 转换为HuggingFace Dataset（方便后续处理）
return HFDataset.from_dict({'text': texts})
def generate_sample_data(save_dir='./sample_data'):
"""
生成示例文本数据（当无本地数据时使用）
参数:
save_dir: 示例数据保存目录
返回:
保存目录路径
"""
os.makedirs(save_dir, exist_ok=True) # 确保目录存在（exist_ok=True避免重复创建报错）
# 加长示例文本（避免因过短导致后续预处理失败）
sample_texts = [
"GPT是基于Transformer的生成式预训练模型。它通过自监督学习从海量文本中学习语言规律。",
"预训练后，模型可通过微调适配具体任务，如文本生成、问答系统、机器翻译等。",
"训练GPT需要大量计算资源和高质量文本数据，常见数据来源包括书籍、网页和百科。",
"文本预处理是训练前的关键步骤，包括分词、清洗、归一化等操作，直接影响模型性能。",
"Transformer架构的核心是自注意力机制，能有效捕捉文本中的长距离依赖关系。"
]
# 保存为多个文本文件（模拟真实数据分布）
for i, text in enumerate(sample_texts):
with open(os.path.join(save_dir, f"sample_{i}.txt"), 'w', encoding='utf-8') as f:
f.write(text)
return save_dir
def prepare_dataset():
"""
主函数：准备训练集和测试集原始文本文件
返回:
train_raw: 训练集文本文件路径（train_raw.txt）
test_raw: 测试集文本文件路径（test_raw.txt）
"""
# 优先使用本地数据，否则生成示例数据
if os.path.exists('./local_data') and os.path.isdir('./local_data'):
dataset = load_local_text_data('./local_data')
else:
print("未找到本地数据，使用示例数据...")
sample_dir = generate_sample_data()
dataset = load_local_text_data(sample_dir)
# 过滤空文本（避免后续处理出错）
dataset = dataset.filter(lambda x: len(x['text'].strip()) > 0)
if len(dataset) == 0:
raise ValueError("数据集为空，请检查数据质量") # 无有效数据时终止流程
# 划分训练集（90%）和测试集（10%），固定随机种子确保划分一致
train_test = dataset.train_test_split(test_size=0.1, seed=42)
train_dataset = train_test['train']
test_dataset = train_test['test']
# 保存为文本文件（每行一条数据，方便后续加载）
with open('train_raw.txt', 'w', encoding='utf-8') as f:
f.write('\n'.join(train_dataset['text']))
with open('test_raw.txt', 'w', encoding='utf-8') as f:
f.write('\n'.join(test_dataset['text']))
print(f"数据集准备完成：训练集{len(train_dataset)}条，测试集{len(test_dataset)}条")
return 'train_raw.txt', 'test_raw.txt'
# ---------------------------二、训练词元分析器（子词提取与统计）---------------------------
class WordPieceAnalyzer:
"""
基于WordPiece算法的词元分析器
功能：从文本中学习子词词汇表，实现分词，并统计词频特征
"""
def __init__(self, vocab_size=1000):
self.vocab = set() # 子词词汇表（存储学到的子词）
self.vocab_size = vocab_size # 目标词汇表大小
self.max_subword_len = 6 # 最大子词长度（控制拆分粒度）
def fit(self, text):
"""
从文本中学习子词词汇表（核心：迭代合并高频子词对）
参数:
text: 原始文本字符串
"""
# 初始词汇表：单个字符（最小子词单位）
chars = list(set(text))
self.vocab = set(chars)
# 迭代合并子词对，直到达到目标词汇表大小
while len(self.vocab) < self.vocab_size:
# 统计所有子词对的出现频率
pairs = Counter()
tokens = self.tokenize(text, use_vocab=True) # 用当前词汇表分词
for i in range(len(tokens)-1):
pair = (tokens[i], tokens[i+1]) # 相邻子词对
pairs[pair] += 1 # 计数
if not pairs: # 无更多可合并的子词对（提前终止）
break
# 合并频率最高的子词对（WordPiece核心逻辑）
best_pair = pairs.most_common(1)[0][0] # 取频率最高的对子
new_subword = ''.join(best_pair) # 合并为新子词
self.vocab.add(new_subword) # 加入词汇表
print(f"新增子词：{new_subword}（当前词汇表大小：{len(self.vocab)}）")
def tokenize(self, text, use_vocab=False):
"""
将文本拆分为子词（带##前缀标识非起始子词）
参数:
text: 输入文本
use_vocab: 是否使用已学习的词汇表（False时仅按规则拆分）
返回:
子词列表（如["GPT", "##模型", "##可以"]）
"""
# 清洗文本：保留字母、数字、中文，去除其他符号
text = re.sub(r'[^\w\s\u4e00-\u9fa5]', '', text)
tokens = []
i = 0
while i < len(text):
matched = False
# 最长匹配原则：优先匹配长个子词
for l in range(min(self.max_subword_len, len(text)-i), 0, -1):
subword = text[i:i+l] # 截取长度为l的子串
# 若使用词汇表，需检查子词是否在表中；否则直接拆分
if (use_vocab and subword in self.vocab) or (not use_vocab):
tokens.append(subword)
i += l # 移动指针
matched = True
break
if not matched: # 未匹配到子词，按单个字符拆分
tokens.append(text[i])
i += 1
# 为非起始子词添加##前缀（区分是否为词的开头）
result = []
for i, token in enumerate(tokens):
if i > 0 and token in self.vocab: # 非第一个且在词汇表中
result.append(f"##{token}")
else:
result.append(token)
return result
def analyze(self, text):
"""
分析文本的词频和子词分布特征
参数:
text: 输入文本
返回:
包含词频统计的字典
"""
# 停用词表（过滤无意义词汇）
STOPWORDS = {'的', '了', '在', '是', 'a', 'an', 'the'}
tokens = self.tokenize(text) # 分词
filtered = [t for t in tokens if t not in STOPWORDS] # 过滤停用词
subword_counts = Counter(filtered) # 子词频率统计
# 从子词重构完整词汇（合并##前缀的子词）
words = []
current_word = []
for t in filtered:
if t.startswith('##'):
current_word.append(t[2:]) # 去除##前缀
else:
if current_word: # 保存上一个完整词
words.append(''.join(current_word))
current_word = [t] # 开始新的词
word_counts = Counter(words) # 完整词频率统计
return {
'total_tokens': len(filtered), # 有效词元总数
'unique_subwords': len(subword_counts), # 去重子词数
'top_subwords': subword_counts.most_common(10), # 高频子词TOP10
'top_words': word_counts.most_common(10) # 高频完整词TOP10
}
def train_analyzer(train_file):
"""
训练词元分析器并生成分析报告
参数:
train_file: 训练集文本文件路径
返回:
训练好的WordPieceAnalyzer实例
"""
# 加载训练集文本
with open(train_file, 'r', encoding='utf-8') as f:
text = f.read()
# 初始化分析器（小词汇表，适合演示）
analyzer = WordPieceAnalyzer(vocab_size=50)
print("开始训练词元分析器...")
analyzer.fit(text) # 学习子词词汇表
# 生成分析报告
report = analyzer.analyze(text)
print("\n【词元分析报告】")
print(f"总有效词元数：{report['total_tokens']}")
print(f"去重子词数：{report['unique_subwords']}")
print("高频子词TOP10：", report['top_subwords'])
print("高频词汇TOP10：", report['top_words'])
# 保存分析器（供后续预处理使用）
with open('wordpiece_analyzer.pkl', 'wb') as f:
pickle.dump(analyzer, f)
print("词元分析器已保存至 wordpiece_analyzer.pkl")
return analyzer
# ---------------------------三、预处理数据集（适配模型输入格式）---------------------------
class GPTDataset(Dataset):
"""
GPT模型专用数据集
功能：将文本转换为模型可接受的输入格式（上下文窗口+下一个token预测）
"""
def __init__(self, file_path, analyzer, subword_to_idx, block_size=32):
"""
参数:
file_path: 原始文本文件路径
analyzer: 训练好的WordPieceAnalyzer
subword_to_idx: 子词到索引的映射表
block_size: 上下文窗口大小（一次输入的词元数）
"""
self.block_size = block_size # 上下文窗口大小
self.subword_to_idx = subword_to_idx # 子词→索引映射
self.analyzer = analyzer # 词元分析器
# 加载文本并转换为词元索引
with open(file_path, 'r', encoding='utf-8') as f:
text = f.read()
# 分词并转换为索引（未知子词用<unk>的索引）
tokens = analyzer.tokenize(text)
self.indices = [
subword_to_idx.get(t, subword_to_idx['<unk>']) for t in tokens
]
# 处理文本过短问题
if len(self.indices) < self.block_size:
print(f"警告：{file_path} 文本过短（{len(self.indices)}词元），小于block_size（{block_size}）")
# 重复文本以满足最小长度（仅演示用，实际应增加数据）
self.indices = self.indices * (self.block_size // len(self.indices) + 1)
print(f"预处理完成：{file_path} 转换为 {len(self.indices)} 个词元索引")
def __len__(self):
"""返回数据集样本数（确保非负）"""
# 样本数 = 总词元数 - 窗口大小（每个窗口对应一个样本）
return max(0, len(self.indices) - self.block_size) # 确保≥0
def __getitem__(self, idx):
"""
获取单个样本（输入-目标对）
参数:
idx: 样本索引
返回:
x: 输入序列（[idx, idx+1, ..., idx+block_size-1]）
y: 目标序列（[idx+1, ..., idx+block_size]）→ 预测下一个词元
"""
x = self.indices[idx:idx+self.block_size] # 输入窗口
y = self.indices[idx+1:idx+self.block_size+1] # 目标窗口（输入的偏移+1）
return torch.tensor(x, dtype=torch.long), torch.tensor(y, dtype=torch.long)
def prepare_dataloaders(train_file, test_file, analyzer, batch_size=8, block_size=32):
"""
准备训练集和测试集的数据加载器
参数:
train_file/test_file: 训练/测试文本文件路径
analyzer: 词元分析器
batch_size: 批次大小
block_size: 上下文窗口大小
返回:
数据加载器、子词映射表、词汇表大小、窗口大小
"""
# 定义特殊符号（填充符和未知词）
special_tokens = {'<pad>': 0, '<unk>': 1}
# 构建子词列表（特殊符号+学到的子词）
subword_list = list(special_tokens.keys()) + list(analyzer.vocab)
# 子词→索引映射（用于将子词转换为模型输入的整数）
subword_to_idx = {t: i for i, t in enumerate(subword_list)}
vocab_size = len(subword_to_idx) # 词汇表大小
print(f"预处理词汇表大小：{vocab_size}（含特殊符号）")
# 创建数据集
train_dataset = GPTDataset(
train_file, analyzer, subword_to_idx, block_size=block_size
)
test_dataset = GPTDataset(
test_file, analyzer, subword_to_idx, block_size=block_size
)
# 检查数据集是否为空
if len(train_dataset) == 0:
raise ValueError("训练集为空，请增大block_size或增加数据量")
if len(test_dataset) == 0:
raise ValueError("测试集为空，请增大block_size或增加数据量")
# 创建数据加载器（批量加载数据，支持打乱）
train_loader = DataLoader(
train_dataset, batch_size=batch_size, shuffle=True, drop_last=True # 训练集打乱
)
test_loader = DataLoader(
test_dataset, batch_size=batch_size, shuffle=False, drop_last=True # 测试集不打乱
)
print(f"数据加载器准备完成：训练集{len(train_loader)}批，测试集{len(test_loader)}批")
return train_loader, test_loader, subword_to_idx, vocab_size, block_size
# ---------------------------四、训练模型（GPT核心实现）---------------------------
class MultiHeadAttention(nn.Module):
"""多头自注意力模块（Transformer核心组件）"""
def __init__(self, embed_dim, num_heads, block_size):
"""
参数:
embed_dim: 嵌入维度（模型隐藏层维度）
num_heads: 注意力头数（并行注意力机制）
block_size: 上下文窗口大小（用于生成掩码）
"""
super().__init__()
self.embed_dim = embed_dim # 总嵌入维度
self.num_heads = num_heads # 头数
self.head_dim = embed_dim // num_heads # 每个头的维度（必须整除）
# 线性变换：将输入转换为Q（查询）、K（键）、V（值）
self.qkv_proj = nn.Linear(embed_dim, 3 * embed_dim) # 一次性计算QKV
self.out_proj = nn.Linear(embed_dim, embed_dim) # 注意力输出投影
self.dropout = nn.Dropout(0.1) # 防止过拟合
# 因果掩码（上三角矩阵）：确保只能看到前文（核心！GPT是自回归模型）
self.register_buffer('mask', torch.triu(torch.ones(block_size, block_size), diagonal=1).bool())
def forward(self, x):
"""
前向传播：计算多头自注意力
参数:
x: 输入张量，形状(B, T, C)→(批次, 时间步, 嵌入维度)
返回:
注意力输出，形状(B, T, C)
"""
B, T, C = x.shape # B:批次, T:时间步, C:嵌入维度
# 计算Q、K、V并拆分多头
qkv = self.qkv_proj(x).view(B, T, 3, self.num_heads, self.head_dim).permute(2, 0, 3, 1, 4)
q, k, v = qkv.unbind(0) # 拆分Q、K、V，形状均为(B, H, T, D)→H:头数, D:头维度
# 计算注意力分数：Q·K^T / sqrt(D)（缩放点积注意力）
attn_scores = (q @ k.transpose(-2, -1)) / (self.head_dim ** 0.5)
# 应用因果掩码：遮挡未来位置（置为负无穷，softmax后为0）
attn_scores = attn_scores.masked_fill(self.mask[:T, :T], -float('inf'))
# 转换为概率分布
attn_probs = F.softmax(attn_scores, dim=-1)
attn_probs = self.dropout(attn_probs) # dropout防止过拟合
# 注意力加权求和：概率×V
out = attn_probs @ v # (B, H, T, D)
# 拼接多头结果并投影
out = out.transpose(1, 2).contiguous().view(B, T, C) # 合并头维度
return self.out_proj(out) # 输出投影
class TransformerBlock(nn.Module):
"""Transformer解码器块（自注意力+前馈网络）"""
def __init__(self, embed_dim, num_heads, block_size):
super().__init__()
self.attn = MultiHeadAttention(embed_dim, num_heads, block_size) # 多头自注意力
# 前馈网络（两层线性+激活函数）
self.ffn = nn.Sequential(
nn.Linear(embed_dim, 4 * embed_dim), # 升维
nn.GELU(), # 激活函数（优于ReLU，有梯度平滑特性）
nn.Linear(4 * embed_dim, embed_dim), # 降维回原维度
nn.Dropout(0.1) # dropout
)
self.norm1 = nn.LayerNorm(embed_dim) # 层归一化（稳定训练）
self.norm2 = nn.LayerNorm(embed_dim)
self.dropout = nn.Dropout(0.1)
def forward(self, x):
"""
前向传播：残差连接+层归一化（Transformer标准结构）
参数:
x: 输入张量(B, T, C)
返回:
输出张量(B, T, C)
"""
# 自注意力+残差连接
x = x + self.dropout(self.attn(self.norm1(x)))
# 前馈网络+残差连接
x = x + self.dropout(self.ffn(self.norm2(x)))
return x
class GPT(nn.Module):
"""GPT模型主类（生成式预训练Transformer）"""
def __init__(self, vocab_size, embed_dim=64, num_heads=2, num_layers=2, block_size=32):
"""
参数:
vocab_size: 词汇表大小
embed_dim: 嵌入维度
num_heads: 注意力头数
num_layers: Transformer块数量
block_size: 上下文窗口大小
"""
super().__init__()
self.embed_dim = embed_dim
self.block_size = block_size # 用于生成限制
# 嵌入层：词嵌入+位置嵌入
self.token_embedding = nn.Embedding(vocab_size, embed_dim) # 词嵌入表
self.pos_embedding = nn.Embedding(block_size, embed_dim) # 位置嵌入表（编码位置信息）
# Transformer解码器块堆叠
self.layers = nn.Sequential(*[
TransformerBlock(embed_dim, num_heads, block_size) for _ in range(num_layers)
])
# 输出层：预测下一个词元
self.ln_final = nn.LayerNorm(embed_dim) # 最终层归一化
self.head = nn.Linear(embed_dim, vocab_size) # 投影到词汇表空间
def forward(self, idx, targets=None):
"""
前向传播：计算输出和损失
参数:
idx: 输入词元索引，形状(B, T)
targets: 目标词元索引（用于计算损失），形状(B, T)
返回:
logits: 预测概率对数，形状(B, T, vocab_size)
loss: 交叉熵损失（若targets不为None）
"""
B, T = idx.shape # B:批次, T:时间步
# 词嵌入 + 位置嵌入（两者相加融合语义和位置信息）
tok_emb = self.token_embedding(idx) # (B, T, C)
pos_emb = self.pos_embedding(torch.arange(T, device=idx.device)) # (T, C)
x = tok_emb + pos_emb # (B, T, C)
# 通过Transformer块
x = self.layers(x)
x = self.ln_final(x)
logits = self.head(x) # (B, T, vocab_size)→预测每个位置的下一个词元
# 计算损失（交叉熵）
if targets is None:
loss = None
else:
B, T, C = logits.shape
# 展平为(B*T, C)和(B*T)，计算交叉熵
loss = F.cross_entropy(logits.view(B*T, C), targets.view(B*T))
return logits, loss
def train_model(train_loader, test_loader, vocab_size, block_size, epochs=10):
"""
训练GPT模型
参数:
train_loader/test_loader: 训练/测试数据加载器
vocab_size: 词汇表大小
block_size: 上下文窗口大小
epochs: 训练轮数
返回:
训练好的模型
"""
# 选择设备（GPU优先）
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f"使用设备：{device}")
# 初始化模型并移动到设备
model = GPT(
vocab_size=vocab_size,
embed_dim=64, # 小维度适合演示
num_heads=2, # 注意力头数
num_layers=2, # Transformer块数
block_size=block_size
).to(device)
print(f"模型参数数量：{sum(p.numel() for p in model.parameters())/1e3:.1f}K") # 参数量统计
# 优化器：AdamW（带权重衰减的Adam，常用）
optimizer = optim.AdamW(model.parameters(), lr=3e-4)
# 训练循环
for epoch in range(epochs):
model.train() # 切换到训练模式（启用dropout等）
train_loss = 0.0
# 检查加载器是否为空
if len(train_loader) == 0:
print("训练加载器为空，跳过本轮训练")
continue
# 迭代训练数据
for x, y in tqdm(train_loader, desc=f"Epoch {epoch+1}/{epochs}"):
x, y = x.to(device), y.to(device) # 移动到设备
optimizer.zero_grad() # 清零梯度
logits, loss = model(x, y) # 前向传播
loss.backward() # 反向传播计算梯度
optimizer.step() # 更新参数
train_loss += loss.item() # 累加损失
# 测试集评估（不更新参数）
model.eval() # 切换到评估模式（禁用dropout等）
test_loss = 0.0
if len(test_loader) == 0:
print("测试加载器为空，跳过测试评估")
test_loss = float('inf')
else:
with torch.no_grad(): # 禁用梯度计算（加速+省内存）
for x, y in test_loader:
x, y = x.to(device), y.to(device)
_, loss = model(x, y)
test_loss += loss.item()
# 计算平均损失并打印
train_loss /= len(train_loader)
test_loss /= len(test_loader) if len(test_loader) > 0 else 1
print(f"Epoch {epoch+1}：训练损失 {train_loss:.4f}，测试损失 {test_loss:.4f}")
# 保存模型参数
torch.save({
'model_state_dict': model.state_dict(), # 模型参数
'vocab_size': vocab_size,
'block_size': block_size
}, 'gpt_model.pt')
print("模型已保存至 gpt_model.pt")
return model
# ---------------------------五、运用模型（文本生成）---------------------------
def generate_text(model, analyzer, subword_to_idx, prompt, max_length=50, temperature=0.7):
"""
使用训练好的模型生成文本
参数:
model: 训练好的GPT模型
analyzer: 词元分析器（用于分词）
subword_to_idx: 子词→索引映射
prompt: 提示文本（生成起点）
max_length: 生成的最大词元数
temperature: 温度参数（控制随机性，越小越确定）
返回:
生成的文本字符串
"""
model.eval() # 评估模式
device = next(model.parameters()).device # 获取模型所在设备
idx_to_subword = {v: k for k, v in subword_to_idx.items()} # 索引→子词映射
# 提示文本转换为词元索引
tokens = analyzer.tokenize(prompt)
indices = [subword_to_idx.get(t, subword_to_idx['<unk>']) for t in tokens]
idx = torch.tensor(indices, dtype=torch.long, device=device).unsqueeze(0) # 增加批次维度(B=1)
# 迭代生成新词元
for _ in range(max_length):
# 截断上下文至模型最大窗口（避免超出位置嵌入范围）
idx_cond = idx[:, -model.block_size:]
# 预测下一个词元
with torch.no_grad(): # 不计算梯度
logits, _ = model(idx_cond)
# 取最后一个时间步的预测结果
logits = logits[:, -1, :] / temperature # 温度调节（降低温度=提高确定性）
probs = F.softmax(logits, dim=-1) # 转换为概率分布
next_idx = torch.multinomial(probs, num_samples=1) # 按概率采样
idx = torch.cat([idx, next_idx], dim=1) # 拼接新索引
# 索引转换为文本
generated_indices = idx[0].tolist() # 取第一个批次
generated_tokens = [idx_to_subword[i] for i in generated_indices]
# 合并子词（去除##前缀）
text = []
for token in generated_tokens:
if token.startswith('##'):
text.append(token[2:]) # 去除##
else:
text.append(token)
return ''.join(text)
# ---------------------------主流程执行---------------------------
if __name__ == '__main__':
try:
# 1. 准备数据集
train_raw, test_raw = prepare_dataset()
# 2. 训练词元分析器
analyzer = train_analyzer(train_raw)
# 3. 预处理数据集（减小窗口和批次，适合小数据）
train_loader, test_loader, subword_to_idx, vocab_size, block_size = prepare_dataloaders(
train_raw, test_raw, analyzer, batch_size=4, block_size=16 # 小窗口适合小样本
)
# 4. 训练模型（减少轮次，加快演示）
model = train_model(train_loader, test_loader, vocab_size, block_size, epochs=5)
# 5. 运用模型生成文本
prompt = "GPT模型可以"
generated = generate_text(model, analyzer, subword_to_idx, prompt, max_length=30)
print(f"\n【文本生成结果】")
print(f"提示：{prompt}")
print(f"生成：{generated}")
except Exception as e:
print(f"执行失败：{e}") # 捕获全局异常，避免崩溃4.2、实验结果
未找到本地数据，使用示例数据...
加载本地数据：./sample_data
数据集准备完成：训练集9条，测试集1条
开始训练词元分析器...【词元分析报告】
总有效词元数：50
去重子词数：50
高频子词TOP10： [('数据预处理是', 1), ('训练过程中的', 1), ('重要步骤包括', 1), ('分词归一化等', 1), ('操作\nGPT', 1), ('是基于Tra', 1), ('nsform', 1), ('er的生成式', 1), ('预训练模型它', 1), ('通过自监督学', 1)]
高频词汇TOP10： [('数据预处理是', 1), ('训练过程中的', 1), ('重要步骤包括', 1), ('分词归一化等', 1), ('操作\nGPT', 1), ('是基于Tra', 1), ('nsform', 1), ('er的生成式', 1), ('预训练模型它', 1), ('通过自监督学', 1)]
词元分析器已保存至 wordpiece_analyzer.pkl
预处理词汇表大小：171（含特殊符号）
预处理完成：train_raw.txt 转换为 50 个词元索引
警告：test_raw.txt 文本过短（4词元），小于block_size（16）
预处理完成：test_raw.txt 转换为 20 个词元索引
数据加载器准备完成：训练集8批，测试集1批
Filter: 100%|██████████| 10/10 [00:00<00:00, 4155.66 examples/s]
使用设备：cuda
模型参数数量：123.2K
Epoch 1/5: 100%|██████████| 8/8 [00:00<00:00, 40.60it/s]
Epoch 2/5: 0%| | 0/8 [00:00<?, ?it/s]Epoch 1：训练损失 4.3993，测试损失 3.4116
Epoch 2：训练损失 2.8352，测试损失 2.1006
Epoch 2/5: 100%|██████████| 8/8 [00:00<00:00, 277.57it/s]
Epoch 3/5: 100%|██████████| 8/8 [00:00<00:00, 290.11it/s]
Epoch 4/5: 0%| | 0/8 [00:00<?, ?it/s]Epoch 3：训练损失 1.7900，测试损失 1.3070
Epoch 4/5: 100%|██████████| 8/8 [00:00<00:00, 316.72it/s]
Epoch 5/5: 0%| | 0/8 [00:00<?, ?it/s]Epoch 4：训练损失 1.1829，测试损失 0.8717
Epoch 5/5: 100%|██████████| 8/8 [00:00<00:00, 315.20it/s]
Epoch 5：训练损失 0.8360，测试损失 0.6246
模型已保存至 gpt_model.pt【文本生成结果】
提示：GPT模型可以
生成：<unk><unk>各效断<unk><unk><unk><unk>许<unk><unk>上管架心籍<unk><unk><unk><unk><unk><unk><unk><unk><unk><unk><unk><unk><unk><unk><unk>
4.3、代码“活起来”
这个 GPT 全流程代码就像 “教机器人学说话” 的完整流程，每一步都对应人类学习语言的某个环节。用通俗的话解释各部分作用如下：
一、准备数据集：给机器人找 “课本”
作用：收集供模型学习的 “原材料”（文本数据），就像给学生准备课本和练习册。
- load_local_text_data：从电脑本地文件夹里找文本文件（比如小说、文章），相当于从图书馆借书。
- generate_sample_data：如果没找到本地文件，就自己写一些简单的示例文本（比如 “GPT 是生成式模型”），相当于老师手写讲义。
- prepare_dataset：把收集到的文本分成 “训练集”（主要学习用）和 “测试集”（检验学习效果用），就像把课本分成正文和课后题。
二、训练词元分析器：给机器人编 “字典”
作用：让模型理解 “词语的组成规则”，比如 “苹果” 可以拆成 “苹” 和 “果”，方便模型记住更多词。
- WordPieceAnalyzer 类：相当于一本 “子词字典”，专门记录常用的小词片段（子词）。
- fit 方法：从文本中学习哪些子词最常见（比如 “GPT”“## 模型”），就像老师总结学生常写错的字，单独整理成表。
- tokenize 方法：用这本字典把文本拆成子词（比如 “GPT 模型”→“GPT”+“## 模型”），就像学生查字典给生字注音。
- train_analyzer：最终生成一本 “高频子词表”，告诉模型哪些子词更重要，方便后续学习。
三、预处理数据集：把 “课本” 翻译成 “机器人能懂的语言”
作用：把文本转换成模型能计算的数字，因为模型只认数字，不认文字。
- GPTDataset 类：把文本拆成固定长度的 “短句”（比如每 32 个词一段），每段的输入和下一个词作为 “问题” 和 “答案”，就像老师把课文切成短句，让学生练习 “看前半句猜后半句”。
- prepare_dataloaders：给每个子词编一个数字编号（比如 “GPT”=10，“## 模型”=23），然后把这些编号打包成批次（比如一次练 8 个短句），方便模型批量学习，就像把练习题整理成作业本。
四、训练模型：教机器人 “学规律”
作用：让模型通过大量练习，学会 “根据前文猜下一个词” 的规律，就像学生通过大量阅读和做题，学会句子的搭配规则。
- GPT 类：模型本身就像一个 “学生大脑”，由多个 “Transformer 块” 组成（类似一层层的思考步骤）。
- 词嵌入 + 位置嵌入：给每个词的编号加一个 “意义”（比如 “猫” 和 “狗” 的编号意义相近），同时记录词的位置（比如 “我吃苹果” 和 “苹果吃我” 意义不同），就像学生不仅记单词，还记单词的意思和顺序。
- 多头注意力：让模型学会 “关注重要的词”（比如 “吃” 后面更可能接 “饭” 而不是 “书”），就像学生读句子时重点看动词和名词。
- 前向传播 + 损失计算：模型先猜下一个词，再根据 “答案”（正确的下一个词）调整自己的 “思考方式”，就像学生做题后看答案纠错，不断改进。
- train_model：通过多轮练习（epochs），让模型的猜测越来越准，直到能熟练根据前文接出合理的下一个词。
五、运用模型：让机器人 “说句话试试”
作用：检验模型是否真的学会了，让它根据一个开头自己编出完整的句子，就像让学生用学到的语法自己写作文。
- generate_text：给模型一个开头（比如 “GPT 模型可以”），模型根据训练时学到的规律，一个词一个词往后接（先猜 “做”，再猜 “文本”，再猜 “生成”……），最后连成完整的句子，就像学生根据开头 “我今天” 写出 “我今天去公园玩”。
总结：整个流程就像 “教小孩学说话”
- 准备数据集 = 买绘本和故事书；
- 训练词元分析器 = 教孩子认字和拆字（比如 “森林”=“木”+“林”）；
- 预处理数据集 = 把故事切成短句，让孩子练习 “接话”；
- 训练模型 = 孩子通过大量练习，慢慢学会句子的搭配规则；
- 运用模型 = 让孩子自己编故事，检验学习效果。
每一步都是为了让模型从 “看不懂文本” 到 “能自己写文本”，核心是通过 “猜下一个词” 的练习，学会人类语言的规律。