分词原理与工程实践：BPE/SentencePiece/WordPiece

1. 将所有词拆为字符序列，词尾加 </w>
   "low" → ['l', 'o', 'w', '</w>']
2. 统计相邻字符对频率
3. 合并最高频字符对为新 token
4. 重复步骤 2-3 直到达到目标词表大小

from collections import Counter, defaultdict

def train_bpe(corpus, vocab_size):
    """BPE 训练算法"""
    # 初始化：每个词拆为字符
    word_freqs = Counter(corpus.split())
    splits = {word: list(word) + ['</w>'] for word in word_freqs}
    
    merges = []
    vocab = set(char for word in word_freqs for char in list(word) + ['</w>'])
    
    while len(vocab) < vocab_size:
        # 统计相邻对频率
        pair_freqs = defaultdict(int)
        for word, freq in word_freqs.items():
            split = splits[word]
            for i in range(len(split) - 1):
                pair_freqs[(split[i], split[i+1])] += freq
        
        if not pair_freqs:
            break
        
        # 合并最高频对
        best_pair = max(pair_freqs, key=pair_freqs.get)
        merges.append(best_pair)
        new_token = best_pair[0] + best_pair[1]
        vocab.add(new_token)
        
        # 更新所有词的分词
        for word in word_freqs:
            split = splits[word]
            i = 0
            while i < len(split) - 1:
                if (split[i], split[i+1]) == best_pair:
                    split = split[:i] + [new_token] + split[i+2:]
                else:
                    i += 1
            splits[word] = split
    
    return vocab, merges

def bpe_encode(text, merges):
    """BPE 编码"""
    tokens = list(text) + ['</w>']
    for pair in merges:
        i = 0
        while i < len(tokens) - 1:
            if (tokens[i], tokens[i+1]) == pair:
                tokens = tokens[:i] + [tokens[i]+tokens[i+1]] + tokens[i+2:]
            else:
                i += 1
    return tokens

"playing" → ["play", "##ing"]
"unhappiness" → ["un", "##happ", "##iness"]

import sentencepiece as spm

# 训练
spm.SentencePieceTrainer.train(
    input='corpus.txt',
    model_prefix='tokenizer',
    vocab_size=32000,
    model_type='bpe',  # 或 'unigram'
    character_coverage=0.9995,
    normalization_rule_name='nmt_nfkc',
)

# 加载使用
sp = spm.SentencePieceProcessor(model_file='tokenizer.model')
tokens = sp.encode('你好世界', out_type=str)
ids = sp.encode('你好世界', out_type=int)
print(tokens)  # ['▁你好', '世界']
print(ids)     # [12345, 6789]

import tiktoken

enc = tiktoken.get_encoding("cl100k_base")
tokens = enc.encode("Hello, 你好世界！")
print(tokens)        # [9906, 11, 220, 57668, 53901, 31663, 8346]
print(len(tokens))   # 7

<\|.*?\|>|'s|'t|'re|'ve|'m|'ll|'d|[\p{L}]+|[\p{N}]+|[^\s\p{L}\p{N}]+

# 同一段话的 token 成本差异
text_zh = "人工智能是计算机科学的一个分支"
text_en = "Artificial intelligence is a branch of computer science"

enc_gpt4 = tiktoken.get_encoding("cl100k_base")
print(f"中文 tokens: {len(enc_gpt4.encode(text_zh))}")  # ~18
print(f"英文 tokens: {len(enc_gpt4.encode(text_en))}")  # ~8

# 使用 HuggingFace tokenizers 库
from tokenizers import Tokenizer
from tokenizers.models import BPE
from tokenizers.trainers import BpeTrainer
from tokenizers.pre_tokenizers import ByteLevel

tokenizer = Tokenizer(BPE(unk_token="<unk>"))
tokenizer.pre_tokenizer = ByteLevel(add_prefix_space=True)

trainer = BpeTrainer(
    vocab_size=50000,
    special_tokens=["<unk>", "<s>", "</s>", "<pad>", "<mask>"],
    initial_alphabet=ByteLevel.alphabet(),
)

tokenizer.train(files=["corpus.txt"], trainer=trainer)
tokenizer.save("custom-tokenizer.json")