揭秘大模型的魔法：训练你的tokenizer

大家好，我是写代码的中年人。在这个人人谈论“Token量”、“百万上下文”、“按Token计费”的AI时代，“Tokenizer（分词器）”这个词频频出现在开发者和研究者的视野中。它是连接自然语言与神经网络之间的一座桥梁，是大模型运行逻辑中至关重要的一环。很多时候，你以为自己在和大模型对话，其实你和它聊的是一堆Token。

今天我们就来揭秘大模型背后的魔法之一：Tokenizer。我们不仅要搞懂什么是Tokenizer，还要了解BPE（Byte Pair Encoding）的分词原理，最后还会带你看看大模型是怎么进行分词的。我还会用代码演示：如何训练你自己的Tokenizer！

注：揭秘大模型的魔法属于连载文章，一步步带你打造一个大模型。

Tokenizer 是什么

Tokenizer是大模型语言处理中用于将文本转化为模型可处理的数值表示（通常是token ID序列）的关键组件。它负责将输入文本分割成最小语义单元（tokens），如单词、子词或字符，并将其映射到对应的ID。在大模型的世界里，模型不会直接处理我们熟悉的文本。例如，输入：

Hello, world!

模型并不会直接理解“H”、“e”、“l”、“l”、“o”，它理解的是这些字符被转换成的数字——准确地说，是Token ID。Tokenizer的作用就是：

把原始文本分割成“Token”：通常是词、词干、子词，甚至字符或字节。

将这些Token映射为唯一的整数ID：也就是模型训练和推理中使用的“输入向量”。

最终的流程是：

文本 => Token列表 => Token ID => 输入大模型

每个模型的 Tokenizer 通常都是不一样的，下表列举了一些影响Tokenizer的因素：Tokenizer 是语言模型的“地基”之一，并不是可以通用的。一个合适的 tokenizer 会大幅影响：模型的 token 分布、收敛速度、上下文窗口利用率、稀疏词的处理能力。如上图，不同模型，分词方法不同，对应的Token ID也不同。

常见的分词方法介绍

常见的分词方法是大模型语言处理中将文本分解为最小语义单元（tokens）的核心技术。不同的分词方法适用于不同场景，影响模型的词汇表大小、处理未登录词（OOV）的能力以及计算效率。以下是常见分词方法的介绍：

01、基于单词的分词

原理：将文本按空格或标点分割为完整的单词，每个单词作为一个token。

实现：通常结合词汇表，将单词映射到ID。未在词汇表中的词被标记为[UNK]（未知）。

优点：简单直观，token具有明确的语义。适合英语等以空格分隔的语言。

缺点：词汇表可能很大（几十万到百万），增加了模型的参数和内存。未登录词（OOV）问题严重，如新词、拼写错误无法处理。对中文等无明显分隔的语言不适用。

应用场景：早期NLP模型，如Word2Vec。适合词汇量有限的特定领域任务。

示例：文本: "I love coding" → Tokens: ["I", "love", "coding"]

02、基于字符的分词

原理：将文本拆分为单个字符（或字节），每个字符作为一个token。

实现：词汇表只包含字符集（如ASCII、Unicode），无需复杂的分词规则。

优点：词汇表极小（几十到几百），内存占用低。无未登录词问题，任何文本都能被分解。适合多语言和拼写变体。

缺点：token序列长，增加模型计算负担（如Transformer的注意力机制）。丢失单词级语义，模型需学习更复杂的上下文关系。

应用场景：多语言模型（如mBERT的部分实现）。处理拼写错误或非标准文本的任务。

示例：文本: "I love" → Tokens: ["I", " ", "l", "o", "v", "e"]

03、基于子词的分词

原理：将文本分解为介于单词和字符之间的子词单元，常见算法包括BPE、WordPiece和Unigram LM。子词通常是高频词或词片段。

实现：通过统计或优化算法构建词汇表，动态分割文本，保留常见词并拆分稀有词。

优点：平衡了词汇表大小和未登录词处理能力。能处理新词、拼写变体和多语言文本。token具有一定语义，序列长度适中。

缺点：分词结果可能不直观（如"playing"拆为"play" + "##ing"）。需要预训练分词器，增加前期成本。

常见子词算法

01、Byte-Pair Encoding (BPE)

