动手学LLM（ch2）——文本数据处理

前言

在这里，您将学习如何为训练大型语言模型（LLMs）准备输入文本。这包括将文本分割成单个词汇和子词汇token，然后将它们编码成向量表示，供大型语言模型（LLM）使用。您还将了解字节对编码等高级分词方案，这些方案已经在GPT等流行的大型语言模型（LLMs）中得到应用。最后，我们将介绍采样和数据加载策略，这些策略用于生成后续中训练大型语言模型（LLMs)所需的输入输出对。本文内容从以下几个方面展开：

- 为大型语言模型训练准备文本

- 将文本分割为词汇和子词汇token

- 字节对编码是一种更高级的文本分词方法

- 使用滑动窗口方法抽样训练示例

- 将token转换为向量输入大型语言模型

2.1 理解词嵌入

深度神经网络模型，包括大型语言模型（LLMs），无法直接处理原始文本，因为文本是分类数据，与神经网络的数学运算不兼容。为了达到这个目的，需要将单词转换为连续值向量。记住一句话就行，为了兼容神经网络的数学运算，把所有原始数据类型转变为连续值向量就是嵌入（embedding）。

如下图所示，通过特定的神经网络层或预训练模型，可以将视频、音频和文本等不同类型的数据嵌入为密集向量表示，使深度学习架构能够理解和处理这些原始数据。

需要注意的是，不同的数据格式需要各自专用的嵌入模型，例如，针对文本设计的嵌入模型不适用于音频或视频数据。

嵌入本质上是将离散对象（如单词、图像或文档）映射到连续向量空间的过程，目的是将非数字数据转换为神经网络可处理的格式。虽然词嵌入是最常见的形式，但也可以扩展到句子、段落或整个文档。当前，词嵌入的生成有多种算法和框架，其中Word2Vec是早期和流行的选择，它通过预测上下文生成词嵌入，基于相似上下文中词的相似含义。词嵌入可以有不同维度，从一维到数千维，高维度可以捕捉更多细微关系，尽管计算效率较低。大型语言模型（LLMs）通常生成自己的嵌入，优化以适应特定任务和数据。以GPT-2和GPT-3为例，嵌入大小根据模型变种变化，最小的GPT-2和GPT-3使用768维，而最大的GPT-3使用12,288维，这反映了性能与效率之间的权衡。

如下图所示，如果词嵌入是二维的，我们可以将它们绘制在二维散点图中以便于可视化。在使用词嵌入技术，例如 Word2Vec 时，对应于相似概念的词在嵌入空间中通常彼此接近。例如，在嵌入空间中，不同类型的鸟类相对于国家和城市更为靠近。

注意的是，LLMs 还可以创建上下文化的输出嵌入。这个后面会去讨论

2.2 文本分词（序列化）

如何将输入文本分割成单独的token，这是大型语言模型（LLM）创建嵌入的必需预处理步骤。

如下图所示，文本处理步骤在大型语言模型（LLM）中的最开始的阶段。我们将输入文本分割成单独的token，这些token可能是单词或特殊字符，例如标点符号。后续，我们会将把文本转换成token ID 并创建token嵌入。

读取本地的文本文件

这里没什么需要注意的，就是很简单的读取文件的代码。

# 设置要读取的文件路径，这里是本地文件
file_path = "./the-verdict.txt"
# 打开文件并读取内容
with open(file_path, 'r', encoding='utf-8') as file:
    raw_text = file.read()  # 读取文件内容
# 输出文本的总字符数
print("Total number of characters:", len(raw_text))
# 输出文件内容的前99个字符，以便预览
print(raw_text[:99])

使用 Python 的正则表达式库 re 模块来示例说明分词

这里其实看看就行，知道分词后大概什么样子就好了，因为我们后面转用预构建的分词器。

# 设置要读取的文件路径，这里是本地文件
file_path = "./the-verdict.txt"
# 打开文件并读取内容
with open(file_path, 'r', encoding='utf-8') as file:
    raw_text = file.read()  # 读取文件内容
# # 输出文本的总字符数
# print("Total number of characters:", len(raw_text))
# # 输出文件内容的前99个字符，以便预览
# print(raw_text[:99])

import re  # 导入正则表达式模块

# 定义一个包含标点和空格的字符串
text = "Hello, world. This, is a test."

# 使用正则表达式按照空格进行分割，并保留空格作为分割符
result = re.split(r'(\s)', text)
print(result)  # 输出分割结果

