Counts pre-token frequencies
- Reached from
train.main- Used by, in the pipeline
tokenizer.train_bpe_on_text- Used by, in tests
tests.test_tokenizer.test_count_words_splits_words_spaces_punctuation- Mathematics
- Pre-tokens and word counts
Byte-pair-encoding (BPE) tokenizer, working on characters
10 of 10 functions, classes and methods here are reachable from train.main, and 6 from talk.main.
| Pipeline stage | Step | Explained on |
|---|---|---|
| Text to training windows | Tokenizer | tokenization |
recheckpoint.py, data.py, tests/test_checkpoint.py, tests/test_tokenizer.py, train.pycheckpoint.py, data.py, talk.py, train.pytests/test_checkpoint.py, tests/test_tokenizer.py| Concept | What this file does | Lines | Status |
|---|---|---|---|
| Pre-tokens and word counts | Computes it | 8, 9 | Implemented |
| Applying a merge | Computes it, Calls it | 16, 17, 57, 70 | Implemented |
| Pair frequency and the best pair | Computes it | 35, 37, 38, 46, 48 | Implemented |
| Encode and decode | Computes it, Calls it | 70, 89, 95, 96 | Implemented |
| The vocabulary | Computes it | 102, 105, 110 | Implemented |
Counts pre-token frequencies
train.maintokenizer.train_bpe_on_texttests.test_tokenizer.test_count_words_splits_words_spaces_punctuationReplaces adjacent pair occurrences with merged, scanning left to right
train.main, talk.maintokenizer.apply_merges, tokenizer.train_bpetests.test_tokenizer.test_merge_tokens_overlap_is_left_to_right, tests.test_tokenizer.test_merge_tokens_simpleLearns merges; returns dicts with pair, merged, frequency
train.maintokenizer.train_bpe_on_texttokenizer.merge_tokensPre-tokenizes text and replays all merges on each pre-token
train.main, talk.maintokenizer.Tokenizer.encodetokenizer.merge_tokensRuns train_bpe(count_words(text), max_merges)
train.maintokenizer.build_tokenizertests.test_tokenizer.test_first_three_merges_match_hand_tracetokenizer.count_words, tokenizer.train_bpeclass TokenizerLines 79-96GitHub: tokenizer.TokenizerHolds merges and vocabulary; encodes and decodes
train.main, talk.maincheckpoint.load_checkpoint, tokenizer.build_tokenizerStores its arguments and sets vocab_size = len(itos)
train.main, talk.maincheckpoint.load_checkpoint, tokenizer.build_tokenizerText to ids; tokens not in the vocabulary map to the <unk> id
train.main, talk.maindata.encode_corpus, talk.maintests.test_checkpoint.test_loaded_model_gives_identical_logits, tests.test_checkpoint.test_non_ascii_characters_survive, tests.test_checkpoint.test_tokenizer_round_trips, tests.test_tokenizer.test_no_characters_dropped, tests.test_tokenizer.test_round_trip, tests.test_tokenizer.test_unseen_character_becomes_unktokenizer.apply_mergesIds to text by joining itos entries
train.main, talk.maintalk.main, train.maintests.test_checkpoint.test_non_ascii_characters_survive, tests.test_checkpoint.test_tokenizer_round_trips, tests.test_tokenizer.test_round_tripTrains merges, builds itos (<unk>, then characters, then merged tokens) and stoi
train.maintrain.maintests.test_checkpoint.saved, tests.test_checkpoint.test_non_ascii_characters_survive, tests.test_tokenizer.toktokenizer.Tokenizer, tokenizer.Tokenizer.__init__, tokenizer.train_bpe_on_textimport reUNK = "<unk>"PRE_TOKEN_RE = re.compile(r"\w+|\s|[^\w\s]")def count_words(text): counts = {} for word in PRE_TOKEN_RE.findall(text): counts[word] = counts.get(word, 0) + 1 return countsdef merge_tokens(tokens, pair, merged): result = [] i = 0 while i < len(tokens): if i < len(tokens) - 1 and tokens[i] == pair[0] and tokens[i + 1] == pair[1]: result.append(merged) i += 2 else: result.append(tokens[i]) i += 1 return resultdef train_bpe(word_freqs, max_merges=1000): word_splits = {} for word in word_freqs: word_splits[word] = list(word) merges = [] for step in range(max_merges): pair_freq = {} for word in word_splits: tokens = word_splits[word] weight = word_freqs[word] for i in range(len(tokens) - 1): key = (tokens[i], tokens[i + 1]) pair_freq[key] = pair_freq.get(key, 0) + weight if len(pair_freq) == 0: break best_pair = None best_count = 0 for pair in pair_freq: if pair_freq[pair] > best_count: best_count = pair_freq[pair] best_pair = pair if best_pair is None: break a, b = best_pair merged = a + b for word in word_splits: word_splits[word] = merge_tokens(word_splits[word], best_pair, merged) merges.append({"pair": best_pair, "merged": merged, "frequency": best_count}) return mergesdef apply_merges(text, merges): result = [] for pre_token in PRE_TOKEN_RE.findall(text): tokens = list(pre_token) for m in merges: tokens = merge_tokens(tokens, m["pair"], m["merged"]) result.extend(tokens) return resultdef train_bpe_on_text(text, max_merges=1000): return train_bpe(count_words(text), max_merges)class Tokenizer: def __init__(self, merges, stoi, itos): self.merges = merges self.stoi = stoi self.itos = itos self.vocab_size = len(itos) def encode(self, text): ids = [] for token in apply_merges(text, self.merges): ids.append(self.stoi.get(token, self.stoi[UNK])) return ids def decode(self, ids): tokens = [] for i in ids: tokens.append(self.itos[i]) return "".join(tokens)def build_tokenizer(text, max_merges=1000): merges = train_bpe_on_text(text, max_merges) itos = [UNK] seen = {UNK} for ch in text: if ch not in seen: seen.add(ch) itos.append(ch) for m in merges: if m["merged"] not in seen: seen.add(m["merged"]) itos.append(m["merged"]) stoi = {} for i in range(len(itos)): stoi[itos[i]] = i return Tokenizer(merges, stoi, itos)