聊天机器人无疑改变了我们与数字平台的互动方式。尽管底层语言模型处理复杂任务的能力有了令人印象深刻的进步，但用户体验往往不尽人意，感觉缺乏个性和脱节。

为了让对话更贴切，我设想了一个能够模仿我随意写作风格的聊天机器人，就像通过WhatsApp给朋友发短信一样。

在本文中，我将引导你了解我构建一个（小型）语言模型的旅程，该模型使用我的WhatsApp聊天消息作为输入数据来生成合成对话。在这个过程中，我试图以一种直观且希望容易理解的方式来阐明GPT架构的内部工作原理，并通过实际的Python实现来补充说明。

1. 选择的方法

在针对特定语料库调整语言模型时，可以采取几种不同的方法：

1. 模型构建：这涉及从零开始构建和训练模型，提供最大的灵活性，可以自由选择模型架构和训练数据。
2. 微调：这种方法利用现有的预训练模型，调整其权重以更紧密地与手头的特定数据对齐。
3. 提示工程：这也使用现有的预训练模型，但在这里，将独特的语料库直接整合到提示中，而不改变模型的权重。

鉴于我的项目主要是出于自我教育的目的，我对现代语言模型的架构相当感兴趣，因此我选择了第一种方法。

2. 数据来源

WhatsApp作为我主要的通信渠道，是捕捉我谈话风格的理想来源。导出超过六年的群聊历史记录，总计超过150万个单词，过程相当直接。

数据使用正则表达式模式解析为一个包含日期、联系人姓名和聊天消息的元组列表。

pattern = r'\[(.*?)\] (.*?): (.*)'r'\[(.*?)\] (.*?): (.*)'
matches = re.findall(pattern, text)
text = [(x1, x2.lower()) for x0, x1, x2 in matches]

[
    (2018-03-12 16:03:59, "Alice", "Hi, how are you guys?"),
    (2018-03-12 16:05:36, "Tom", "I am good thanks!"),
    ...
]

现在，每个元素都被分别处理。

• 发送日期：除了将其转换为datetime对象之外，我还没有利用这些信息。然而，人们可以通过查看时间差来区分讨论话题的开始和结束。
• 联系人姓名：在对文本进行分词时，每个联系人姓名都被当作一个独特的令牌。这确保了名字和姓氏的组合仍然被视为单一实体。
• 聊天信息：在每条信息的末尾都添加了一个特殊的“<END>”令牌。

3. 分词

为了训练一个语言模型，我们需要将语言拆分成片段（所谓的令牌），并逐渐地将它们输入到模型中。分词可以在多个层次上进行。

• 字符级别：文本被视为一系列单独的字符（包括空白）。这种细粒度的方法允许从字符序列中形成每一个可能的单词。然而，抓住单词之间的语义关系更加困难。
• 词级别：文本被表示为一系列的词。然而，模型的词汇表受限于训练数据中的现有词汇。
• 子词级别：把文本拆分成比词小但比字符大的子词单元。

虽然我开始使用的是字符级分词器，但我感觉训练时间被浪费在了学习重复单词的字符序列上，而不是关注句子中单词之间的语义关系。

为了概念上的简单，我决定转换到使用词级分词器，暂时不考虑使用更复杂的分词策略的现有库。

from nltk.tokenize import RegexpTokenizer
def custom_tokenizer(txt: str, spec_tokens: List[str], pattern: str="|\d|\\w+|[^\\s]") -> List[str]:
    """
    Tokenize text into words or characters using NLTK's RegexpTokenizer, considerung 
    given special combinations as single tokens.
    :param txt: The corpus as a single string element.
    :param spec_tokens: A list of special tokens (e.g. ending, out-of-vocab).
    :param pattern: By default the corpus is tokenized on a word level (split by spaces).
                    Numbers are considered single tokens.
    >> note: The pattern for character level tokenization is '|.'
    """
    pattern = "|".join(spec_tokens) + pattern
    tokenizer = RegexpTokenizer(pattern)
    tokens = tokenizer.tokenize(txt)
    return tokens
["Alice:", "Hi", "how", "are", "you", "guys", "?", "<END>", "Tom:", ... ]

事实证明，我的训练数据包含大约70,000个独特的单词。然而，由于很多单词只出现一次或两次，我决定用一个特殊标记“ ”来替换这些罕见的单词。这样做的结果是词汇量减少到大约25,000个单词，这导致了之后需要训练的模型规模变小。

