揭秘大模型的魔法：訓(xùn)練你的tokenizer
大家好，我是寫代碼的中年人。在這個(gè)人人談?wù)摗癟oken量”、“百萬(wàn)上下文”、“按Token計(jì)費(fèi)”的AI時(shí)代，“Tokenizer（分詞器）”這個(gè)詞頻頻出現(xiàn)在開(kāi)發(fā)者和研究者的視野中。它是連接自然語(yǔ)言與神經(jīng)網(wǎng)絡(luò)之間的一座橋梁，是大模型運(yùn)行邏輯中至關(guān)重要的一環(huán)。很多時(shí)候，你以為自己在和大模型對(duì)話，其實(shí)你和它聊的是一堆Token。
今天我們就來(lái)揭秘大模型背后的魔法之一：Tokenizer。我們不僅要搞懂什么是Tokenizer，還要了解BPE（Byte Pair Encoding）的分詞原理，最后還會(huì)帶你看看大模型是怎么進(jìn)行分詞的。我還會(huì)用代碼演示：如何訓(xùn)練你自己的Tokenizer！
注：揭秘大模型的魔法屬于連載文章，一步步帶你打造一個(gè)大模型。
Tokenizer 是什么
Tokenizer是大模型語(yǔ)言處理中用于將文本轉(zhuǎn)化為模型可處理的數(shù)值表示（通常是token ID序列）的關(guān)鍵組件。它負(fù)責(zé)將輸入文本分割成最小語(yǔ)義單元（tokens），如單詞、子詞或字符，并將其映射到對(duì)應(yīng)的ID。
在大模型的世界里，模型不會(huì)直接處理我們熟悉的文本。例如，輸入：
Hello, world!模型并不會(huì)直接理解“H”、“e”、“l(fā)”、“l(fā)”、“o”，它理解的是這些字符被轉(zhuǎn)換成的數(shù)字——準(zhǔn)確地說(shuō)，是Token ID。Tokenizer的作用就是：
把原始文本分割成“Token”：通常是詞、詞干、子詞，甚至字符或字節(jié)。
將這些Token映射為唯一的整數(shù)ID：也就是模型訓(xùn)練和推理中使用的“輸入向量”。
最終的流程是：
文本 => Token列表 => Token ID => 輸入大模型每個(gè)模型的 Tokenizer 通常都是不一樣的，下表列舉了一些影響Tokenizer的因素：
Tokenizer 是語(yǔ)言模型的“地基”之一，并不是可以通用的。一個(gè)合適的 tokenizer 會(huì)大幅影響：模型的 token 分布、收斂速度、上下文窗口利用率、稀疏詞的處理能力。
如上圖，不同模型，分詞方法不同，對(duì)應(yīng)的Token ID也不同。
常見(jiàn)的分詞方法介紹
常見(jiàn)的分詞方法是大模型語(yǔ)言處理中將文本分解為最小語(yǔ)義單元（tokens）的核心技術(shù)。不同的分詞方法適用于不同場(chǎng)景，影響模型的詞匯表大小、處理未登錄詞（OOV）的能力以及計(jì)算效率。以下是常見(jiàn)分詞方法的介紹：
01 基于單詞的分詞
原理：將文本按空格或標(biāo)點(diǎn)分割為完整的單詞，每個(gè)單詞作為一個(gè)token。
實(shí)現(xiàn)：通常結(jié)合詞匯表，將單詞映射到ID。未在詞匯表中的詞被標(biāo)記為[UNK]（未知）。
優(yōu)點(diǎn)：簡(jiǎn)單直觀，token具有明確的語(yǔ)義。適合英語(yǔ)等以空格分隔的語(yǔ)言。
缺點(diǎn)：詞匯表可能很大（幾十萬(wàn)到百萬(wàn)），增加了模型的參數(shù)和內(nèi)存。未登錄詞（OOV）問(wèn)題嚴(yán)重，如新詞、拼寫錯(cuò)誤無(wú)法處理。對(duì)中文等無(wú)明顯分隔的語(yǔ)言不適用。
應(yīng)用場(chǎng)景：早期NLP模型，如Word2Vec。適合詞匯量有限的特定領(lǐng)域任務(wù)。
示例：文本: "I love coding" → Tokens: ["I", "love", "coding"]
02 基于字符的分詞
原理：將文本拆分為單個(gè)字符（或字節(jié)），每個(gè)字符作為一個(gè)token。
實(shí)現(xiàn)：詞匯表只包含字符集（如ASCII、Unicode），無(wú)需復(fù)雜的分詞規(guī)則。
優(yōu)點(diǎn)：詞匯表極?。◣资綆装伲瑑?nèi)存占用低。無(wú)未登錄詞問(wèn)題，任何文本都能被分解。適合多語(yǔ)言和拼寫變體。
缺點(diǎn)：token序列長(zhǎng)，增加模型計(jì)算負(fù)擔(dān)（如Transformer的注意力機(jī)制）。丟失單詞級(jí)語(yǔ)義，模型需學(xué)習(xí)更復(fù)雜的上下文關(guān)系。
應(yīng)用場(chǎng)景：多語(yǔ)言模型（如mBERT的部分實(shí)現(xiàn)）。處理拼寫錯(cuò)誤或非標(biāo)準(zhǔn)文本的任務(wù)。
示例：文本: "I love" → Tokens: ["I", " ", "l", "o", "v", "e"]
03 基于子詞的分詞
原理：將文本分解為介于單詞和字符之間的子詞單元，常見(jiàn)算法包括BPE、WordPiece和Unigram LM。子詞通常是高頻詞或詞片段。
實(shí)現(xiàn)：通過(guò)統(tǒng)計(jì)或優(yōu)化算法構(gòu)建詞匯表，動(dòng)態(tài)分割文本，保留常見(jiàn)詞并拆分稀有詞。
優(yōu)點(diǎn)：平衡了詞匯表大小和未登錄詞處理能力。能處理新詞、拼寫變體和多語(yǔ)言文本。token具有一定語(yǔ)義，序列長(zhǎng)度適中。
缺點(diǎn)：分詞結(jié)果可能不直觀（如"playing"拆為"play" + "##ing"）。需要預(yù)訓(xùn)練分詞器，增加前期成本。
常見(jiàn)子詞算法
01 Byte-Pair Encoding (BPE)
原理：從字符開(kāi)始，迭代合并高頻字符對(duì)，形成子詞。
應(yīng)用：GPT系列、RoBERTa。
示例："lowest" → ["low", "##est"]。
02 WordPiece
原理：類似BPE，但基于最大化語(yǔ)言模型似然選擇合并。
應(yīng)用：BERT、Electra。
示例："unhappiness" → ["un", "##hap", "##pi", "##ness"]。
03 Unigram Language Model
原理：通過(guò)語(yǔ)言模型優(yōu)化選擇最優(yōu)子詞集合，允許多種分割路徑。
應(yīng)用：T5、ALBERT
應(yīng)用場(chǎng)景：幾乎所有現(xiàn)代大模型（如BERT、GPT、T5）。多語(yǔ)言、通用NLP任務(wù)。
示例：文本: "unhappiness" → Tokens: ["un", "##hap", "##pi", "##ness"]
04 基于SentencePiece的分詞
原理：一種無(wú)監(jiān)督的分詞方法，將文本視為字符序列，直接學(xué)習(xí)子詞分割，不依賴語(yǔ)言特定的預(yù)處理（如空格分割）。支持BPE或Unigram LM算法。
實(shí)現(xiàn)：訓(xùn)練一個(gè)模型（.model文件），包含詞匯表和分詞規(guī)則，直接對(duì)原始文本編碼/解碼。
優(yōu)點(diǎn)：語(yǔ)言無(wú)關(guān)，適合多語(yǔ)言和無(wú)空格語(yǔ)言（如中文、日文）。統(tǒng)一處理原始文本，無(wú)需預(yù)分詞。能處理未登錄詞，靈活性高。
缺點(diǎn)：需要額外訓(xùn)練分詞模型。分詞結(jié)果可能不夠直觀。
應(yīng)用場(chǎng)景：T5、LLaMA、mBART等跨語(yǔ)言模型。中文、日文等無(wú)明確分隔的語(yǔ)言。
示例：文本: "こんにちは"（日語(yǔ)：你好） → Tokens: ["▁こ", "ん", "に", "ち", "は"]
05 基于規(guī)則的分詞
原理：根據(jù)語(yǔ)言特定的規(guī)則（如正則表達(dá)式）將文本分割為單詞或短語(yǔ)，常結(jié)合詞典或語(yǔ)法規(guī)則。
實(shí)現(xiàn)：使用工具（如Jieba for Chinese、Mecab for Japanese）或自定義規(guī)則進(jìn)行分詞。
優(yōu)點(diǎn)：分詞結(jié)果符合語(yǔ)言習(xí)慣，語(yǔ)義清晰。適合特定語(yǔ)言或領(lǐng)域（如中文分詞）。
缺點(diǎn)：依賴語(yǔ)言特定的規(guī)則和詞典，跨語(yǔ)言通用性差。維護(hù)成本高，難以處理新詞或非標(biāo)準(zhǔn)文本。
應(yīng)用場(chǎng)景：中文（Jieba、THULAC）、日文（Mecab）、韓文等分詞。特定領(lǐng)域的專業(yè)術(shù)語(yǔ)分詞。
示例：文本: "我愛(ài)編程"（中文） → Tokens: ["我", "愛(ài)", "編程"]
06 基于Byte-level Tokenization
原理：直接將文本編碼為字節(jié)序列（UTF-8編碼），每個(gè)字節(jié)作為一個(gè)token。常結(jié)合BPE（如Byte-level BPE）。
實(shí)現(xiàn)：無(wú)需預(yù)定義詞匯表，直接處理字節(jié)序列，動(dòng)態(tài)生成子詞。
優(yōu)點(diǎn)：完全語(yǔ)言無(wú)關(guān)，詞匯表極?。?56個(gè)字節(jié)）。無(wú)未登錄詞問(wèn)題，適合多語(yǔ)言和非標(biāo)準(zhǔn)文本。
缺點(diǎn)：序列長(zhǎng)度較長(zhǎng)，計(jì)算開(kāi)銷大。語(yǔ)義粒度低，模型需學(xué)習(xí)復(fù)雜模式。
應(yīng)用場(chǎng)景：GPT-3、Bloom等大規(guī)模多語(yǔ)言模型。處理原始字節(jié)輸入的任務(wù)。
示例：文本: "hello" → Tokens: ["h", "e", "l", "l", "o"]（或字節(jié)表示）。
從零實(shí)現(xiàn)BPE分詞器
子詞分詞（BPE、WordPiece、SentencePiece）是現(xiàn)代大模型的主流，因其在詞匯表大小、未登錄詞處理和序列長(zhǎng)度之間取得平衡，本次我們使用純Python，不依賴任何開(kāi)源框架來(lái)實(shí)現(xiàn)一個(gè)BPE分詞器。
我們先實(shí)現(xiàn)一個(gè)BPETokenizer類：
import json
from collections import defaultdict
import re
import 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+|[^\w\s]', 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+|[^\w\s]', 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+|[^\w\s]', 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", [])函數(shù)說(shuō)明：
__init__：初始化分詞器，創(chuàng)建詞匯表、合并規(guī)則等數(shù)據(jù)結(jié)構(gòu)。
get_stats：統(tǒng)計(jì)詞頻字典中相鄰符號(hào)對(duì)的頻率。
merge_vocab：根據(jù)符號(hào)對(duì)合并詞頻字典中的token。
train：基于語(yǔ)料庫(kù)訓(xùn)練BPE分詞器，構(gòu)建詞匯表。
encode：將文本編碼為token id序列。
decode：將token id序列解碼為文本。
save：保存分詞器狀態(tài)到指定目錄。
export_token_map：導(dǎo)出token映射到文件。
print_visualization：可視化文本的BPE分詞過(guò)程。
load：從指定路徑加載分詞器狀態(tài)。
加載測(cè)試數(shù)據(jù)進(jìn)行訓(xùn)練：
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')}")此處我選擇了開(kāi)源的數(shù)據(jù)，水滸傳全文檔進(jìn)行訓(xùn)練，請(qǐng)注意：訓(xùn)練數(shù)據(jù)應(yīng)該以章節(jié)分割，請(qǐng)根據(jù)具體上下文決定。
文章如下：
在這里要注意vocab_size值的選擇：
小語(yǔ)料測(cè)試 → vocab_size=100~500
訓(xùn)練 AI 語(yǔ)言模型前分詞器 → vocab_size=1000~30000
實(shí)際場(chǎng)景調(diào)優(yōu) → 可實(shí)驗(yàn)不同大小，看 token 數(shù)、OOV 情況等
進(jìn)行訓(xùn)練：
我們執(zhí)行完訓(xùn)練代碼后，程序會(huì)在bpe_tokenizer文件夾下生成4個(gè)文件：
vocab.json：存儲(chǔ)詞匯表，記錄每個(gè)token到其id的映射（如{"[PAD]": 0, "he": 256}）。
merges.txt：存儲(chǔ)BPE合并規(guī)則，每行是一對(duì)合并的符號(hào)（如h e表示合并為he）。
tokenizer_config.json：存儲(chǔ)分詞器配置，包括模型類型、詞匯表大小、特殊token等信息。
token_map.tsv：存儲(chǔ)token id到token的映射，每行格式為id\ttoken\ttoken的字符序列（如256\the\th e），用于調(diào)試或分析。
我們本次測(cè)試vocab_size選擇了500，我們打開(kāi)vocab.json查看，里面有500個(gè)詞：
進(jìn)行測(cè)試：
我們執(zhí)行如下代碼進(jìn)行測(cè)試：
if __name__ == '__main__':
# 加載分詞器
tokenizer = BPETokenizer()
tokenizer.load('./bpe_tokenizer')
# 測(cè)試分詞和還原
text = "且說(shuō)魯智深自離了五臺(tái)山文殊院，取路投東京來(lái)，行了半月之上。"
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: 且說(shuō)魯智深[UNK]離了[UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK]東京[UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK][UNK]
Visualization:
且說(shuō) 魯智深 自 離了 五 臺(tái) 山 文 殊 院 | ， | 取 路 投 東京 來(lái) | ， | 行 了 半 月 之 上 | 。我們看到解碼后，輸出很多[UNK]，出現(xiàn) [UNK] 并非編碼器的問(wèn)題，而是訓(xùn)練語(yǔ)料覆蓋不夠和vocab設(shè)置的值太小， 導(dǎo)致token 沒(méi)有進(jìn)入 vocab。這個(gè)到后邊我們真正訓(xùn)練時(shí)，再說(shuō)明。
BPE它是一種壓縮+分詞混合技術(shù)。初始時(shí)我們把句子分成單字符。然后統(tǒng)計(jì)出現(xiàn)頻率最高的字符對(duì)，不斷合并，直到詞表大小滿足預(yù)設(shè)。
