从零开始手搓一个GPT大语言模型：从理论到实践的完整指南（一）

现在人工智能飞速发展时代，LLM绝对可以算是人工智能领域得一颗明珠，也是现在许多AI项目落地得必不可少得一个模块，可以说，不管你之前得研究领域是AI得哪个方向，现在都需要会一些LLM基础，在这个系列，文章会从最基础的数据处理开始，一步步构建属于我们自己的GPT2模型，如果之前没有了解过LLM，强烈推荐Build a Large Language Model (From Scratch)这本书，本文也是对这本书学习总结，代码图片都出自随书教学视频。

• github地址：https://github.com/rasbt/LLMs-from-scratch

项目概览

本项目将分为三个核心阶段，每个阶段都有其独特的挑战和技术要点：

🔧 Stage 1: 数据处理与模型架构构建

在这个阶段，我们将深入探讨：

• 数据预处理与采样管道：如何高效处理海量文本数据
• 注意力机制的实现：Transformer架构的核心组件
• LLM架构设计：从零搭建完整的语言模型框架

🚀Stage 2: 大模型预训练

预训练是整个项目的核心，包含：

• 训练循环设计：如何稳定训练大规模模型
• 模型评估策略：设计损失函数，实时监控训练效果
• 权重管理：预训练模型的保存与openAI预训练GPT2权重加载

🎯 Stage 3: 模型微调与应用

最终阶段将专注于实际应用：

• 分类任务微调：针对特定任务优化模型性能
• 指令数据集有监督微调(SFT)：让模型更好地理解人类指令

下载文本数据集

以一个简单的txt文本为例，接下来将进行word embedding,同时之后也会利用这个文本训练自己的LLM,

import os
import urllib.request
if not os.path.exists("the-verdict.txt"):url = ("https://raw.githubusercontent.com/rasbt/""LLMs-from-scratch/main/ch02/01_main-chapter-code/""the-verdict.txt")file_path = "the-verdict.txt"urllib.request.urlretrieve(url, file_path)

分词器

拿到一个text文本，先对其进行划分。这一过程就叫Tokenized，先得到Tokenized text,对划分的每一个块（Token），进行数字编码得到Token IDs，对于Tokenized其实简单来说就是来做文本划分，弄清楚这个就可以实现一个最简单的分词器

import re
text = "Hello, world. This, is a test."result = re.split(r'(\s)', text)
print(result)