from collections import Counter
def get_infrequent_tokens(tokens: Union[List[str], str], min_count: int) -> List[str]:
    """
    Identify tokens that appear less than a minimum count.
    
    :param tokens: When it is the raw text in a string, frequencies are counted on character level.
                   When it is the tokenized corpus as list, frequencies are counted on token level.
    :min_count: Threshold of occurence to flag a token.
    :return: List of tokens that appear infrequently. 
    """
    counts = Counter(tokens)
    infreq_tokens = set([k for k,v in counts.items() if v<=min_count])
    return infreq_tokens
def mask_tokens(tokens: List[str], mask: Set[str]) -> List[str]:
    """
    Iterate through all tokens. Any token that is part of the set, is replaced by the unknown token.
    :param tokens: The tokenized corpus.
    :param mask: Set of tokens that shall be masked in the corpus.
    :return: List of tokenized corpus after the masking operation.
    """
    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 torch
def encode(s: list, vocab: list) -> torch.tensor:
    """
    Encode a list of tokens into a tensor of integers, given a fixed vocabulary. 
    When a token is not found in the vocabulary, the special unknown token is assigned. 
    When the training set did not use that special token, a random token is assigned.
    """
    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）。

1. 查询 (Q)：查询本质上是当前token为了计算注意力所需要的一个表示。就像是在问，“作为当前token，我应该在其他上下文中注意什么？”
2. 键 (K)：键是输入序列中每个token的表示。它们与查询配对来决定注意力分数。这种比较衡量了查询token应该在上下文中其它token上投入多少注意力。高分数意味着应该投入更多的注意力。
3. 值 (V)：值也是输入序列中每个token的表示。然而，它们的作用不同，因为它们对注意力分数应用最终加权。

例子：

在我们的例子中，上下文中的每一个标记(token)已经以嵌入形式存在，即n维向量（e1, e2, e3）。自注意力头(self-attention head)将它们作为输入，一次一个地输出它们的上下文化版本。

1. 在评估标记“name”时，通过将其嵌入向量v2与可训练矩阵M_Q相乘，获得一个查询向量q。
2. 同时，通过将每个嵌入向量（e1, e2, e3）与可训练矩阵M_K相乘，为上下文中的每个标记计算键向量(key vectors)（k1, k2, k3）。
3. 值向量(value vectors)（v1, v2, v3）也是以同样的方式获得的，只是乘以了不同的可训练矩阵M_V。
4. 注意力得分(attention-scores) w 是通过查询向量和每个键向量之间的点乘得出的。
5. 最后，我们将所有值向量堆叠成一个矩阵，并且用注意力得分乘以该矩阵，以获得标记“name”的上下文化向量。

class Head(nn.Module):
    """
    This module performs self-attention operations on the input tensor, producing 
    an output tensor with the same time-steps but different channels. 
    
    :param head_size: The size of the head in the multi-head attention mechanism.
    """
    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):
        """
        # input of size (batch, time-step, channels)
        # output of size (batch, time-step, head size)
        """
        B,T,C = x.shape
        k = self.key(x)                                     
        q = self.query(x)                                   
        # compute attention scores
        wei = q @ k.transpose(-2,-1)                        
        wei /= math.sqrt(k.shape[-1])                       
        
        # avoid look-ahead
        tril = torch.tril(torch.ones(T, T))
        wei = wei.masked_fill(tril == 0, float("-inf"))    
        wei = F.softmax(wei, dim=-1)
        
        # weighted aggregation of the values
        v = self.value(x)    
        out = wei @ v
        return out

5.4. 掩码多头注意力

语言复杂多变，要捕捉其全部细微差别绝非易事。单一的注意力计算往往不足以捕获词语合作时的微妙之处。这就是GPT模型中多头注意力概念派上用场的地方。

想象一下，多头注意力就像有好几对眼睛以不同的方式观察数据，每一对都能发现独特的细节。然后将这些分开的观察结果汇集成一个大画面。为了保持这个大画面的可管理性，并且与我们模型的其它部分兼容，我们使用线性层（可训练的权重）将其压缩回我们原始的嵌入大小。

最后，为了确保我们的模型不仅仅是记住训练数据，而且也能在新文本上做出好的预测，我们使用了一个丢弃层。这会在训练期间随机关闭数据的某些部分，这有助于模型变得更加适应性强。

class MultiHeadAttention(nn.Module):
    """
    This class contains multiple `Head` objects, which perform self-attention 
    operations in parallel.
    """
    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 out

5.5 前馈

