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