1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
| def get_tokenizer(
vocab: dict[int, bytes],
merges: list[tuple[bytes, bytes]],
special_tokens: list[str] | None = None,
) -> Any:
"""Given a vocabulary, a list of merges, and a list of special tokens,
return a BPE tokenizer that uses the provided vocab, merges, and special tokens.
Args:
vocab (dict[int, bytes]): The tokenizer vocabulary, a mapping from int (token ID in the vocabulary)
to bytes (token bytes)
merges (list[tuple[bytes, bytes]]): BPE merges. Each list item is a tuple of bytes (<token1>, <token2>),
representing that <token1> was merged with <token2>.
Merges are ordered by order of creation.
special_tokens (list[str] | None): A list of string special tokens for the tokenizer. These strings will never
be split into multiple tokens, and will always be kept as a single token.
Returns:
A BPE tokenizer that uses the provided vocab, merges, and special tokens.
"""
bpe_tokenizer=tokenizer.BPE_Tokenizer(vocab,merges,special_tokens)
return bpe_tokenizer
def run_train_bpe(
input_path: str | os.PathLike,
vocab_size: int,
special_tokens: list[str],
**kwargs,
) -> tuple[dict[int, bytes], list[tuple[bytes, bytes]]]:
"""Given the path to an input corpus, run train a BPE tokenizer and
output its vocabulary and merges.
Args:
input_path (str | os.PathLike): Path to BPE tokenizer training data.
vocab_size (int): Total number of items in the tokenizer's vocabulary (including special tokens).
special_tokens (list[str]): A list of string special tokens to be added to the tokenizer vocabulary.
These strings will never be split into multiple tokens, and will always be
kept as a single token. If these special tokens occur in the `input_path`,
they are treated as any other string.
Returns:
tuple[dict[int, bytes], list[tuple[bytes, bytes]]]:
vocab:
The trained tokenizer vocabulary, a mapping from int (token ID in the vocabulary)
to bytes (token bytes)
merges:
BPE merges. Each list item is a tuple of bytes (<token1>, <token2>),
representing that <token1> was merged with <token2>.
Merges are ordered by order of creation.
"""
with open(input_path,"rb") as f:
text=f.read().decode("utf-8")
if special_tokens:
sorted_specials = sorted(special_tokens, key=len, reverse=True)
sp_pattern = "|".join(map(regex.escape, sorted_specials))
documents = regex.split(sp_pattern, text)
else:
documents = [text]
PAT = r"""'(?:[sdmt]|ll|ve|re)| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"""
pattern = regex.compile(PAT)
chunk_counts: dict[tuple[int, ...], int] = Counter()
for doc in documents:
if not doc:
continue
for m in pattern.finditer(doc):
token_str = m.group()
if token_str:
token_tuple = tuple(token_str.encode("utf-8"))
chunk_counts[token_tuple] += 1
vocab={i:bytes([i]) for i in range(256)}
merges=[]
for sp_token in special_tokens:
vocab[len(vocab)]=sp_token.encode("utf-8")
merge_num=vocab_size-len(vocab)
for _ in range(merge_num):
stats={}
for chunk, count in chunk_counts.items():
for i in range(len(chunk) - 1):
pair = (chunk[i], chunk[i + 1])
stats[pair] = stats.get(pair, 0) + count
if not stats:
break
pair = max(stats, key=lambda p: (stats[p], vocab[p[0]], vocab[p[1]]))
new_token_id=len(vocab)
new_chunk_counts: dict[tuple[int, ...], int] = {}
for chunk, count in chunk_counts.items():
if pair[0] not in chunk or pair[1] not in chunk:
new_chunk_counts[chunk] = new_chunk_counts.get(chunk, 0) + count
continue
new_chunk = tokenizer._merge_chunk(chunk, pair, new_token_id)
new_chunk_counts[new_chunk] = (
new_chunk_counts.get(new_chunk, 0) + count
)
chunk_counts = new_chunk_counts
pair_bytes = vocab[pair[0]]+vocab[pair[1]]
vocab[new_token_id]=pair_bytes
merges.append((vocab[pair[0]],vocab[pair[1]]))
return vocab,merges
|
