通过一个应用程序，以视觉向导的形式介绍GPT架构
无可否认，聊天机器人已经改变了我们与数字平台的互动方式。尽管底层语言模型在处理复杂任务方面取得了令人印象深刻的进展，用户体验往往不尽人意，感觉冷漠和疏离。
为了使对话更具关联性，我设想了一个可以模拟我随意的写作风格的聊天机器人，类似于在WhatsApp上给朋友发短信。
在这篇文章中，我将带您走过我构建一个（小）语言模型的旅程，该模型使用我的WhatsApp聊天消息作为输入数据来生成合成对话。在此过程中，我尝试以视觉和易于理解的方式解开GPT架构的内部工作原理，并提供实际的Python实现。您可以在我的GitHub上找到完整的项目。
注意：模型类本身的大部分代码取自Andrej Karpathy的视频系列，并根据我的需要进行了调整。我强烈推荐他的教程。
lad-gpt
从头开始训练一个完全基于您的WhatsApp群聊的语言模型。
github.com
目录
- 选定的方法
- 数据来源
- 分词
- 索引
- 模型架构
- 模型训练
- 聊天模式
1. 选定的方法
当调整语言模型以适应特定数据语料库时，有几种方法可以选择：
- 模型构建：这涉及从头构建和训练模型，提供了在模型架构和训练数据选择方面最大的灵活性。
- 微调：这种方法利用现有的预训练模型，调整其权重以更好地与手头的特定数据相吻合。
- 提示工程：这也利用现有的预训练模型，但在这种情况下，将唯一的语料库直接融入提示中，而不改变模型的权重。
由于我这个项目的动机主要是自我教育，而我对现代语言模型的架构更感兴趣，所以我选择了第一种方法。然而，这个选择显然存在限制。考虑到我的数据规模和可用的计算资源，我没有期望能够达到与任何最先进的预训练模型相媲美的结果。
尽管如此，我还是希望我的模型能够捕捉到一些有趣的语言模式，它最终确实做到了。探索第二个选项（微调）可能是未来一篇文章的重点。
2. 数据来源
WhatsApp是我主要的沟通渠道，非常适合捕捉我的对话风格。导出六年多的群聊历史记录，总计超过150万个词非常简单。
使用正则表达式模式将数据解析为包含日期、联系人姓名和聊天消息的元组列表。
pattern = r'\[(.*?)\] (.*?): (.*)'matches = re.findall(pattern, text)text = [(x1, x2.lower()) for x0, x1, x2 in matches]
[ (2018-03-12 16:03:59, "Alice", "嗨，你们好吗？"), (2018-03-12 16:05:36, "Tom", "谢谢，我很好！"), ...]现在，每个元素都被单独处理。
- 发送日期：除了将其转换为日期时间对象外，我没有利用这个信息。然而，可以查看时间差来区分话题讨论的开始和结束。
- 联系人姓名：在对文本进行标记化时，每个联系人姓名都被视为一个唯一的标记。这确保了名字和姓氏的组合仍被视为单个实体。
- 聊天消息：每条消息的末尾添加一个特殊的”<END>”标记。
3. 分词
为了训练语言模型，我们需要将语言分解成碎片（所谓的标记），并逐步将它们提供给模型。分词可以在多个级别上进行。
- 字符级：文本被视为单个字符的序列（包括空格）。这种粒度的方法可以从字符序列中形成任何可能的单词。然而，更难捕捉单词之间的语义关系。
- 词级：文本被表示为单词的序列。但是，模型的词汇表受到训练数据中现有单词的限制。
- 子词级：文本被分解为比单词更小但比字符大的子词单位。
虽然我最初使用了字符级分词器，但我觉得训练时间被浪费在学习重复单词的字符序列上，而不是关注句子之间单词的语义关系。
出于概念上的简单性考虑，我决定切换到词级分词器，暂时不使用可用的库进行更复杂的分词策略。
from nltk.tokenize import RegexpTokenizerdef custom_tokenizer(txt: str, spec_tokens: List[str], pattern: str="|\d|\\w+|[^\\s]") -> List[str]: """ 使用NLTK的正则表达式标记器将文本标记化为单词或字符，并将给定的特殊组合视为单个标记。 :param txt: 作为单个字符串元素的语料库。 :param spec_tokens: 一组特殊标记（例如结束符，未登录词）。 :param pattern: 默认情况下，语料库进行单词级的标记化（按空格分割）。 数字被视为单个标记。 >> 注意：字符级标记化的模式是“|.” """ pattern = "|".join(spec_tokens) + pattern tokenizer = RegexpTokenizer(pattern) tokens = tokenizer.tokenize(txt) return tokens
["Alice:", "Hi", "how", "are", "you", "guys", "?", "<END>", "Tom:", ... ]结果显示，我的训练数据有大约70,000个唯一单词的词汇量。然而，由于很多单词只出现一次或两次，我决定将这样的罕见词替换为特殊的”<UNK>”标记。这样可以将词汇表减少到大约25,000个单词，从而需要训练一个更小的模型。
from collections import Counterdef get_infrequent_tokens(tokens: Union[List[str], str], min_count: int) -> List[str]: """ 识别出现次数小于最小计数阈值的标记。 :param tokens: 如果是原始文本字符串，则在字符级别计算频率。 如果是以标记为单位的分词语料库列表，则在标记级别计算频率。 :param min_count: 用于标记的出现次数阈值。 :return: 出现频率低的标记列表。 """ counts = Counter(tokens) infreq_tokens = set([k for k,v in counts.items() if v<=min_count]) return infreq_tokensdef mask_tokens(tokens: List[str], mask: Set[str]) -> List[str]: """ 遍历所有标记。将属于集合中的任何标记替换为未知标记。 :param tokens: 分词的语料库。 :param mask: 需要在语料库中屏蔽的标记集合。 :return: 屏蔽操作后的分词语料库列表。 """ return [t.replace(t, unknown_token) if t in mask else t for t in tokens]infreq_tokens = get_infrequent_tokens(tokens, min_count=2)tokens = mask_tokens(tokens, infreq_tokens)
["Alice:", "Hi", "how", "are", "you", "<UNK>", "?", "<END>", "Tom:", ... ]4. 索引化
在进行分词后，下一步是将单词和特殊标记转换为数值表示。使用固定的词汇表，每个单词根据其位置进行索引。然后将编码的单词准备为PyTorch张量。
import torchdef encode(s: list, vocab: list) -> torch.tensor: """ 将标记列表编码为整数张量，给定固定的词汇表。 当在词汇表中找不到标记时，将分配特殊的未知标记。 当训练集没有使用该特殊标记时，将分配一个随机标记。 """ rand_token = random.randint(0, len(vocab)) map = {s:i for i,s in enumerate(vocab)} enc = [map.get(c, map.get(unknown_token, rand_token)) for c in s] enc = torch.tensor(enc, dtype=torch.long) return enc
torch.tensor([8127, 115, 2363, 3, ..., 14028])因为我们需要评估我们的模型对一些未见数据的质量，所以我们将张量分成两部分。然后，我们就有了训练集和验证集，可以直接提供给语言模型进行训练。
5. 模型架构
我决定采用GPT架构，这是一篇有影响力的论文“Attention is All you Need”中推广的架构。因为我试图构建一个语言生成器，而不是一个问答机器人，所以仅有解码器（右侧）的架构对此目的已足够。
在接下来的几个部分中，我将详细解释GPT架构的每个组件，解释其作用和基础矩阵操作。从准备好的训练集开始，我将追踪一个由3个单词组成的示例上下文，直到它导致对下一个标记的预测。
5.1. 模型目标
在深入技术细节之前，理解我们模型的主要目标是至关重要的。在仅有解码器的设置中，我们的目标是解码语言的结构，以准确预测序列中下一个标记，给定先前标记的上下文。
当我们将索引化的标记序列输入模型时，它将通过一系列与不同权重矩阵的矩阵乘法进行计算。输出是一个向量，表示每个标记是序列中下一个标记的概率，基于输入的上下文。
模型评估：
我们模型的性能是根据训练数据进行评估的，其中实际的下一个标记是已知的。目标是最大化正确预测这个下一个标记的概率。
然而，在机器学习中，我们经常关注“损失”这个概念，它量化了错误或不正确预测的可能性。为了计算这个损失，我们将模型的输出概率与实际的下一个标记进行比较（使用交叉熵）。
优化：
通过反向传播，我们希望最小化当前的损失。该过程涉及将令牌序列迭代地输入模型，并调整权重矩阵以提高性能。
在每个图中，我将用黄色突出显示在该过程中将进行优化的权重矩阵。
5.2. 输出嵌入
到目前为止，我们的序列中的每个标记都由一个整数索引表示。然而，这种简单的形式并不能反映单词之间的关系或相似性。为了解决这个问题，我们通过嵌入将我们的一维索引提升到更高维度的空间。
- 词嵌入：一个单词的本质可以通过一个n维浮点数向量来表达。
- 位置嵌入：这些嵌入突出了一个单词在句子中的重要性，也表示为n维浮点数向量。
对于每个标记，我们查找它的词嵌入和位置嵌入，然后逐元素求和。这样就得到了上下文中每个标记的输出嵌入。
在下面的示例中，上下文由3个标记组成。在嵌入过程结束时，每个标记都由一个n维向量表示（其中n是嵌入大小，一个可调节的超参数）。
PyTorch提供了专门的类来实现这样的嵌入。在我们的模型类中，我们按照以下方式定义词嵌入和位置嵌入（通过参数传递矩阵维度）：
self.word_embedding = nn.Embedding(vocab_size, embed_size)self.pos_embedding = nn.Embedding(block_size, embed_size)5.3. 自注意力头部
虽然词嵌入提供了单词相似性的一般概念，但一个单词的真正含义往往取决于其周围的上下文。例如，“bat”在句子中可以指动物或者体育设备，这取决于上下文。这就是GPT架构的一个关键组成部分——自注意力机制的作用。
自注意力机制主要关注三个核心概念：查询（Q）、键（K）和值（V）。
- 查询（Q）：查询是当前标记的表示，需要计算注意力的标记。可以理解为：“作为当前标记，我应该在剩下的上下文中关注什么？”
- 键（K）：键是输入序列中每个标记的表示。它们与查询配对，用于确定注意力分数。这个比较衡量了查询标记在上下文中应该关注其他标记的程度。较高的分数意味着应该更加关注。
- 值（V）：值也是输入序列中每个标记的表示。然而，它们的作用不同，它们对注意力分数应用最终的加权。
示例：
在我们的示例中，上下文的每个标记已经是嵌入形式，作为n维向量（e1, e2, e3）。自注意力头部将它们作为输入，为它们中的每一个输出一个上下文化版本。
- 在评估标记“name”时，通过将其嵌入向量v2与可训练矩阵M_Q相乘，获得查询向量q。
- 同时，通过将每个嵌入向量（e1, e2, e3）与可训练矩阵M_K相乘，计算出每个上下文中标记的键向量（k1, k2, k3）。
- 通过以相同的方式获取值向量（v1, v2, v3），只是用另一个可训练矩阵M_V进行乘法。
- 注意力分数w被计算为查询向量和每个键向量之间的点积。
- 最后，我们将所有值向量堆叠成一个矩阵，并将其乘以注意力分数，以获得标记“name”的上下文化向量。
class Head(nn.Module): """ 该模块在输入张量上执行自注意力操作，产生具有相同时间步长但不同通道的输出张量。 :param head_size: 多头注意力机制中头部的大小。 """ def __init__(self, head_size): super().__init__() self.key = nn.Linear(embed_size, head_size, bias=False) self.query = nn.Linear(embed_size, head_size, bias=False) self.value = nn.Linear(embed_size, head_size, bias=False) def forward(self, x): """ # 输入大小为(批次大小，时间步长，通道数) # 输出大小为(批次大小，时间步长，头部大小) """ B,T,C = x.shape k = self.key(x) q = self.query(x) # 计算注意力分数 wei = q @ k.transpose(-2,-1) wei /= math.sqrt(k.shape[-1]) # 避免前瞻 tril = torch.tril(torch.ones(T, T)) wei = wei.masked_fill(tril == 0, float("-inf")) wei = F.softmax(wei, dim=-1) # 值的加权聚合 v = self.value(x) out = wei @ v return out5.4. 带屏蔽的多头注意力
语言是复杂的，抓住其所有细微差别并不直接。一个单一集合的注意力计算通常不足以捕捉单词如何共同工作的微妙之处。这就是GPT模型中多头注意力的好处所在。
将多头注意力视为几双从不同角度观察数据的眼睛，每个眼睛都会注意到独特的细节。然后将这些独立的观察结果合并为一个完整的画面。为了使这个完整的画面易于管理并与我们模型的其余部分兼容，我们使用一个线性层（可训练权重）将其压缩回到我们的原始嵌入大小。
最后，为了确保我们的模型不仅仅记住了训练数据，而且在新文本上做出了良好的预测，我们使用了一个丢弃层。在训练过程中，这会随机关闭部分数据，从而帮助模型变得更具适应性。
class MultiHeadAttention(nn.Module): """ 此类包含多个`Head`对象，它们并行执行自注意力操作。 """ def __init__(self): super().__init__() head_size = embed_size // n_heads heads_list = [Head(head_size) for _ in range(n_heads)] self.heads = nn.ModuleList(heads_list) self.linear = nn.Linear(n_heads * head_size, embed_size) self.dropout = nn.Dropout(dropout) def forward(self, x): heads_list = [h(x) for h in self.heads] out = torch.cat(heads_list, dim=-1) out = self.linear(out) out = self.dropout(out) return out5.5. 前馈神经网络
多头注意力层最初捕获序列内的上下文关系。通过两个连续的线性层，向网络添加更多深度，这些层构成了前馈神经网络。
在初始线性层中，我们增加了维度（在我们的情况下增加了4倍），从而有效扩大了网络学习和表示更复杂特征的能力。在结果矩阵的每个元素上应用ReLU函数，这使非线性模式能够被识别。
随后，第二个线性层作为一个压缩器，将扩展维度减小回原始形状（块大小 x 嵌入大小）。最后，一个丢弃层结束整个过程，随机停用矩阵的元素，以达到模型泛化的目的。
“`html
class FeedFoward(nn.Module): """ This module passes the input tensor through a series of linear transformations and non-linear activations. """ def __init__(self): super().__init__() self.net = nn.Sequential( nn.Linear(embed_size, 4 * embed_size), nn.ReLU(), nn.Linear(4 * embed_size, embed_size), nn.Dropout(dropout), ) def forward(self, x): return self.net(x)5.6. Add & Norm
现在我们通过引入两个关键要素，将多头注意力和前馈组件连接起来：
- 残差连接（Add）：这些连接对一个层的输出与其未改变的输入进行逐元素加法。在训练过程中，模型根据它们的有用性调整对层变换的重视。如果一个变换被认为是非必要的，它的权重以及其层的输出会趋向于零。在这种情况下，至少会将未改变的输入传递给残差连接。这种技术有助于缓解梯度消失问题。
- 层归一化（Norm）：该方法通过减去每个上下文中嵌入向量的均值并除以其标准差来对每个嵌入向量进行归一化。这个过程还确保反向传播中的梯度既不爆炸也不消失。
多头注意力和前馈层的链条，与”Add & Norm”连接在一起，形成一个模块化的设计。这种设计允许我们形成一个块的序列。这些块的数量是一个超参数，它决定了模型架构的深度。
class Block(nn.Module): """ This module contains a single transformer block, which consists of multi-head self-attention followed by feed-forward neural networks. """ def __init__(self): super().__init__() self.sa = MultiHeadAttention() self.ffwd = FeedFoward() self.ln1 = nn.LayerNorm(embed_size) self.ln2 = nn.LayerNorm(embed_size) def forward(self, x): x = x + self.sa(self.ln1(x)) x = x + self.ffwd(self.ln2(x)) return x5.7. Softmax
通过遍历多个块组件，我们得到一个维度为（块尺寸 x 嵌入尺寸）的矩阵。为了将该矩阵重新形成所需的维度（块尺寸 x 词汇大小），我们将其通过最后一个线性层。这个形状表示上下文中每个位置上词汇表中的每个单词的条目。
最后，我们对这些值应用软最大化变换，将它们转换为概率。我们已经成功地获得了上下文中每个位置上下一个标记的概率分布。
6. 模型训练
为了训练语言模型，我从我的训练数据中选择了随机位置的令牌序列。鉴于 WhatsApp 对话的快节奏特性，我确定了一个长度为 32 个词的上下文长度是足够的。因此，我选择了随机的 32 个词的块作为上下文输入，并将相应的向量向后移动一个词来作为比较的目标。
训练过程循环执行以下步骤：
- 样本多个上下文的批次。
- 将这些样本输入模型以计算当前损失。
- 根据当前损失和模型权重进行反向传播。
- 每 500 次迭代更全面地评估损失。
一旦确定了其他模型超参数（如嵌入尺寸、自注意头数等），我最终选择了一个具有 250 万个参数的模型。考虑到我对输入数据大小和计算资源的限制，我认为这是我最理想的设置。
整个训练过程大约需要 12 小时进行 10,000 次迭代。从损失在验证集和训练集之间的差异变大可以观察到，在此之前训练可以停止。
“`
import json
import torch
from config import eval_interval, learn_rate, max_iters
from src.model import GPTLanguageModel
from src.utils import current_time, estimate_loss, get_batch
def model_training(update: bool) -> None:
"""
使用预加载数据训练或更新GPTLanguageModel。
这个函数根据`update`参数初始化一个新的模型或加载现有的模型。
然后，使用AdamW优化器在训练和验证数据集上训练模型。
最后，保存训练好的模型。
:param update: 布尔标志，指示是否更新现有模型。
"""
# 加载数据
train_data = torch.load("assets/output/train.pt")
valid_data = torch.load("assets/output/valid.pt")
with open("assets/output/vocab.txt", "r", encoding="utf-8") as f:
vocab = json.loads(f.read())
# 初始化/加载模型
if update:
try:
model = torch.load("assets/models/model.pt")
print("加载现有模型以继续训练。")
except FileNotFoundError:
print("找不到现有模型。初始化一个新模型。")
model = GPTLanguageModel(vocab_size=len(vocab))
else:
print("初始化一个新模型。")
model = GPTLanguageModel(vocab_size=len(vocab))
# 初始化优化器
optimizer = torch.optim.AdamW(model.parameters(), lr=learn_rate)
# 模型参数数量
n_params = sum(p.numel() for p in model.parameters())
print(f"待优化的参数数目：{n_params}\n")
# 模型训练
for i in range(max_iters):
# 每'eval_interval'步骤评估训练和验证集的损失
if i % eval_interval == 0 or i == max_iters - 1:
train_loss = estimate_loss(model, train_data)
valid_loss = estimate_loss(model, valid_data)
time = current_time()
print(f"{time} | 第{i}步：训练损失 {train_loss:.4f}，验证损失 {valid_loss:.4f}")
# 获取一批数据
x_batch, y_batch = get_batch(train_data)
# 评估损失
logits, loss = model(x_batch, y_batch)
optimizer.zero_grad(set_to_none=True)
loss.backward()
optimizer.step()
torch.save(model, "assets/models/model.pt")
print("模型已保存")
def conversation() -> None:
"""
通过从预训练的GPTLanguageModel中进行采样来模拟聊天对话。
这个函数加载一个经过训练的GPTLanguageModel，以及词汇表和特殊标记列表。
然后，它进入一个循环，用户指定一个联系人名称。
给定这个输入，模型生成一个样本响应。
对话将持续到用户输入结束标记。
"""
with open("assets/output/vocab.txt", "r", encoding="utf-8") as f:
vocab = json.loads(f.read())
with open("assets/output/contacts.txt", "r", encoding="utf-8") as f:
contacts = json.loads(f.read())
spec_tokens = contacts + [end_token]
model = torch.load("assets/models/model.pt")
completer = WordCompleter(spec_tokens, ignore_case=True)
input = prompt("消息 >> ", completer=completer, default="")
output = torch.tensor([], dtype=torch.long)
print()
while input != end_token:
for _ in range(n_chats):
add_tokens = custom_tokenizer(input, spec_tokens)
add_context = encode(add_tokens, vocab)
context = torch.cat((output, add_context)).unsqueeze(1).T
n0 = len(output)
output = model.generate(context, vocab)
n1 = len(output)
print_delayed(decode(output[n0-n1:], vocab))
input = random.choice(contacts)
input = prompt("\n回复 >> ", completer=completer, default="")
print()
7. 聊天模式
为了与训练好的模型进行交互，我创建了一个函数，通过下拉菜单选择联系人名称，并输入一条消息供模型回复。
参数“n_chats”确定模型同时生成的回复数量。
当模型预测到下一个标记为<END>时，模型会结束生成的消息。
import json
import random
import torch
from prompt_toolkit import prompt
from prompt_toolkit.completion import WordCompleter
from config import end_token, n_chats
from src.utils import custom_tokenizer, decode, encode, print_delayed
def conversation() -> None:
"""
通过从预训练的GPTLanguageModel中进行采样来模拟聊天对话。
这个函数加载一个经过训练的GPTLanguageModel，以及词汇表和特殊标记列表。
然后，它进入一个循环，用户指定一个联系人名称。
给定这个输入，模型生成一个样本响应。
对话将持续到用户输入结束标记。
"""
with open("assets/output/vocab.txt", "r", encoding="utf-8") as f:
vocab = json.loads(f.read())
with open("assets/output/contacts.txt", "r", encoding="utf-8") as f:
contacts = json.loads(f.read())
spec_tokens = contacts + [end_token]
model = torch.load("assets/models/model.pt")
completer = WordCompleter(spec_tokens, ignore_case=True)
input = prompt("消息 >> ", completer=completer, default="")
output = torch.tensor([], dtype=torch.long)
print()
while input != end_token:
for _ in range(n_chats):
add_tokens = custom_tokenizer(input, spec_tokens)
add_context = encode(add_tokens, vocab)
context = torch.cat((output, add_context)).unsqueeze(1).T
n0 = len(output)
output = model.generate(context, vocab)
n1 = len(output)
print_delayed(decode(output[n0-n1:], vocab))
input = random.choice(contacts)
input = prompt("\n回复 >> ", completer=completer, default="")
print()
结论：
由于我的个人聊天的隐私，我无法在这里呈现例子和对话。
尽管如此，您可以期望这样的模型成功地学习句子的一般结构，从而产生有意义的输出。在我的案例中，它还捕捉到了训练数据中突出的某些主题的背景信息。例如，由于我的个人聊天经常围绕网球展开，网球运动员的姓名和与网球相关的词汇通常会一起输出。
然而，当评估生成句子的连贯性时，我承认结果并没有达到我已经很谨慎的期望。当然，我也可以归咎于我的朋友们太多地聊些废话，限制了模型学习有用知识的能力…
为了至少展示一些末尾的例子输出，您可以看看这个虚拟模型在200个训练过的虚拟消息上的表现 😉