原理：从字符开始，迭代合并高频字符对，形成子词。

应用：GPT系列、RoBERTa。

示例："lowest" → ["low", "##est"]。

02、WordPiece

原理：类似BPE，但基于最大化语言模型似然选择合并。

应用：BERT、Electra。

示例："unhappiness" → ["un", "##hap", "##pi", "##ness"]。

03、Unigram Language Model

原理：通过语言模型优化选择最优子词集合，允许多种分割路径。

应用：T5、ALBERT

应用场景：几乎所有现代大模型（如BERT、GPT、T5）。多语言、通用NLP任务。

示例：文本: "unhappiness" → Tokens: ["un", "##hap", "##pi", "##ness"]

04、基于SentencePiece的分词

原理：一种无监督的分词方法，将文本视为字符序列，直接学习子词分割，不依赖语言特定的预处理（如空格分割）。支持BPE或Unigram LM算法。

实现：训练一个模型（.model文件），包含词汇表和分词规则，直接对原始文本编码/解码。

优点：语言无关，适合多语言和无空格语言（如中文、日文）。统一处理原始文本，无需预分词。能处理未登录词，灵活性高。

缺点：需要额外训练分词模型。分词结果可能不够直观。

应用场景：T5、LLaMA、mBART等跨语言模型。中文、日文等无明确分隔的语言。

示例：文本: "こんにちは"（日语：你好） → Tokens: ["▁こ", "ん", "に", "ち", "は"]

05、基于规则的分词

原理：根据语言特定的规则（如正则表达式）将文本分割为单词或短语，常结合词典或语法规则。

实现：使用工具（如Jieba for Chinese、Mecab for Japanese）或自定义规则进行分词。

优点：分词结果符合语言习惯，语义清晰。适合特定语言或领域（如中文分词）。

缺点：依赖语言特定的规则和词典，跨语言通用性差。维护成本高，难以处理新词或非标准文本。

应用场景：中文（Jieba、THULAC）、日文（Mecab）、韩文等分词。特定领域的专业术语分词。

示例：文本: "我爱编程"（中文） → Tokens: ["我", "爱", "编程"]

06、基于Byte-level Tokenization

原理：直接将文本编码为字节序列（UTF-8编码），每个字节作为一个token。常结合BPE（如Byte-level BPE）。

实现：无需预定义词汇表，直接处理字节序列，动态生成子词。

优点：完全语言无关，词汇表极小（256个字节）。无未登录词问题，适合多语言和非标准文本。

缺点：序列长度较长，计算开销大。语义粒度低，模型需学习复杂模式。

应用场景：GPT-3、Bloom等大规模多语言模型。处理原始字节输入的任务。

示例：文本: "hello" → Tokens: ["h", "e", "l", "l", "o"]（或字节表示）。

从零实现BPE分词器

子词分词（BPE、WordPiece、SentencePiece）是现代大模型的主流，因其在词汇表大小、未登录词处理和序列长度之间取得平衡，本次我们使用纯Python，不依赖任何开源框架来实现一个BPE分词器。

我们先实现一个BPETokenizer类：

