Tokenizer 原理详解：BPE/WordPiece/Unigram/SentencePiece

输入语料: {"low": 5, "lower": 2, "newest": 6, "widest": 3}

步骤 1: 拆分为字符序列
  l o w </w> : 5
  l o w e r </w> : 2
  n e w e s t </w> : 6
  w i d e s t </w> : 3

步骤 2: 统计相邻 Token 对频率
  (l, o): 7, (o, w): 7, (e, r): 2, (n, e): 6, ...

步骤 3: 合并最高频对 (l, o) → "lo"
  lo w </w> : 5
  lo w e r </w> : 2
  n e w e s t </w> : 6
  w i d e s t </w> : 3

重复直到词汇表达到目标大小

import re
from collections import Counter, defaultdict

class BPETokenizer:
    def __init__(self):
        self.merges = []  # 训练得到的合并规则序列
        self.vocab = set()
    
    def train(self, corpus, vocab_size=1000):
        # 初始化：将每个词拆分为字符
        word_freqs = Counter()
        for text in corpus:
            for word in text.split():
                word_freqs[' '.join(list(word)) + ' </w>'] += 1
        
        # 转为字符级表示
        splits = {word: word.split() for word in word_freqs}
        
        while len(self.vocab) < vocab_size:
            # 统计相邻 pair 频率
            pair_freqs = defaultdict(int)
            for word, freq in word_freqs.items():
                symbols = splits[word]
                for i in range(len(symbols) - 1):
                    pair_freqs[(symbols[i], symbols[i+1])] += freq
            
            if not pair_freqs:
                break
            
            # 合并最高频 pair
            best_pair = max(pair_freqs, key=pair_freqs.get)
            self.merges.append(best_pair)
            new_token = best_pair[0] + best_pair[1]
            self.vocab.add(new_token)
            
            # 更新所有 splits
            for word in word_freqs:
                splits[word] = self._merge_pair(splits[word], best_pair, new_token)
    
    def _merge_pair(self, symbols, pair, new_token):
        i = 0
        result = []
        while i < len(symbols):
            if i < len(symbols) - 1 and (symbols[i], symbols[i+1]) == pair:
                result.append(new_token)
                i += 2
            else:
                result.append(symbols[i])
                i += 1
        return result
    
    def encode(self, text):
        # 对新文本按 merges 顺序逐步合并
        symbols = list(text) + ['</w>']
        for pair, new_token in [(m, m[0]+m[1]) for m in self.merges]:
            symbols = self._merge_pair(symbols, pair, new_token)
        return symbols

词汇表: ["play", "##ing", "##ed", "un", "##believ", "##able"]

输入: "unbelievable"
→ "un" + "##believ" + "##able"

def viterbi_segment(text, token_probs):
    n = len(text)
    dp = [0.0] * (n + 1)
    back = [None] * (n + 1)
    
    for i in range(1, n + 1):
        for j in range(i):
            token = text[j:i]
            if token in token_probs:
                score = dp[j] + math.log(token_probs[token])
                if score > dp[i] or dp[i] == 0:
                    dp[i] = score
                    back[i] = j
    
    # 回溯
    tokens = []
    i = n
    while i > 0:
        j = back[i]
        tokens.append(text[j:i])
        i = j
    return tokens[::-1]

import sentencepiece as spm

# 训练
spm.SentencePieceTrainer.train(
    input='corpus.txt',
    model_prefix='tokenizer',
    vocab_size=32000,
    model_type='unigram',  # 或 'bpe'
    character_coverage=0.9995,
    # 特殊 token
    pad_id=0, unk_id=1, bos_id=2, eos_id=3
)

# 使用
sp = spm.SentencePieceProcessor(model_file='tokenizer.model')
tokens = sp.encode('Hello, 世界!', out_type=str)
ids = sp.encode('Hello, 世界!', out_type=int)
decoded = sp.decode(ids)