• 正文
  • 相关推荐
申请入驻 产业图谱

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

04/24 09:00
1894
加入交流群
扫码加入
获取工程师必备礼包
参与热点资讯讨论

大家好,我是写代码的中年人。在这个人人谈论“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架构的详解。

相关推荐

登录即可解锁
  • 海量技术文章
  • 设计资源下载
  • 产业链客户资源
  • 写文章/发需求
立即登录