import jsonfrom collections import defaultdictimport reimport os
class BPETokenizer:    def __init__(self):        self.vocab = {}  # token -> id        self.inverse_vocab = {}  # id -> token        self.merges = []  # List of (token1, token2) pairs        self.merge_ranks = {}  # pair -> rank        self.next_id = 0        self.special_tokens = []
    def get_stats(self, word_freq):        pairs = defaultdict(int)        for word, freq in word_freq.items():            symbols = word.split()            for i in range(len(symbols) - 1):                pairs[(symbols[i], symbols[i + 1])] += freq        return pairs
    def merge_vocab(self, pair, word_freq):        bigram = ' '.join(pair)        replacement = ''.join(pair)        new_word_freq = {}        pattern = re.compile(r'(?<!S)' + re.escape(bigram) + r'(?!S)')        for word, freq in word_freq.items():            new_word = pattern.sub(replacement, word)            new_word_freq[new_word] = freq        return new_word_freq
    def train(self, corpus, vocab_size, special_tokens=None):        if special_tokens is None:            special_tokens = ['[PAD]', '[UNK]', '[CLS]', '[SEP]', '[MASK]']        self.special_tokens = special_tokens
        for token in special_tokens:            self.vocab[token] = self.next_id            self.inverse_vocab[self.next_id] = token            self.next_id += 1
        word_freq = defaultdict(int)        for text in corpus:            words = re.findall(r'w+|[^ws]', text, re.UNICODE)            for word in words:                word_freq[' '.join(list(word))] += 1
        while len(self.vocab) < vocab_size:            pairs = self.get_stats(word_freq)            if not pairs:                break            best_pair = max(pairs, key=pairs.get)            self.merges.append(best_pair)            self.merge_ranks[best_pair] = len(self.merges) - 1            word_freq = self.merge_vocab(best_pair, word_freq)            new_token = ''.join(best_pair)            if new_token not in self.vocab:                self.vocab[new_token] = self.next_id                self.inverse_vocab[self.next_id] = new_token                self.next_id += 1
    def encode(self, text):        words = re.findall(r'w+|[^ws]', text, re.UNICODE)        token_ids = []        for word in words:            tokens = list(word)            while len(tokens) > 1:                pairs = [(tokens[i], tokens[i + 1]) for i in range(len(tokens) - 1)]                merge_pair = None                merge_rank = float('inf')                for pair in pairs:                    rank = self.merge_ranks.get(pair, float('inf'))                    if rank < merge_rank:                        merge_pair = pair                        merge_rank = rank                if merge_pair is None:                    break                new_tokens = []                i = 0                while i < len(tokens):                    if i < len(tokens) - 1 and (tokens[i], tokens[i + 1]) == merge_pair:                        new_tokens.append(''.join(merge_pair))                        i += 2                    else:                        new_tokens.append(tokens[i])                        i += 1                tokens = new_tokens            for token in tokens:                token_ids.append(self.vocab.get(token, self.vocab['[UNK]']))        return token_ids
    def decode(self, token_ids):        tokens = [self.inverse_vocab.get(id, '[UNK]') for id in token_ids]        return ''.join(tokens)
    def save(self, output_dir):        os.makedirs(output_dir, exist_ok=True)        with open(os.path.join(output_dir, 'vocab.json'), 'w', encoding='utf-8') as f:            json.dump(self.vocab, f, ensure_ascii=False, indent=2)        with open(os.path.join(output_dir, 'merges.txt'), 'w', encoding='utf-8') as f:            for pair in self.merges:                f.write(f"{pair[0]} {pair[1]}n")        with open(os.path.join(output_dir, 'tokenizer_config.json'), 'w', encoding='utf-8') as f:            config = {                "model_type": "bpe",                "vocab_size": len(self.vocab),                "special_tokens": self.special_tokens,                "merges_file": "merges.txt",                "vocab_file": "vocab.json"            }            json.dump(config, f, ensure_ascii=False, indent=2)
    def export_token_map(self, path):        with open(path, 'w', encoding='utf-8') as f:            for token_id, token in self.inverse_vocab.items():                f.write(f"{token_id}t{token}t{' '.join(token)}n")
    def print_visualization(self, text):        words = re.findall(r'w+|[^ws]', text, re.UNICODE)        visualized = []        for word in words:            tokens = list(word)            while len(tokens) > 1:                pairs = [(tokens[i], tokens[i + 1]) for i in range(len(tokens) - 1)]                merge_pair = None                merge_rank = float('inf')                for pair in pairs:                    rank = self.merge_ranks.get(pair, float('inf'))                    if rank < merge_rank:                        merge_pair = pair                        merge_rank = rank                if merge_pair is None:                    break                new_tokens = []                i = 0                while i < len(tokens):                    if i < len(tokens) - 1 and (tokens[i], tokens[i + 1]) == merge_pair:                        new_tokens.append(''.join(merge_pair))                        i += 2                    else:                        new_tokens.append(tokens[i])                        i += 1                tokens = new_tokens            visualized.append(' '.join(tokens))        return ' | '.join(visualized)
    def load(self, path):        with open(os.path.join(path, 'vocab.json'), 'r', encoding='utf-8') as f:            self.vocab = json.load(f)            self.vocab = {k: int(v) for k, v in self.vocab.items()}            self.inverse_vocab = {v: k for k, v in self.vocab.items()}            self.next_id = max(self.vocab.values()) + 1
        with open(os.path.join(path, 'merges.txt'), 'r', encoding='utf-8') as f:            self.merges = []            self.merge_ranks = {}            for i, line in enumerate(f):                token1, token2 = line.strip().split()                pair = (token1, token2)                self.merges.append(pair)                self.merge_ranks[pair] = i
        config_path = os.path.join(path, 'tokenizer_config.json')        if os.path.exists(config_path):            with open(config_path, 'r', encoding='utf-8') as f:                config = json.load(f)                self.special_tokens = config.get("special_tokens", [])

