class WordTokenizer(dltk.TraditionalTokenizer):"""A simple word-level tokenizer."""def__init__(self, vocab: dict[str, int] |None=None, unk_token: str='<unk>', ):super().__init__(vocab, unk_token)@overridedef train(self, text: str|list[str], unk_token: str='<unk>') -> Self:ifisinstance(text, str): text = [text] vocab_tokens = {word for line in text for word in line.split()} vocab_tokens = [unk_token] +sorted(vocab_tokens - {unk_token}) vocab = {token: idx for idx, token inenumerate(vocab_tokens)}self.vocab = vocabself.unk_token = unk_tokenself.special_tokens = [unk_token]returnself@overridedef encode(self, text: str) ->list[int]:return [self.token_to_id(word) for word in text.split()]@overridedef decode(self, ids: list[int], skip_special_tokens: bool=True) ->str:if skip_special_tokens: special_tokens =set(self.special_tokens)else: special_tokens =set() tokens = []for i in ids: token =self.id_to_token(i)if token notin special_tokens: tokens.append(token)return' '.join(tokens)
text ='deep learning is fun deep learning is useful'tokenizer = WordTokenizer()tokenizer.train(text)print('Vocab:', tokenizer.vocab)print('Encode known words:', tokenizer.encode('deep learning is fun'))print('Encode unknown word:', tokenizer.encode('MiniGPT is fun'))
def word2symbols(word: str) -> Symbols:"""Convert a word into a tuple of symbols."""returntuple(word) + ('</w>',)
然后,我们统计 corpus 里所有相邻 token pair 的频率:
def get_pair_counts(corpus: Corpus) -> Counter[Pair]:"""Count the frequency of adjacent symbol pairs in the corpus.""" counts = Counter()for symbols, freq in corpus.items():for pair in it.pairwise(symbols): counts[pair] += freqreturn counts
例如,我们有这样一个 corpus:
corpus = { word2symbols('low'): 1, word2symbols('lower'): 1, word2symbols('lowest'): 1,}pprint(corpus)
def merge_pair(symbols: Symbols, pair: Pair) -> Symbols:"""Merge a pair of symbols into one.""" merged = [] i =0while i <len(symbols):if i <len(symbols) -1and (symbols[i], symbols[i +1]) == pair: merged.append(symbols[i] + symbols[i +1]) i +=2else: merged.append(symbols[i]) i +=1returntuple(merged)
def train_bpe(corpus: Corpus, num_merges: int, min_frequency: int=2) ->list[Pair]:"""Train BPE by merging the most frequent pairs.""" corpus =dict(corpus) merges = []for _ inrange(num_merges): pair_counts = get_pair_counts(corpus)ifnot pair_counts:break best_pair, freq = pair_counts.most_common(1)[0]if freq < min_frequency:break merges.append(best_pair) new_corpus = Counter()for symbols, count in corpus.items(): new_symbols = merge_pair(symbols, best_pair) new_corpus[new_symbols] = count corpus = new_corpusreturn merges
训练几个 merge:
words ='low lower lowest low lower newest'.split()words_freq = Counter(words)corpus = {word2symbols(w): f for w, f in words_freq.items()}merges = train_bpe(corpus, num_merges=10)print('Learned merges:')for i, pair inenumerate(merges, start=1):print(f'{i:2d}. {pair} -> {pair[0] + pair[1]}')
def merge_word(word: str, merges: list[Pair]) -> Symbols:"""Merge a word into subword tokens using learned merges.""" symbols = word2symbols(word)for pair in merges: symbols = merge_pair(symbols, pair)return symbols
看几个例子:
words = ['low', 'lower', 'lowest', 'newest', 'newer']n =max(len(word) for word in words)for word in words:print(f'{word:<{n}} -> {merge_word(word, merges)}')
\[
\text{logits} \in \mathbb{R}^{B \times T \times V}
\]
每一个 logit 都对应一个可能的下一个 token。
下面基于 BPE 的结果构造一个小词表。
def build_bpe_vocab( alphabet: set[str], merges: list[Pair], special_tokens: list[str],) ->dict[str, int]:"""Build a BPE vocabulary from the alphabet, merges, and special tokens.""" tokens =set(alphabet) tokens.update(a + b for a, b in merges) vocab_tokens = special_tokens +sorted(tokens -set(special_tokens))return {token: i for i, token inenumerate(vocab_tokens)}alphabet = {sym for word in words for sym in word2symbols(word)}special_tokens = ['<pad>', '<bos>', '<eos>', '<unk>']vocab = build_bpe_vocab(alphabet, merges, special_tokens)id_to_token = {i: token for token, i in vocab.items()}print(vocab)print('Vocab size:', len(vocab))
text ='Here is a simple example of tokenization.'char_tokenizer = CharacterTokenizer()char_tokenizer.train(text)word_tokenizer = WordTokenizer()word_tokenizer.train(text)bpe_tokenizer = BPETokenizer()bpe_tokenizer.train(text, vocab_size=20)print('Character vocab size:', char_tokenizer.vocab_size)print('Word vocab size:', word_tokenizer.vocab_size)print('BPE vocab size:', bpe_tokenizer.vocab_size)
Character vocab size: 19
Word vocab size: 8
BPE vocab size: 20
1. 影响序列长度
同一段文本,用不同 tokenizer 编码后,token 数可能差很多。
text ="I don't like tokenization."char_ids = char_tokenizer.encode(text)word_ids = word_tokenizer.encode(text)bpe_ids = bpe_tokenizer.encode(text)print('Character tokens:', len(char_ids))print('Word tokens:', len(word_ids))print('BPE tokens:', len(bpe_ids))
Character tokens: 26
Word tokens: 4
BPE tokens: 26
text ="""Machine learning models learn patterns from data. A language model reads a sequenceof tokens and predicts what token is likely to appear next. During training, the modelgradually improves its predictions by comparing them with the correct answers.Tokenization is an important step because it determines how raw text is divided intosmaller units. Some tokenizers use characters, some use complete words, and modernlanguage models often use subword tokens to balance vocabulary size and sequence length."""text = text.replace('\n', ' ').strip()tokenizer = BPETokenizer()tokenizer.train(text, vocab_size=100)token_ids = tokenizer.encode(text)print('Token ids:', token_ids[:10], '...')print('Num tokens:', len(token_ids))