# 使用正则表达式按照标点（逗号、句号）和空格进行分割，并保留分割符
result = re.split(r'([,.]|\s)', text)
print(result)  # 输出分割结果

# 去除结果中的空字符串
result = [item.strip() for item in result if item.strip()]
print(result)  # 输出处理后的结果

# 另一个包含不同标点和空格的字符串
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)  # 输出处理后的结果

preprocessed = re.split(r'([,.?_!"()\']|--|\s)', raw_text)
preprocessed = [item.strip() for item in preprocessed if item.strip()]
print("_"*100)
print(preprocessed[:10])
print(len(preprocessed))

2.3 将token转换为tokenID

如下图所示，我们将文本token转换为tokenID，稍后可以通过嵌入层进行处理。

# 将预处理后的单词列表去重并排序
all_words = sorted(set(preprocessed))
# 计算词汇表的大小
vocab_size = len(all_words)
# 创建一个字典，将每个单词映射到唯一的整数索引
vocab = {token: integer for integer, token in enumerate(all_words)}
# 输出词汇表的大小
print(vocab_size)

# 遍历词汇字典并输出前50个单词及其对应的索引
for i, item in enumerate(vocab.items()):
    print(item)  # 输出单词及其索引
    if i >= 50:  # 当索引达到50时停止输出
        break

封装标记器类

该类实现了一个简单的文本编码和解码器。构造函数接收一个词汇表，将其映射为字符串到整数和整数到字符串的字典。encode 方法将输入文本分割、预处理并转换为整数 ID 列表，decode 方法则将 ID 列表转换回文本，并处理标点前的空格。如下图所示

class SimpleTokenizerV1:
    def __init__(self, vocab):
        # 初始化时接收一个词汇表字典
        self.str_to_int = vocab  # 字符串到整数的映射
        # 创建整数到字符串的反向映射
        self.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()
        ]
        # 将处理后的每个单词转换为对应的整数ID
        ids = [self.str_to_int[s] for s in preprocessed]
        return ids  # 返回编码后的ID列表

    def decode(self, ids):
        # 根据ID列表生成对应的文本字符串
        text = " ".join([self.int_to_str[i] for i in ids])
        # 替换指定标点前的空格
        text = re.sub(r'\s+([,.?!"()\'])', r'\1', text)
        return text  # 返回解码后的文本

# 创建一个 SimpleTokenizerV1 实例，传入词汇表
tokenizer = SimpleTokenizerV1(vocab)
# 定义要编码的文本
text = """"It's the last he painted, you know," 
           Mrs. Gisburn said with pardonable pride."""

# 使用 tokenizer 的 encode 方法将文本编码为 ID 列表
ids = tokenizer.encode(text)
# 打印编码后的 ID 列表
print("Encoded IDs:", ids)
# 使用 tokenizer 的 decode 方法将 ID 列表解码回文本
decoded_text = tokenizer.decode(ids)
# 打印解码后的文本
print("Decoded text:", decoded_text)
# 再次使用 encode 和 decode 方法，验证编码和解码的一致性
# 编码文本后再解码
consistent_decoded_text = tokenizer.decode(tokenizer.encode(text))
# 打印一致性验证的结果
print("Consistent decoded text:", consistent_decoded_text)

2.4 添加特殊上下文token

在文本处理中，为了提供额外的上下文，添加一些“特殊”标记是非常有用的，特别是在处理未知单词和文本结束时。一些标记化器使用特殊标记来帮助大语言模型（LLM）理解文本，如下图所示。以下是一些常见的特殊标记：

[BOS]（序列开始）表示文本的开头。
[EOS]（序列结束）标记文本结束的位置，通常用于连接多段不相关的文本，例如两篇不同的维基百科文章或两本不同的书籍。
[PAD]（填充）在训练 LLM 时，当批量大小大于 1 时，可能会包含长度不同的多个文本；使用填充标记可以将较短的文本填充到最长的长度，以使所有文本具有相同的长度。
[UNK] 用于表示不在词汇表中的单词。

此外，GPT-2 不使用 <UNK> 标记来处理超出词汇表的单词，而是采用字节对编码（BPE）标记化器，将单词分解为子词单元，这将在后面中讨论。

注意，在进行文本标记化时，如果输入文本中的单词不在词汇表中，例如 "Hello"，则会产生错误。为了解决这种情况，可以在词汇表中添加特殊标记，如 "<|unk|>"，用于表示未知单词。此外，由于我们已经在扩展词汇表，可以再添加一个名为 "<|endoftext|>" 的标记，它在 GPT-2 的训练中用于表示文本的结束（并且在连接多个文本时也会使用，例如当训练数据集包含多篇文章、书籍等时）。

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

