引言:为什么大模型需要“分词“?

admin2026-10-29 22:46:548252

拆开大模型的"词汇表":从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:

# 添加标记词边界

word_tokens = ' '.join(list(word)) + ' '

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 = {'': 0, '': 1, '': 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('')

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) + ' '] += 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(' ', '').split())

base_tokens.update(['', '', '', ''])

# 迭代合并

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