得到结果：[‘Hello,’, ’ ', ‘world.’, ’ ', ‘This,’, ’ ', ‘is’, ’ ', ‘a’, ’ ', ‘test.’]
目前主流LLM的分词器无非是在这个的基础上做一些改进优化，比如说增加根据逗号句号等标点符号分词

text = "Hello, world. Is this-- a test?"
result = re.split(r'([,.:;?_!"()\']|--|\s)', text)
result = [item.strip() for item in result if item.strip()]
print(result)

结果：[‘Hello’, ‘,’, ‘world’, ‘.’, ‘Is’, ‘this’, ‘–’, ‘a’, ‘test’, ‘?’]
将文本分块之后就是将由token得到token IDs，把所有token放入集合中，利用集合的唯一性去重，就可以得到每一个不同的token的ids

all_words = sorted(set(preprocessed))
vocab_size = len(all_words)
print(vocab_size)

vocab = {token:integer for integer,token in enumerate(all_words)}
for i, item in enumerate(vocab.items()):print(item)if i >= 50:break

相应的，根据唯一的ids也可以将token ids解码成对应的token，总之token ids与token是一一对应的关系，由此就可以得到一个最简单的分词器SimpleTokenizerV1

class SimpleTokenizerV1:
def __init__(self, vocab):self.str_to_int = vocabself.int_to_str = {i:s for s,i in vocab.items()}
def encode(self, text):preprocessed = re.split(r'([,.:;?_!"()\']|--|\s)', text)preprocessed = [item.strip() for item in preprocessed if item.strip()]ids = [self.str_to_int[s] for s in preprocessed]return ids
def decode(self, ids):text = " ".join([self.int_to_str[i] for i in ids])# Replace spaces before the specified punctuationstext = re.sub(r'\s+([,.?!"()\'])', r'\1', text)return text

可以使用分词器将文本编码（即分词）为整数ids，这些ids随后可以被向量嵌入作为大型语言模型（LLM）的输入

tokenizer = SimpleTokenizerV1(vocab)
text = """"It's the last he painted, you know,"           Mrs. Gisburn said with pardonable pride."""
ids = tokenizer.encode(text)
print(ids)

tokenizer.decode(tokenizer.encode(text))

这一个简便的分词器有一个最明显的缺点，LLM在得到输入时不可能每个token都在训练集里面出现过，遇到没有出现在训练集里的token ids就会报错

tokenizer = SimpleTokenizerV1(vocab)
text = "Hello, do you like tea. Is this-- a test?"
tokenizer.encode(text)

如这个hallo就没在txt文本中出现过，运行代码就会报错，因此需要改进，我们可以增加特殊标记，如“<|unk|>”来表示未知单词。“<|endoftext|>”表示文本的结束，这对于训练集包含不同本文txt时这个标记是非常有必要的

all_tokens = sorted(list(set(preprocessed)))
all_tokens.extend(["<|endoftext|>", "<|unk|>"])
vocab = {token:integer for integer,token in enumerate(all_tokens)}

len(vocab.items())

for i, item in enumerate(list(vocab.items())[-5:]):print(item)

由此得到改进后的分词器：

class SimpleTokenizerV2:
def __init__(self, vocab):self.str_to_int = vocabself.int_to_str = { i:s for s,i in vocab.items()}
def encode(self, text):preprocessed = re.split(r'([,.:;?_!"()\']|--|\s)', text)preprocessed = [item.strip() for item in preprocessed if item.strip()]preprocessed = [item if item in self.str_to_int else "<|unk|>" for item in preprocessed]ids = [self.str_to_int[s] for s in preprocessed]return ids
def decode(self, ids):text = " ".join([self.int_to_str[i] for i in ids])# Replace spaces before the specified punctuationstext = re.sub(r'\s+([,.:;?!"()\'])', r'\1', text)return text

tokenizer = SimpleTokenizerV2(vocab)
text1 = "Hello, do you like tea?"
text2 = "In the sunlit terraces of the palace."
text = " <|endoftext|> ".join((text1, text2))print(text)

当然这个版本离实际GPT2使用的分词器还有些差距，比如说面对未知词时，并不是直接都将其归为“<|unk|>，
例如，如果 GPT-2 的词汇表中没有“unfamiliarword”这个词，它可能会将其分词为 [“unfam”, “iliar”, “word”] 或其他子词分解，具体取决于其训练的 BPE 合并规则。这里推荐一个在线网页可以体验不同LLM的分词策略

https://tiktokenizer.vercel.app/?model=Qwen%2FQwen2.5-72B

接下来我们使用来自 OpenAI 的开源 tiktoken 库的 BPE 分词器，该库使用 Rust 实现了其核心算法以提高计算性能。

import importlib
import tiktoken
print("tiktoken version:", importlib.metadata.version("tiktoken"))

tokenizer = tiktoken.get_encoding("gpt2")

text = ("Hello, do you like tea? <|endoftext|> In the sunlit terraces""of someunknownPlace.")integers = tokenizer.encode(text, allowed_special={"<|endoftext|>"})print(integers)

strings = tokenizer.decode(integers)
print(strings)

数据采样

有了分词器我们就可以对txt文本数据进行采样了
LLM的input和target是什么呢，其实从LLM的工作流程就可以看出一些，LLM都是根据现有文本预测下一个token,

所以LLM是通过滑动窗口对数据进行采样，一段在训练文本中连续的token作为输入，那么他的target就是滑动一定窗口后得到的相同长度的token序列，所以LLM依旧属于有监督学习的范畴，值得注意的是这里为了便于理解只滑动了一个token，但是实际LLM训练是有step可以指定的，很少有设为1的,由此就可以实现一个简单的文本数据加载函数

from torch.utils.data import Dataset, DataLoader
class GPTDatasetV1(Dataset):def __init__(self, txt, tokenizer, max_length, stride):self.input_ids = []self.target_ids = []token_ids = tokenizer.encode(txt, allowed_special={"<|endoftext|>"})assert len(token_ids) > max_length, "Number of tokenized inputs must at least be                     equal to max_length+1"   for i in range(0, len(token_ids) - max_length, stride):input_chunk = token_ids[i:i + max_length]target_chunk = token_ids[i + 1: i + max_length + 1]self.input_ids.append(torch.tensor(input_chunk))self.target_ids.append(torch.tensor(target_chunk))def __len__(self):return len(self.input_ids)def __getitem__(self, idx):return self.input_ids[idx], self.target_ids[idx]

def create_dataloader_v1(txt, batch_size=4, max_length=256, stride=128, shuffle=True, drop_last=True, num_workers=0):# Initialize the tokenizertokenizer = tiktoken.get_encoding("gpt2")# Create datasetdataset = GPTDatasetV1(txt, tokenizer, max_length, stride)# Create dataloaderdataloader = DataLoader(dataset, batch_size=batch_size, shuffle=shuffle, drop_last=drop_last, num_workers=num_workers)return dataloader

有了这个dataloader，就可以得到任意batch size的token id序列，

dataloader = create_dataloader_v1(raw_text, batch_size=8, max_length=4, stride=4, shuffle=False)data_iter = iter(dataloader)
inputs, targets = next(data_iter)
print("Inputs:\n", inputs)
print("\nTargets:\n", targets)

向量嵌入

然而，输入大型语言模型（LLM）的并不是直接的 token IDs，而是需要通过嵌入层（token embeddings，也称为词嵌入）将其转换为连续的向量表示。这些向量作为嵌入层的权重，在模型训练过程中会通过梯度下降不断更新，以优化对下一个 token 的预测能力。值得一提的是，虽然每个 token ID 对应的嵌入向量值在训练中会发生变化，但其在嵌入矩阵中的行索引（即与 token ID 的固定对应关系）始终保持不变。这种稳定的映射关系确保了模型在编码和解码过程中的一致性。

嵌入层方法本质上是一种更高效的实现方式，等价于先进行 one-hot 编码，然后通过全连接层进行矩阵乘法，由于嵌入层只是独热编码加矩阵乘法的一种更高效的实现方式，因此它可以被视为一个神经网络层，并且可以通过反向传播进行优化。

input_ids = torch.tensor([2, 3, 5, 1])

vocab_size = 6
output_dim = 3torch.manual_seed(123)
embedding_layer = torch.nn.Embedding(vocab_size, output_dim)
print(embedding_layer.weight)

位置编码：

只对不同token进行词嵌入对于LLM来说是不够的，因为Transformer本身是不具备处理序列顺利的能力的，而同一个单词位于一句话的不同位置是可以表达出不同的意思的。

所以要引入token的位置编码，为txt文本的每一个token引入一个位置编码，这里我们用的是绝对位置编码，也是个gpt2所使用的，绝对位置嵌入是指为输入序列中每个位置分配一个固定编号（0, 1, 2, …），并为每个编号对应地创建一个向量。这些向量表示每个 token 在序列中的具体位置，与 token 本身无关。所以位置编码的行数就是整个文本的长度

vocab_size = 50257 
output_dim = 256 token_embedding_layer = torch.nn.Embedding(vocab_size, output_dim)

max_length = 4 
dataloader = create_dataloader_v1( raw_text, batch_size=8, max_length=max_length, stride=max_length, shuffle=False 
) 
data_iter = iter(dataloader) 
inputs, targets = next(data_iter)

print("Token IDs:\n", inputs) 
print("\nInputs shape:\n", inputs.shape)

token_embeddings = token_embedding_layer(inputs) 
print(token_embeddings.shape)

context_length = max_length 
pos_embedding_layer = torch.nn.Embedding(context_length, output_dim)
pos_embeddings = pos_embedding_layer(torch.arange(max_length)) 
print(pos_embeddings.shape)

为了创建在大语言模型（LLM）中使用的输入嵌入，我们只需将词元嵌入和位置嵌入相加。

input_embeddings = token_embeddings + pos_embeddings
print(input_embeddings.shape)

总结

上图总结了一个完整的文本数据的处理过程：
文本经过分词器划分为token，得到token id后经行向量嵌入，这一步可以模型得以进行loss的反向传播，每一个词向量还需要加上与之对应的位置编码，这是为了提升LLM的顺序序列处理能力，得到的input embedings就可以输入
LLM进行训练了，关于位置编码，目前的改进很多，绝对位置编码的应用少了很多，比如千文3使用的RoPE（Rotary Position Embedding），也是目前大语言模型中常用的一种位置编码方式
和相对位置编码相比，RoPE 具有更好的外推性，目前是大模型相对位置编码中应用最广的方式之一：

其原理用直观的话来说就是将位置编码看作一个二维旋转角度，让QK的点乘运算本身隐含顺序差异

因为旋转可以表示相对位置，所以天然支持相对位置感知

备注：什么是大模型外推性？
外推性是指大模型在训练时和预测时的输入长度不一致，导致模型的泛化能力下降的问题。例如，如果一个模型在训练时一个batch只使用了512个 token
的文本，那么在预测时如果输入超过512个 token，模型可能无法正确处理。这就限制了大模型在处理长文本或多轮对话等任务时的效果。

详细可以参考博客：https://www.zhihu.com/tardis/bd/art/647109286