Le tokenizer BPE.
Fichier : packages/minilab/tokenizer.py
Tout commence par le texte. Avant de parler de modèle, il faut transformer le roman de Jules Verne en une suite d’identifiants, puis savoir refaire le chemin inverse. Les leçons 01 et 02 découpaient caractère par caractère ; la leçon 13 a montré comment les vrais modèles fusionnent les morceaux fréquents avec le BPE ; la leçon 16, comment un caractère peut être coupé entre deux tokens. Dans cette première étape, vous écrivez votre propre tokenizer, celui que toutes les étapes suivantes utiliseront.
Comprendre l’architecture du fichier
Le fichier transforme du texte en identifiants, et inversement. Son entrée est le corpus, environ 154 000 caractères ; sa sortie, une liste ordonnée de fusions, sauvegardée dans checkpoints/minilab/tokenizer.json. Avec cette liste, n’importe quel texte peut être encodé, puis décodé à l’identique.
Trois choix de conception structurent le fichier. D’abord, on part des 256 octets plutôt que des caractères : l’encodage UTF-8 écrit tout texte avec eux, donc un emoji ou un caractère chinois ne seront jamais inconnus. Ensuite, le vocabulaire compte 1 024 tokens : 256 octets, 766 fusions et 2 tokens spéciaux. C’est assez pour réduire nettement les séquences, et assez petit pour que la table d’embeddings du modèle reste légère. Enfin, les tokens spéciaux <|im_start|> et <|im_end|> ne sont reconnus que sur demande : un utilisateur qui les tape obtient du texte ordinaire, pas le vrai marqueur de fin de tour.
Construire pas à pas : constantes et pré-tokenisation
Le premier bloc fixe les chemins du projet : ROOT remonte de deux dossiers depuis le fichier, pour trouver le corpus et le dossier des checkpoints, quel que soit l’endroit d’où vous lancez la commande. Il déclare aussi les tokens spéciaux et la taille du vocabulaire.
Vient ensuite la pré-tokenisation de la leçon 13 : une expression régulière découpe le texte en mots précédés de leur espace, en groupes d’au plus trois chiffres et en ponctuation. Les fusions ne franchiront jamais ces frontières : sans elles, une fin de mot et le début du suivant finiraient par fusionner en un token absurde.
"""Étape 1 — un tokenizer BPE au niveau des octets, avec des tokens spéciaux (leçons 01, 02, 13 et 16).
Le vocabulaire part des 256 octets possibles : tout texte UTF-8 s'écrit avec eux, doncaucun caractère n'est jamais inconnu. L'entraînement fusionne ensuite, encore et encore,la paire de tokens voisins la plus fréquente. L'encodage rejoue ces fusions dans l'ordreoù elles ont été apprises. Deux tokens spéciaux, jamais produits à partir d'un texteordinaire, encadrent les tours d'une conversation (leçon 12) : <|im_start|> et <|im_end|>.
uv run python -m packages.minilab.tokenizer"""
import codecsimport itertoolsimport jsonimport refrom collections import Counter, defaultdictfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]CORPUS = ROOT / "packages" / "minillm" / "data" / "verne_tour_du_monde.txt"CHECKPOINTS = ROOT / "checkpoints" / "minilab"TOKENIZER_FILE = CHECKPOINTS / "tokenizer.json"
IM_START, IM_END = "<|im_start|>", "<|im_end|>"SPECIAL_TOKENS = (IM_START, IM_END)VOCAB_SIZE = 1024 # 256 octets + 766 fusions + 2 tokens spéciaux
# Pré-tokenisation (leçon 13) : un mot avec l'espace qui le précède, un groupe d'au plus# trois chiffres, une suite de ponctuation, ou des espaces. Les fusions ne franchissent# jamais ces frontières.PIECES = re.compile(r" ?[^\W\d_]+| ?\d{1,3}| ?[^\s\w]+|\s+(?!\S)|\s+")
Pair = tuple[int, int]
def pre_tokenize(text: str) -> list[str]: return PIECES.findall(text)
def read_corpus() -> str: return CORPUS.read_text(encoding="utf-8")Construire pas à pas : l’entraînement du BPE
train_bpe applique l’algorithme de la leçon 13 : compter toutes les paires de tokens voisins, fusionner la plus fréquente, recommencer. Deux astuces le rendent rapide. On compte chaque morceau distinct une seule fois, avec son nombre d’occurrences : « ␣de » apparaît des milliers de fois, mais n’est traité qu’une fois. Et l’on retient, pour chaque paire, les morceaux qui la contiennent : après une fusion, on ne recompte que ceux-là. L’entraînement prend ainsi moins d’une seconde.
En cas d’égalité entre deux paires, la plus petite gagne. Ce détail rend l’entraînement déterministe : vous obtiendrez exactement les mêmes fusions que dans ce cours. merge, enfin, remplace une paire par son nouveau token, de gauche à droite.
def train_bpe(text: str, vocab_size: int = VOCAB_SIZE) -> list[Pair]: """Apprend vocab_size - 256 - 2 fusions, la paire la plus fréquente d'abord.""" n_merges = vocab_size - 256 - len(SPECIAL_TOKENS) if n_merges < 0: raise ValueError("Le vocabulaire doit contenir au moins 258 tokens.") # Chaque morceau distinct n'est gardé qu'une fois, avec son nombre d'occurrences. counts = Counter(pre_tokenize(text)) words = [list(piece.encode("utf-8")) for piece in counts] freqs = list(counts.values()) pair_counts: Counter[Pair] = Counter() where: dict[Pair, set[int]] = defaultdict(set) # quels mots contiennent chaque paire for index, word in enumerate(words): for pair in itertools.pairwise(word): pair_counts[pair] += freqs[index] where[pair].add(index) merges: list[Pair] = [] for new_id in range(256, 256 + n_merges): if not pair_counts: break # La paire la plus fréquente ; en cas d'égalité, la plus petite paire gagne, # pour que l'entraînement donne toujours le même résultat. best = min(pair_counts, key=lambda pair: (-pair_counts[pair], pair)) if pair_counts[best] < 1: break merges.append(best) for index in list(where[best]): word, freq = words[index], freqs[index] for pair in itertools.pairwise(word): # on retire les anciennes paires… pair_counts[pair] -= freq if pair_counts[pair] <= 0: del pair_counts[pair] words[index] = word = merge(word, best, new_id) for pair in itertools.pairwise(word): # …puis on compte les nouvelles pair_counts[pair] += freq where[pair].add(index) del where[best] pair_counts.pop(best, None) return merges
def merge(ids: list[int], pair: Pair, new_id: int) -> list[int]: """Remplace chaque occurrence de pair par new_id, de gauche à droite.""" out: list[int] = [] i = 0 while i < len(ids): if i + 1 < len(ids) and (ids[i], ids[i + 1]) == pair: out.append(new_id) i += 2 else: out.append(ids[i]) i += 1 return outConstruire pas à pas : la classe Tokenizer
La classe construit le vocabulaire à partir des fusions : chaque token fusionné vaut les octets de ses deux parties, donc le décodage revient simplement à recoller des octets. Pour encoder un morceau, on rejoue les fusions dans l’ordre où elles ont été apprises : tant qu’une paire présente est connue, on fusionne celle de plus petit rang. Un petit cache évite de réencoder les mots déjà vus.
get_tokenizer est le point d’entrée des autres étapes : il relit le tokenizer sauvegardé, ou l’entraîne une fois si le fichier n’existe pas encore.
class Tokenizer: def __init__(self, merges: list[Pair]) -> None: self.merges = merges self.ranks = {pair: 256 + rank for rank, pair in enumerate(merges)} # Les octets de chaque token : un token fusionné vaut les octets de ses deux parties. self.vocab: dict[int, bytes] = {i: bytes([i]) for i in range(256)} for pair, new_id in self.ranks.items(): self.vocab[new_id] = self.vocab[pair[0]] + self.vocab[pair[1]] first_special = 256 + len(merges) self.special = {name: first_special + i for i, name in enumerate(SPECIAL_TOKENS)} for name, token_id in self.special.items(): self.vocab[token_id] = name.encode("utf-8") self._cache: dict[str, list[int]] = {}
@property def vocab_size(self) -> int: return len(self.vocab)
def _encode_piece(self, piece: str) -> list[int]: if piece not in self._cache: ids = list(piece.encode("utf-8")) while len(ids) > 1: # On rejoue la fusion apprise en premier parmi les paires présentes. pair = min(itertools.pairwise(ids), key=lambda p: self.ranks.get(p, 1e9)) if pair not in self.ranks: break ids = merge(ids, pair, self.ranks[pair]) self._cache[piece] = ids return self._cache[piece]
def encode(self, text: str, allow_special: bool = False) -> list[int]: """Du texte aux identifiants. Les tokens spéciaux ne sont reconnus que si allow_special est vrai : un utilisateur qui tape « <|im_end|> » obtient du texte ordinaire, pas le vrai token.""" parts = [text] if allow_special: pattern = "(" + "|".join(re.escape(name) for name in SPECIAL_TOKENS) + ")" parts = re.split(pattern, text) ids: list[int] = [] for part in parts: if part in self.special and allow_special: ids.append(self.special[part]) else: for piece in pre_tokenize(part): ids.extend(self._encode_piece(piece)) return ids
def decode(self, ids: list[int]) -> str: return b"".join(self.vocab[i] for i in ids).decode("utf-8", errors="replace")
def save(self, path: Path = TOKENIZER_FILE) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps({"merges": self.merges, "special": list(SPECIAL_TOKENS)}))
@classmethod def load(cls, path: Path = TOKENIZER_FILE) -> "Tokenizer": data = json.loads(path.read_text()) return cls([tuple(pair) for pair in data["merges"]])
def get_tokenizer() -> Tokenizer: """Le tokenizer du projet : relu depuis le checkpoint, ou entraîné une fois sur Jules Verne.""" if TOKENIZER_FILE.exists(): return Tokenizer.load() tokenizer = Tokenizer(train_bpe(read_corpus())) tokenizer.save() return tokenizerConstruire pas à pas : le décodage au fil de l’eau
Quand le serveur de l’étape 6 enverra la réponse token par token, il devra afficher chaque morceau dès qu’il arrive. Mais un token peut ne contenir qu’une partie d’un caractère : l’emoji « 🙂 » s’écrit avec 4 octets. StreamDecoder utilise le décodeur UTF-8 incrémental de Python, qui garde les octets incomplets en mémoire et attend la suite au lieu d’afficher un caractère cassé. La fonction main rassemble enfin toutes les vérifications à afficher dans le terminal.
class StreamDecoder: """Détokenise token par token (leçon 16). Un caractère de plusieurs octets peut être coupé entre deux tokens : le décodeur UTF-8 incrémental garde les premiers octets et attend la suite, au lieu d'afficher un caractère cassé."""
def __init__(self, tokenizer: Tokenizer) -> None: self.tokenizer = tokenizer self.decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
def feed(self, token_id: int) -> str: return self.decoder.decode(self.tokenizer.vocab[token_id])
def flush(self) -> str: return self.decoder.decode(b"", final=True)
def main() -> None: import time
text = read_corpus() start = time.perf_counter() tokenizer = Tokenizer(train_bpe(text)) seconds = time.perf_counter() - start tokenizer.save() ids = tokenizer.encode(text) n_bytes = len(text.encode("utf-8")) print( f"Tokenizer BPE entraîné en {seconds:.1f} s sur {len(text):,} caractères".replace(",", " ") ) print( f" vocabulaire : {tokenizer.vocab_size} tokens (256 octets + {len(tokenizer.merges)} fusions + 2 spéciaux)" ) print(f" corpus : {n_bytes:,} octets → {len(ids):,} tokens".replace(",", " ")) print(f" compression : {n_bytes / len(ids):.2f} octets par token") print(f" aller-retour decode(encode(corpus)) == corpus : {tokenizer.decode(ids) == text}") print( "\nPremières fusions :", ", ".join(repr(tokenizer.vocab[256 + i].decode("utf-8", "replace")) for i in range(8)), ) sample = "Phileas Fogg parie vingt mille livres." pieces = [tokenizer.vocab[i].decode("utf-8", "replace") for i in tokenizer.encode(sample)] print(f"\n« {sample} » → {len(pieces)} tokens : {' | '.join(pieces)}") chat = f"{IM_START}user\nBonjour{IM_END}" print( f"\nTokens spéciaux : {tokenizer.encode(chat, allow_special=True)[:3]}… ({IM_START} = {tokenizer.special[IM_START]}, {IM_END} = {tokenizer.special[IM_END]})" ) print( f" sans allow_special, « {IM_END} » reste du texte : {len(tokenizer.encode(IM_END))} tokens" ) emoji = "Bravo 🙂 !" stream = StreamDecoder(tokenizer) print(f"\nDécodage au fil de l'eau de « {emoji} » :") for token_id in tokenizer.encode(emoji): out = stream.feed(token_id) print( f" token {token_id:4d} = {tokenizer.vocab[token_id]!r:<14} → {'« ' + out + ' »' if out else '(attend la suite du caractère)'}" )
if __name__ == "__main__": main()Tester et observer dans le terminal
Lancez l’étape, puis ses tests :
- Lancer l’étape
uv run python -m packages.minilab.tokenizer - Lancer les tests de l’étape
uv run pytest tests/minilab/test_tokenizer.py
Tokenizer BPE entraîné en 0.3 s sur 154 220 caractères vocabulaire : 1024 tokens (256 octets + 766 fusions + 2 spéciaux) corpus : 158 198 octets → 57 976 tokens compression : 2.73 octets par token aller-retour decode(encode(corpus)) == corpus : True Premières fusions : ' d', ' l', 'es', 'é', 'en', 'ai', ' s', ' p' « Phileas Fogg parie vingt mille livres. » → 8 tokens : Phileas | Fogg | par | ie | vingt | mille | livres | . Tokens spéciaux : [1022, 314, 273]… (<|im_start|> = 1022, <|im_end|> = 1023) sans allow_special, « <|im_end|> » reste du texte : 6 tokens Décodage au fil de l'eau de « Bravo 🙂 ! » : token 66 = b'B' → « B » token 114 = b'r' → « r » token 376 = b'av' → « av » token 111 = b'o' → « o » token 32 = b' ' → « » token 240 = b'\xf0' → (attend la suite du caractère) token 159 = b'\x9f' → (attend la suite du caractère) token 153 = b'\x99' → (attend la suite du caractère) token 130 = b'\x82' → « 🙂 » token 32 = b' ' → « » token 33 = b'!' → « ! »
Les fusions apprises en premier sont les plus fréquentes du français : « ␣d » (de, des, du…), « ␣l », « es », puis « é », écrit sur deux octets que le BPE a très vite réunis. Le corpus passe de 158 198 octets à 57 976 tokens : 2,73 octets par token. Et l’emoji montre le décodage au fil de l’eau : trois tokens sans rien afficher, puis le caractère entier.
Dans main, entraînez un vocabulaire de 300 tokens au lieu de 1 024 : Tokenizer(train_bpe(text, 300)). La compression tombe nettement, et la phrase d’exemple demande beaucoup plus de tokens. Remettez ensuite la valeur par défaut et relancez la commande : main réécrit checkpoints/minilab/tokenizer.json, et les étapes suivantes repartent du bon tokenizer.
Le tokenizer part des 256 octets, apprend 766 fusions par fréquence (en moins d’une seconde) et ajoute 2 tokens spéciaux : 1 024 tokens, 2,73 octets par token sur le corpus. L’encodage pré-tokenise, puis rejoue les fusions dans l’ordre ; le décodage recolle les octets. Les tokens spéciaux ne sont reconnus qu’avec allow_special=True, et le décodage au fil de l’eau attend les octets manquants d’un caractère.