为了让标记化器能够正确使用新的 <|unk|> 标记，我们需要对其进行相应的调整。接下来，我们可以尝试使用修改后的标记化器进行标记化

import re  # 导入正则表达式模块

class SimpleTokenizerV2:
    def __init__(self, vocab):
        # 初始化时接收一个词汇表字典
        self.str_to_int = vocab  # 字符串到整数的映射
        # 创建整数到字符串的反向映射
        self.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()]
        # 将不在词汇表中的单词替换为 "<|unk|>"
        preprocessed = [
            item if item in self.str_to_int
            else "<|unk|>" for item in preprocessed
        ]

        # 将处理后的每个单词转换为对应的整数ID
        ids = [self.str_to_int[s] for s in preprocessed]
        return ids  # 返回编码后的ID列表

    def decode(self, ids):
        # 根据ID列表生成对应的文本字符串
        text = " ".join([self.int_to_str[i] for i in ids])
        # 替换指定标点前的空格
        text = re.sub(r'\s+([,.:;?!"()\'])', r'\1', text)
        return text  # 返回解码后的文本

# 文件路径
file_path = "./the-verdict.txt"
# 打开文件并读取内容
with open(file_path, 'r', encoding='utf-8') as file:
    raw_text = file.read()  # 读取文件内容

# 对文本进行预处理，分割并去除空白
preprocessed = re.split(r'([,.?_!"()\']|--|\s)', raw_text)
preprocessed = [item.strip() for item in preprocessed if item.strip()]

# 将预处理后的单词列表去重并排序
all_tokens = sorted(set(preprocessed))
# 在词汇表中添加特殊标记
all_tokens.extend(["<|endoftext|>", "<|unk|>"])
# 计算词汇表的大小
vocab_size = len(all_tokens)

# 创建一个字典，将每个单词映射到唯一的整数索引
vocab = {token: integer for integer, token in enumerate(all_tokens)}

# 打印词汇表中最后五个单词及其索引
for i, item in enumerate(list(vocab.items())[-5:]):
    print(item)

# 创建标记化器实例
tokenizer = SimpleTokenizerV2(vocab)

# 定义两个示例文本
text1 = "Hello, do you like tea?"
text2 = "In the sunlit terraces of the palace."
# 连接两个文本，用 "<|endoftext|>" 标记分隔
text = " <|endoftext|> ".join((text1, text2))
print(text)  # 打印连接后的文本

# 使用标记化器进行编码
ids = tokenizer.encode(text)

# 打印编码后的 ID 列表
print("Encoded IDs:", ids)

# 使用标记化器的 decode 方法将 ID 列表解码回文本
decoded_text = tokenizer.decode(ids)
# 打印解码后的文本
print("Decoded text:", decoded_text)

# 再次使用 encode 和 decode 方法，验证编码和解码的一致性
# 编码文本后再解码
consistent_decoded_text = tokenizer.decode(tokenizer.encode(text))
# 打印一致性验证的结果
print("Consistent decoded text:", consistent_decoded_text)

2.5 字节对编码（BPE）

GPT-2 使用字节对编码（BytePair Encoding, BPE）作为其标记化器。
这种方法允许模型将不在预定义词汇表中的单词拆分为更小的子词单元，甚至是单个字符，从而能够处理超出词汇表的单词。
例如，如果 GPT-2 的词汇表中没有单词 "unfamiliarword"，它可能将其标记化为 ["unfam", "iliar", "word"] 或其他子词拆分，这取决于其训练时的 BPE 合并策略。
原始的 BPE 标记化器代码可以在这里找到：https://github.com/openai/gpt-2/blob/master/src/encoder.py。
在本章中，我们使用了 OpenAI 的开源库 tiktoken 中的 BPE 标记化器，该库的核心算法用 Rust 实现，以提高计算性能。
我在 ./bytepair_encoder 目录中创建了一个笔记本，比较了这两种实现的性能（tiktoken 在样本文本上的速度约快 5 倍）。

# pip install tiktoken

import 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)

基于上述的token IDs以及解码的文本，我们可以做出2点有价值的观察。

第一，`<|endoftext|>`词元（token）被赋值了一个很大的token ID，例如，50256。事实上，被用于训练诸如GPT-2，GPT-3以及被ChatGPT使用的原始模型的BPE分词器，总计词汇的规模是50257，其中`<|endoftext|>`被指定为最大的token ID。