多头注意力层最初捕获序列内的上下文关系。通过两个连续的线性层，为网络增加了更多深度，这两个线性层共同构成前馈神经网络。

在最初的线性层中，我们增加了维数（在我们的案例中增加了4倍），这实际上扩大了网络学习和表示更复杂特征的能力。一个ReLU函数被应用在生成矩阵的每个元素上，这使得网络能够识别非线性模式。

随后，第二个线性层充当压缩器，将扩展的维度还原回原始形状（块大小 x 嵌入大小）。一个dropout层结束了这个过程，它随机地停用矩阵的元素，为了模型的泛化。

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. 添加与规范化

现在我们将多头注意力和前馈组件链接起来，通过引入两个更为关键的元素：

• 残差连接（添加）：这些连接执行层输出和它未改变的输入之间的逐元素相加。在训练过程中，模型调整对层变换的重视程度，基于它们的有效性。如果一个变换被认为是不必要的，它的权重以及因此产生的层输出倾向于趋近于零。至少在这种情况下，未改变的输入通过残差连接传递。这项技术有助于缓解梯度消失问题。

• 层规范化（规范）：这种方法通过减去每个嵌入向量的均值并除以其标准差来规范化上下文中的每个嵌入向量。这个过程也确保了在反向传播期间梯度既不会爆炸也不会消失。

多头注意力链和前馈层通过“加法和归一化”连接，形成一个模块。这种模块化设计使我们能够形成一系列的模块块。这些块的数量是一个超参数，它决定了模型架构的深度。

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 x

5.7. Softmax

在遍历了多个块组件后，我们获得了一个维度为（块大小 x 嵌入大小）的矩阵。为了将这个矩阵重塑为所需的维度（块大小 x 词汇表大小），我们通过一个最终的线性层进行处理。这个形状代表了上下文中每个位置上每个词汇的一个条目。

最后，我们对这些值应用了 soft-max 转换，将它们转换成概率。我们已经成功地获得了每个上下文位置上下一个标记的概率分布。

6. 模型训练

为了训练语言模型，我从训练数据中随机位置选择了标记序列。鉴于 WhatsApp 对话的快节奏特性，我确定32个词的上下文长度足够了。因此，我选择了随机的32词块作为上下文输入，并使用相对应的向量（词向后移动一个位置）作为比较的目标。

训练过程是循环以下步骤：

1. 采样多个上下文批次。
2. 将这些样本输入模型以计算当前损失。
3. 基于当前损失和模型权重应用反向传播。
4. 每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:
    """
    Trains or updates a GPTLanguageModel using pre-loaded data.
    This function either initializes a new model or loads an existing model based
    on the `update` parameter. It then trains the model using the AdamW optimizer
    on the training and validation data sets. Finally the trained model is saved.
    :param update: Boolean flag to indicate whether to update an existing model.
    """
    # LOAD DATA -----------------------------------------------------------------
    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())
    # INITIALIZE / LOAD MODEL ---------------------------------------------------
    if update:
        try:
            model = torch.load("assets/models/model.pt")
            print("Loaded existing model to continue training.")
        except FileNotFoundError:
            print("No existing model found. Initializing a new model.")
            model = GPTLanguageModel(vocab_size=len(vocab))
        
    else:
        print("Initializing a new model.")
        model = GPTLanguageModel(vocab_size=len(vocab))
    # initialize optimizer
    optimizer = torch.optim.AdamW(model.parameters(), lr=learn_rate)
    # number of model parameters
    n_params = sum(p.numel() for p in model.parameters())
    print(f"Parameters to be optimized: {n_params}\n", )
    # MODEL TRAINING ------------------------------------------------------------
    for i in range(max_iters):
        # evaluate the loss on train and valid sets every 'eval_interval' steps
        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} | step {i}: train loss {train_loss:.4f}, valid loss {valid_loss:.4f}")
        # sample batch of data
        x_batch, y_batch = get_batch(train_data)
        # evaluate the loss
        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("Model saved")

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:
    """
    Emulates chat conversations by sampling from a pre-trained GPTLanguageModel.
    This function loads a trained GPTLanguageModel along with vocabulary and 
    the list of special tokens. It then enters into a loop where the user specifies 
    a contact. Given this input, the model generates a sample response. The conversation 
    continues until the user inputs the end token.
    """
    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("message >> ", 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("\nresponse >> ", completer=completer, default="")
        print()

结论

为了至少展示一些示例输出，在最后，你可以看看这个虚拟模型如何处理200条训练过的虚拟信息 。