函数说明：

__init__：初始化分词器，创建词汇表、合并规则等数据结构。

get_stats：统计词频字典中相邻符号对的频率。

merge_vocab：根据符号对合并词频字典中的token。

train：基于语料库训练BPE分词器，构建词汇表。

encode：将文本编码为token id序列。

decode：将token id序列解码为文本。

save：保存分词器状态到指定目录。

export_token_map：导出token映射到文件。

print_visualization：可视化文本的BPE分词过程。

load：从指定路径加载分词器状态。

加载测试数据进行训练：

if __name__ == "__main__":    corpus = load_corpus_from_file("水浒传.txt")
    tokenizer = BPETokenizer()    tokenizer.train(corpus, vocab_size=500)
    tokenizer.save("./bpe_tokenizer")    tokenizer.export_token_map("./bpe_tokenizer/token_map.tsv")
    print("nSaved files:")    print(f"vocab.json: {os.path.exists('./bpe_tokenizer/vocab.json')}")    print(f"merges.txt: {os.path.exists('./bpe_tokenizer/merges.txt')}")    print(f"tokenizer_config.json: {os.path.exists('./bpe_tokenizer/tokenizer_config.json')}")    print(f"token_map.tsv: {os.path.exists('./bpe_tokenizer/token_map.tsv')}")

此处我选择了开源的数据，水浒传全文档进行训练，请注意：训练数据应该以章节分割，请根据具体上下文决定。

文章如下：

在这里要注意vocab_size值的选择：

小语料测试 → vocab_size=100~500

训练 AI 语言模型前分词器 → vocab_size=1000~30000

实际场景调优 → 可实验不同大小，看 token 数、OOV 情况等

进行训练：

我们执行完训练代码后，程序会在bpe_tokenizer文件夹下生成4个文件：

vocab.json：存储词汇表，记录每个token到其id的映射（如{"[PAD]": 0, "he": 256}）。

merges.txt：存储BPE合并规则，每行是一对合并的符号（如h e表示合并为he）。

tokenizer_config.json：存储分词器配置，包括模型类型、词汇表大小、特殊token等信息。

token_map.tsv：存储token id到token的映射，每行格式为idttokenttoken的字符序列（如256theth e），用于调试或分析。

我们本次测试vocab_size选择了500，我们打开vocab.json查看，里面有500个词：

进行测试：

我们执行如下代码进行测试：

if __name__ == '__main__':    # 加载分词器    tokenizer = BPETokenizer()    tokenizer.load('./bpe_tokenizer')
    # 测试分词和还原    text = "且说鲁智深自离了五台山文殊院，取路投东京来，行了半月之上。"    ids = tokenizer.encode(text)    print("Encoded:", ids)    print("Decoded:", tokenizer.decode(ids))
    print("nVisualization:")    print(tokenizer.print_visualization(text))

# 输出Encoded: [60, 67, 1, 238, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 125, 1, 1, 1, 1, 1, 1, 1, 1, 1]Decoded: 且说鲁智深[UNK]离了[UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK]东京[UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK]
Visualization:且说 鲁智深 自 离了 五 台 山 文 殊 院 | ， | 取 路 投 东京 来 | ， | 行 了 半 月 之 上 | 。

我们看到解码后，输出很多[UNK]，出现 [UNK] 并非编码器的问题，而是训练语料覆盖不够和vocab设置的值太小， 导致token 没有进入 vocab。这个到后边我们真正训练时，再说明。

BPE它是一种压缩+分词混合技术。初始时我们把句子分成单字符。然后统计出现频率最高的字符对，不断合并，直到词表大小满足预设。

经过本章学习，我们可以创建一个简单的分词器，你可以复制代码到编译器执行测试，如果需要文档，也可以关注公众号后，加我微信，我会发给你。下一章我们将进入Transformer架构的详解。