第二，上述的BPE分词器可以正确的解码和编码没有见过的词汇，例如"someunknownPlace"。BPE解码器可以处理任何没有见过的词汇。那么，他是怎么无需使用`<|unk|>`词元就做到这个的呢？

BPE使用的算法会将不在预定义词表里的单词分解为更小的子单词单元或者甚至是独立的字母，使BPE可以处理词表外的单词。所以，基于这种算法，如果分词器在分词时遇到了不熟悉的单词，他会使用一系列的子单词词元或者字母来替换它，就像如下图所示，将不认识的单词分解为更小的的词元或字母保证了分词器，以及后续被训练的大模型可以处理任意的文本，即使文本里包含了从来没在测试数据里出现过的单词。对BPE更详细的讨论和实现可以搜搜看看别的博客。

**练习2.1 未知单词的字节对编码**

尝试使用tiktoken库里的BPE分词器，对未知词汇"Akwirw ier"进行处理然后打印出一列词元编码（token IDs）。然后，针对这一列的每一个编码调用解码函数来重现上图所示的映射关系。最后，在词元ID序列上调用解码方法来确认是否它们可以重新构成原始的输入"Akwirw ier"。
# test
import tiktoken

# 初始化分词器，这里使用 GPT-2 模型的 BPE 分词器
bpe = tiktoken.get_encoding("gpt2")

# 输入的未知词汇
input_text = "Akwirw ier"

# 对输入进行分词并打印词元ID
token_ids = bpe.encode(input_text)
print(f"词元ID: {token_ids}")

# 针对每一个词元ID，进行解码，查看它们各自代表的字符
print("单个词元ID对应的解码结果:")
for token_id in token_ids:
    print(f"{token_id}: {bpe.decode([token_id])}")

# 对整个词元ID序列进行解码，并验证是否与原始输入匹配
decoded_text = bpe.decode(token_ids)
print(f"解码后的文本: {decoded_text}")

# 验证解码后的文本是否与原始文本一致
if decoded_text == input_text:
    print("解码后的文本与原始输入匹配。")
else:
    print("解码后的文本与原始输入不匹配。")

2.6 使用滑动窗口进行数据采样

在创建 LLM 的 Embedding 之前，我们需要生成训练 LLM 所需的输入-目标（input-target）对。这些输入-目标对是什么样的呢？LLM 是通过预测文本中的下一个单词来进行预训练的，如图 2.1 所示。如下图所示，给定一个文本样本，提取输入块作为 LLM 的输入子样本，LLM 在训练期间的任务是预测输入块之后的下一个单词。在训练过程中，我们会屏蔽掉目标词之后的所有单词。请注意，在 LLM 处理文本之前，该文本已经进行 token 化，为了便于说明，此图省略了 token 化步骤。本小结中实现了一个数据加载器，它使用滑动窗口方法从训练数据集中获取下图中描述的输入-目标对。

首先，使用BPE 分词器对整个《The Verdict》短篇故事进行分词。

import tiktoken

tokenizer = tiktoken.get_encoding("gpt2")
with open("the-verdict.txt", "r", encoding="utf-8") as f:
    raw_text = f.read()
enc_text = tokenizer.encode(raw_text)
print(len(enc_text))

对训练集应用 BPE 分词器后获得 5145 个 tokens。

接下来，从数据集中剔除前 50 个 toekns，以便在后续步骤中展示更吸引人的文本段落。在创建下一个单词预测任务的输入-目标对时，一种简单直观的方法是创建两个变量 x 和 y。其中，x 用于存储输入的 token 序列，而 y 则用于存放目标 token 序列。目标序列由输入序列中的每个 token 向右移动一个位置构成。从而形成了输入-目标对。

enc_sample = enc_text[50:]
context_size = 4
x = enc_sample[:context_size]
y = enc_sample[1:context_size+1]
print(f"x: {x}")
print(f"y:      {y}")

通过将输入数据向右移动一个位置来生成对应的目标数据后。可以上图，按照以下步骤创建下一个单词的预测任务

for i in range(1, context_size+1):
    context = enc_sample[:i]
    desired = enc_sample[i]
    print(context, "---->", desired)

为了便于更好地理解，我们重复之前的代码，但这次我们将 token ID 转换回文本

for i in range(1, context_size+1):
    context = enc_sample[:i]
    desired = enc_sample[i]
    print(tokenizer.decode(context), "---->", tokenizer.decode([desired]))