3-大语言模型—理论基础：生成式预训练语言模型GPT(代码“活起来”)

目录

1、GPT的模型结构如图所示

2、介绍GPT自监督预训练、有监督下游任务微调及预训练语言模型

2.1、GPT 自监督预训练

2.1.1、 输入编码：词向量与位置向量的融合

2.1.1.1、 输入序列与词表映射

2.1.1.2、 词向量矩阵与查表操作

3. 位置向量矩阵

4. 词向量与位置向量叠加

5. 最终输入向量

2.1.2、 Masked 多头注意力：禁止 “偷看” 未来信息

2.1.3、 损失函数：优化预测概率

2.2 有监督下游任务微调

2.2.1、 任务适配：从文本到标签的映射

2.2.2、组合损失：平衡任务与预训练知识

2.3 预训练语言模型

2.3.1、 结构差异：从 “专用设计” 到 “通用基座”

2.3.2、 能力边界：生成 vs 理解

3、模型验证

3.1、 GPT 模型全流程

3.2、GPT 核心逻辑突出：

4、完整实现

4.1、完整代码

4.2、实验结果 

4.3、代码“活起来”

一、准备数据集：给机器人找 “课本”

二、训练词元分析器：给机器人编 “字典”

三、预处理数据集：把 “课本” 翻译成 “机器人能懂的语言”

四、训练模型：教机器人 “学规律”

五、运用模型：让机器人 “说句话试试”

总结：整个流程就像 “教小孩学说话”

 

---

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、 词向量矩阵与查表操作

词向量矩阵的作用是将离散的词索引映射为连续的向量表示。假设词向量维度为 5，则是一个 3×5 的矩阵：

查表过程：

• 对于 token "猫"（索引 0），其词向量为的第 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：残差连接与层归一化 注意力输出与原始输入相加（残差连接），再经层归一化（使每层输入分布稳定），得到 Transformer Block 的输出。

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 模型全流程

1. 准备数据集：获取原始文本并划分训练 / 测试集；
2. 训练词元分析器：分析词汇特征，构建子词词汇表；
3. 预处理数据集：将文本转换为模型可接受的整数序列；
4. 训练模型：基于 Transformer 架构训练自回归语言模型；
5. 运用模型：通过提示文本生成连贯的新内容。

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_dirdef 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 = 0while 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 = Truebreakif 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 resultdef 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: 训练好的WordPieceAnalyzersubword_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)  # 确保≥0def __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)  # 一次性计算QKVself.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防止过拟合# 注意力加权求和：概率×Vout = 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 xclass 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_dimself.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 = Noneelse: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, lossdef 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.0if 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 1print(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 模型可以”），模型根据训练时学到的规律，一个词一个词往后接（先猜 “做”，再猜 “文本”，再猜 “生成”……），最后连成完整的句子，就像学生根据开头 “我今天” 写出 “我今天去公园玩”。

总结：整个流程就像 “教小孩学说话”

1. 准备数据集 = 买绘本和故事书；
2. 训练词元分析器 = 教孩子认字和拆字（比如 “森林”=“木”+“林”）；
3. 预处理数据集 = 把故事切成短句，让孩子练习 “接话”；
4. 训练模型 = 孩子通过大量练习，慢慢学会句子的搭配规则；
5. 运用模型 = 让孩子自己编故事，检验学习效果。

每一步都是为了让模型从 “看不懂文本” 到 “能自己写文本”，核心是通过 “猜下一个词” 的练习，学会人类语言的规律。