大家好,我是寫代碼的中年人。在這個人人談?wù)摗癟oken量”、“百萬上下文”、“按Token計費”的AI時代,“Tokenizer(分詞器)”這個詞頻頻出現(xiàn)在開發(fā)者和研究者的視野中。它是連接自然語言與神經(jīng)網(wǎng)絡(luò)之間的一座橋梁,是大模型運行邏輯中至關(guān)重要的一環(huán)。很多時候,你以為自己在和大模型對話,其實你和它聊的是一堆Token。
今天我們就來揭秘大模型背后的魔法之一:Tokenizer。我們不僅要搞懂什么是Tokenizer,還要了解BPE(Byte Pair Encoding)的分詞原理,最后還會帶你看看大模型是怎么進(jìn)行分詞的。我還會用代碼演示:如何訓(xùn)練你自己的Tokenizer!
注:揭秘大模型的魔法屬于連載文章,一步步帶你打造一個大模型。
Tokenizer 是什么
Tokenizer是大模型語言處理中用于將文本轉(zhuǎn)化為模型可處理的數(shù)值表示(通常是token ID序列)的關(guān)鍵組件。它負(fù)責(zé)將輸入文本分割成最小語義單元(tokens),如單詞、子詞或字符,并將其映射到對應(yīng)的ID。在大模型的世界里,模型不會直接處理我們熟悉的文本。例如,輸入:
Hello, world!
模型并不會直接理解“H”、“e”、“l(fā)”、“l(fā)”、“o”,它理解的是這些字符被轉(zhuǎn)換成的數(shù)字——準(zhǔn)確地說,是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、基于單詞的分詞
原理:將文本按空格或標(biāo)點分割為完整的單詞,每個單詞作為一個token。
實現(xiàn):通常結(jié)合詞匯表,將單詞映射到ID。未在詞匯表中的詞被標(biāo)記為[UNK](未知)。
優(yōu)點:簡單直觀,token具有明確的語義。適合英語等以空格分隔的語言。
缺點:詞匯表可能很大(幾十萬到百萬),增加了模型的參數(shù)和內(nèi)存。未登錄詞(OOV)問題嚴(yán)重,如新詞、拼寫錯誤無法處理。對中文等無明顯分隔的語言不適用。
應(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序列長,增加模型計算負(fù)擔(dān)(如Transformer的注意力機制)。丟失單詞級語義,模型需學(xué)習(xí)更復(fù)雜的上下文關(guān)系。
應(yīng)用場景:多語言模型(如mBERT的部分實現(xiàn))。處理拼寫錯誤或非標(biāo)準(zhǔ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ī)則(如正則表達(dá)式)將文本分割為單詞或短語,常結(jié)合詞典或語法規(guī)則。
實現(xiàn):使用工具(如Jieba for Chinese、Mecab for Japanese)或自定義規(guī)則進(jìn)行分詞。
優(yōu)點:分詞結(jié)果符合語言習(xí)慣,語義清晰。適合特定語言或領(lǐng)域(如中文分詞)。
缺點:依賴語言特定的規(guī)則和詞典,跨語言通用性差。維護(hù)成本高,難以處理新詞或非標(biāo)準(zhǔn)文本。
應(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é))。無未登錄詞問題,適合多語言和非標(biāo)準(zhǔn)文本。
缺點:序列長度較長,計算開銷大。語義粒度低,模型需學(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+|[^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", [])
函數(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ù)進(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')}")
此處我選擇了開源的數(shù)據(jù),水滸傳全文檔進(jìn)行訓(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 情況等
進(jìn)行訓(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的映射,每行格式為idttokenttoken的字符序列(如256theth e),用于調(diào)試或分析。
我們本次測試vocab_size選擇了500,我們打開vocab.json查看,里面有500個詞:
進(jìn)行測試:
我們執(zhí)行如下代碼進(jìn)行測試:
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 沒有進(jìn)入 vocab。這個到后邊我們真正訓(xùn)練時,再說明。
BPE它是一種壓縮+分詞混合技術(shù)。初始時我們把句子分成單字符。然后統(tǒng)計出現(xiàn)頻率最高的字符對,不斷合并,直到詞表大小滿足預(yù)設(shè)。
經(jīng)過本章學(xué)習(xí),我們可以創(chuàng)建一個簡單的分詞器,你可以復(fù)制代碼到編譯器執(zhí)行測試,如果需要文檔,也可以關(guān)注公眾號后,加我微信,我會發(fā)給你。下一章我們將進(jìn)入Transformer架構(gòu)的詳解。