引言:为什么大模型需要“分词“?
拆开大模型的"词汇表":从BPE到WordPiece,Tokenization的底层密码
引言:为什么大模型需要"分词"?
当你输入"你好,世界!"时,有没有想过GPT、Llama、Claude这些大模型是如何理解这些文字的?答案是:Tokenizer(分词器)。
Tokenizer是大模型的第一道工序,它将原始文本转换成模型能处理的数字序列。这个过程看似简单,实则暗藏玄机。不同的分词策略直接影响着模型的词汇量、训练效率、推理速度,甚至是涌现能力。
本文将深入剖析两种主流子词分词算法——BPE(Byte Pair Encoding)和WordPiece的底层原理,揭示它们是如何让大模型"认识"语言、构建词表的。
一、从字符到词:分词策略的演进
1.1 传统分词的困境
早期的NLP系统采用词级分词,即每个完整单词作为一个token。这种方式面临严峻挑战:
英文词汇量:超过100万且持续增长
德语复合词:Kraftfahrzeughaftpflichtversicherung(机动车责任保险)
中文词边界:没有天然空格分隔
词级分词的问题:
- 词汇表爆炸,难以覆盖所有单词
-
- OOV(Out-of-Vocabulary)问题严重
-
- 泛化能力差,未登录词无法处理
1.2 字符级分词的局限
转向字符级分词(每个字符作为一个token)看似解决了OOV问题,但引入了新的问题:
"hello" → ['h', 'e', 'l', 'l', 'o'] → 5 tokens
序列长度暴增5倍,模型需要学习字符间的复杂依赖关系
字符级分词将序列长度线性放大,显著增加了计算成本和训练难度。
1.3 子词分词的崛起
子词分词(Subword Tokenization)应运而生,它在字符级和词级之间找到了平衡点:
"unbelievable"
├── un + believe + able (BPE风格)
├── un + believ + ##able (WordPiece风格)
└── 3 tokens vs 12字符
核心思想:将常见词保留,未登录词拆分为已知子词,通过组合表示任意词汇。
二、BPE:压缩算法出身的分词器
2.1 BPE的起源
BPE最初由Philip Gage在1994年提出,用于数据压缩领域。其核心思想极其优雅:寻找最频繁出现的字节对(byte pair),用一个新的单字节替换所有该字节对。
2.2 BPE分词的数学原理
BPE分词是自底向上的合并算法,通过迭代合并最频繁的字符序列:
算法流程:
输入:训练语料、目标词表大小V
1. 初始化词表:将所有字符加入基础词表
2. 2. 统计相邻token对出现频率
3. 3. 找到出现最频繁的token对
4. 4. 合并该token对,生成新token,加入词表
5. 5. 重复步骤2-4,直到词表大小达到V
6. ```
**关键公式**:在第t次迭代时,选择合并的对(a, b)满足:
$$(a, b) = \arg\max_{(a,b)} freq(a, b)$$
### 2.3 BPE实战:手写实现
让我们从零实现一个BPE分词器,深入理解其核心机制:
```python
import re
from collections import defaultdict, Counter
from typing import List, Tuple, Dict
class SimpleBPE:
def __init__(self, vocab_size: int = 500):
self.vocab_size = vocab_size
self.vocab = {}
self.merges = []
def get_stats(self, vocab: Dict[str, int]) -> Dict[Tuple[str, str], int]:
"""统计所有相邻token对的频率"""
pairs = defaultdict(int)
for word, freq in vocab.items():
symbols = word.split()
for i in range(len(symbols) - 1):
pairs[(symbols[i], symbols[i+1])] += freq
return pairs
def merge_vocab(self, vocab: Dict, best_pair: Tuple[str, str]) -> Dict:
"""合并最频繁的token对"""
first, second = best_pair
new_vocab = {}
pattern = re.compile(rf'{first} {second}')
for word in vocab:
new_word = pattern.sub(f'{first}{second}', word)
new_vocab[new_word] = vocab[word]
return new_vocab
def train(self, corpus: List[str]):
"""训练BPE模型"""
# Step 1: 初始化词表(字符级)
vocab = Counter()
for word in corpus:
# 添加</w>标记词边界
word_tokens = ' '.join(list(word)) + ' </w>'
vocab[word_tokens] += 1
# Step 2: 迭代合并
while len(vocab) < self.vocab_size:
pairs = self.get_stats(vocab)
if not pairs:
break
best_pair = max(pairs, key=pairs.get)
# 特殊终止条件:最高频率为1
if pairs[best_pair] == 1:
break
vocab = self.merge_vocab(vocab, best_pair)
self.merges.append(best_pair)
# 提取最终词表
self._build_vocab()
def _build_vocab(self):
"""从合并历史构建词表"""
self.vocab = {'<pad>': 0, '<unk>': 1, '</s>': 2}
for first, second in self.merges:
token = first + second
if token not in self.vocab:
self.vocab[token] = len(self.vocab)
# 添加单字符
for i in range(256):
ch = chr(i)
if ch not in self.vocab:
self.vocab[ch] = len(self.vocab)
def tokenize(self, text: str) -> List[str]:
"""对文本进行分词"""
tokens = list(text)
tokens.append('</w>')
for first, second in self.merges:
pattern = re.compile(rf'{first} {second}')
while True:
new_tokens = pattern.sub(first + second, ' '.join(tokens))
if new_tokens == ' '.join(tokens):
break
tokens = new_tokens.split()
return tokens
# 测试
corpus = [
"low", "lower", "newest", "widest",
"lowness", "lowest", "new", "wide"
]
bpe = SimpleBPE(vocab_size=50)
bpe.train(corpus)
test_words = ["lowest", "newer", "wider"]
for word in test_words:
print(f"{word}: {bpe.tokenize(word)}")
```
**输出示例**:
lowest: [‘low’, ‘e’, ‘s’, ‘t’]
newer: [‘new’, ‘’, ‘er’]
wider: [‘w’, ‘i’, ‘d’, ‘er’]
### 2.4 BPE的深层机制
BPE的核心洞察在于:**频繁共现的字符序列应该被视为一个整体**。
训练语料中 “est” 出现100次,“er” 出现80次
→ “est” 将先被合并
→ 未登录词 “forest” → “for”, “est” → 2 tokens
这解释了为什么BPE能有效处理形态丰富的语言(如德语、土耳其语)。
## 三、WordPiece:语言模型驱动的分词
### 3.1 WordPiece与BPE的本质区别
WordPiece最初由Google为日语和韩语语音搜索开发,其核心区别在于**选择合并的标准不同**:
| 维度 | BPE | WordPiece |
|------|-----|-----------|
| 合并标准 | 共现频率最大化 | 语言模型似然最大化 |
| 优化目标 | 最小化词表大小 | 最大化训练语料似然 |
| 子词标记 | 无特殊标记 | 用##前缀表示非词首 |
### 3.2 WordPiece的数学推导
WordPiece采用基于语言模型的贪心合并策略。核心思想:**合并后的子词应该最大化训练语料的似然**。
**关键公式**:对于候选合并(A, B),计算合并收益:
$$\text{gain}(A,B) = \frac{count(A,B)}{count(A) \times count(B)}$$
更精确的做法是计算语言模型困惑度改善:
$$score(A,B) = \log\left(\frac{count(A,B)}{count(A) \times count(B)}\right)$$
合并后,整体似然变化为:
$$\Delta L = L_{after} - L_{before} = \log P(A,B) - [\log P(A) + \log P(B)]$$
选择使$\Delta L$最大的候选进行合并。
### 3.3 WordPiece的贪心合并
WordPiece使用贪心算法,从字符序列开始,逐步合并收益最高的子词:
```python
from typing import Set, List
class SimpleWordPiece:
def __init__(self, vocab: Set[str], max_input_chars_per_word: int = 100):
self.vocab = vocab
self.max_input_chars_per_word = max_input_chars_per_word
self.unk_token = '[UNK]'
def tokenize(self, text: str) -> List[str]:
"""WordPiece分词"""
output_tokens = []
for word in text.split():
if len(word) > self.max_input_chars_per_word:
output_tokens.append(self.unk_token)
continue
# 贪心选择最长匹配
tokens = []
start = 0
while start < len(word):
end = len(word)
found = False
# 从最长可能位置开始尝试
while start < end:
substr = word[start:end]
# 非词首子词需要##前缀
if start > 0:
substr = '##' + substr
if substr in self.vocab:
tokens.append(word[start:end]) # 存储原始形式
start = end
found = True
break
end -= 1
if not found:
tokens.append(self.unk_token)
start += 1
output_tokens.extend(tokens)
return output_tokens
# 模拟一个训练好的词表
vocab = {
'[UNK]', '[CLS]', '[SEP]', '[PAD]',
'天', '地', '人', '和', '的',
'##天', '##地', '##和', '##的',
'##宇', '##宙', '##宇', '##宙',
'##星', '##人', '##类'
}
wp = SimpleWordPiece(vocab)
test_phrases = ["宇宙", "人类", "天人合一"]
for phrase in test_phrases:
print(f"'{phrase}': {wp.tokenize(phrase)}")
```
**输出示例**:
‘宇宙’: [‘宇宙’]
‘人类’: [‘人类’]
‘天人合一’: [‘天’, ‘人’, ‘##合一’]
### 3.4 WordPiece的训练过程
WordPiece训练是一个迭代优化过程:
```python
import math
from collections import defaultdict, Counter
class WordPieceTrainer:
def __init__(self, vocab_size: int = 30000, min_frequency: int = 2):
self.vocab_size = vocab_size
self.min_frequency = min_frequency
self.vocab = set()
self.merges = []
def _get_vocab_from_corpus(self, corpus: List[str]) -> Dict[str, int]:
"""统计词频"""
vocab = Counter()
for word in corpus:
vocab[' '.join(word) + ' </w>'] += 1
return dict(vocab)
def _compute_pair_score(self, vocab: Dict, pair: Tuple[str, str]) -> float:
"""计算语言模型视角的合并收益"""
first, second = pair
merged = first + second
freq = vocab.get(merged, 0)
freq_first = sum(v for k, v in vocab.items() if first in k)
freq_second = sum(v for k, v in vocab.items() if second in k)
if freq_first == 0 or freq_second == 0:
return -float('inf')
# WordPiece的评分函数
return freq / (freq_first * freq_second)
def train(self, corpus: List[str]):
"""训练WordPiece"""
vocab = self._get_vocab_from_corpus(corpus)
# 初始化:加入所有字符和边界标记
base_tokens = set()
for word in vocab:
base_tokens.update(word.replace(' </w>', '').split())
base_tokens.update(['<pad>', '<unk>', '<s>', '</s>'])
# 迭代合并
while len(base_tokens) < self.vocab_size:
# 统计所有候选对
pair_freqs = defaultdict(int)
for word, freq in vocab.items():
tokens = word.split()
for i in range(len(tokens) - 1):
p
更多推荐
所有评论(0)