boole_bpe_tokenizer

Custom BPE Tokenizer for Car Configurator Expressions

This repository contains a custom Byte Pair Encoding (BPE) tokenizer trained on a synthetically generated dataset of boolean expressions designed to mimic a car product configurator language.

Description

The tokenizer was developed from scratch using a custom BPE implementation. Its primary purpose is to efficiently encode and decode expressions used in a car configuration domain, where complex boolean logic and structured feature tags are prevalent. The training corpus was specifically crafted to contain repeating patterns with varied feature values to allow the BPE algorithm to learn meaningful sub-word units.

Dataset

The tokenizer was trained on a synthetic dataset of car configurator expressions. The dataset includes combinations of car models, engines, drives, colors, trims, packages, tech features, safety features, upgrades, assist systems, connected by boolean operators (AND, OR, XOR, NOT, IF...THEN...ELSE, NAND, ->) and structured tags (FEATURE, PACKAGE, LOCKS, REQUIRES, BUNDLES, PREFERS, TAG, CONTEXT, ENABLE, LOCK). The dataset contains 9750 expressions with an average length of ~494 characters.

Performance

After training, the tokenizer achieved:

• Actual Vocabulary Size: 6069 tokens
• Merges Performed: 6007
• Compression Ratio: 493.88 (characters per BPE token)

This demonstrates the tokenizer's ability to significantly reduce the sequence length compared to character-level encoding while maintaining a rich vocabulary of frequently occurring sub-word units.

How to Use

You can load and use this tokenizer directly from the Hugging Face Hub. Since this is a custom tokenizer implementation and not based on a standard transformers model, you will need to use the provided BPETokenizer class and load the vocabulary and merge files manually.

First, ensure you have the huggingface_hub library installed:

pip install huggingface_hub

Then, you can download the tokenizer files and load the tokenizer using the following Python code:

import json
from huggingface_hub import hf_hub_download

# Assume the custom BPETokenizer class is defined as in the original notebook
# (You would typically have this class available in your project)

# Define the BPETokenizer class (copy-pasted from the original notebook)
from collections import Counter, defaultdict
from dataclasses import dataclass
import math
from typing import Dict, Iterable, List, Sequence, Tuple

@dataclass
class BPETokenizer:
    target_vocab_size: int
    min_pair_frequency: int = 2

    def __post_init__(self) -> None:
        if self.target_vocab_size <= 0:
            raise ValueError("target_vocab_size must be positive")
        self.trained: bool = False
        self.token2id: Dict[str, int] = {}
        self.id2token: List[str] = []
        self.merge_rules: List[Tuple[int, int, int]] = []
        self.merge_to_token: Dict[Tuple[str, str], str] = {}
        self.merge_ranks: Dict[Tuple[str, str], int] = {}
        self.word_sequences: List[List[int]] = []
        self.word_frequencies: List[int] = []
        self.merges_completed: int = 0

    def _initialize_vocabulary(self, corpus: Sequence[str]) -> None:
        symbol_set = set()
        for sample in corpus:
            symbol_set.update(sample)
        self.id2token = sorted(symbol_set)
        self.token2id = {symbol: idx for idx, symbol in enumerate(self.id2token)}

    def _sequences_from_corpus(self, corpus: Sequence[str]) -> None:
        counts = Counter(corpus)
        self.word_sequences = []
        self.word_frequencies = []
        for expression, frequency in counts.items():
            encoded = [self.token2id[ch] for ch in expression]
            self.word_sequences.append(encoded)
            self.word_frequencies.append(frequency)

    def _get_pair_stats(self) -> Counter:
        stats: Counter = Counter()
        for sequence, freq in zip(self.word_sequences, self.word_frequencies):
            if len(sequence) < 2:
                continue
            for pair in zip(sequence, sequence[1:]):
                stats[pair] += freq
        return stats

    def _merge_pair_in_sequences(self, pair: Tuple[int, int], new_token_id: int) -> None:
        left, right = pair
        for idx, sequence in enumerate(self.word_sequences):
            if len(sequence) < 2:
                continue
            merged_sequence: List[int] = []
            i = 0
            length = len(sequence)
            while i < length:
                if i < length - 1 and sequence[i] == left and sequence[i + 1] == right:
                    merged_sequence.append(new_token_id)
                    i += 2
                else:
                    merged_sequence.append(sequence[i])
                    i += 1
            self.word_sequences[idx] = merged_sequence


    def train(self, corpus: Sequence[str]) -> None:
        if not corpus:
            raise ValueError("Corpus is empty")
        self._initialize_vocabulary(corpus)
        self._sequences_from_corpus(corpus)

        next_token_id = len(self.id2token)
        merges_target = max(self.target_vocab_size - next_token_id, 0)
        merges_completed = 0

        while merges_completed < merges_target:
            stats = self._get_pair_stats()
            if not stats:
                break
            (left, right), frequency = stats.most_common(1)[0]
            if frequency < self.min_pair_frequency:
                break

            new_symbol = self.id2token[left] + self.id2token[right]
            if new_symbol in self.token2id:
                # Skip duplicates to avoid infinite loops
                self._merge_pair_in_sequences((left, right), self.token2id[new_symbol])
                continue

            self.id2token.append(new_symbol)
            self.token2id[new_symbol] = next_token_id
            self.merge_rules.append((left, right, next_token_id))
            left_symbol = self.id2token[left]
            right_symbol = self.id2token[right]
            self.merge_to_token[(left_symbol, right_symbol)] = new_symbol
            self.merge_ranks[(left_symbol, right_symbol)] = merges_completed

            self._merge_pair_in_sequences((left, right), next_token_id)

            next_token_id += 1
            merges_completed += 1

        self.merges_completed = merges_completed
        self.trained = True

    @staticmethod
    def _get_adjacent_pairs(symbols: Sequence[str]) -> set[Tuple[str, str]]:
        return {
            (symbols[i], symbols[i + 1])
            for i in range(len(symbols) - 1)
        } if len(symbols) >= 2 else set()

    def _apply_bpe(self, symbols: List[str]) -> List[str]:
        if not symbols:
            return []
        pairs = self._get_adjacent_pairs(symbols)
        while pairs:
            candidate = min(
                pairs,
                key=lambda pair: self.merge_ranks.get(pair, math.inf),
            )
            if candidate not in self.merge_ranks:
                break
            merged_token = self.merge_to_token[candidate]
            left, right = candidate
            new_symbols: List[str] = []
            i = 0
            while i < len(symbols):
                if i < len(symbols) - 1 and symbols[i] == left and symbols[i + 1] == right:
                    new_symbols.append(merged_token)
                    i += 2
                else:
                    new_symbols.append(symbols[i])
                    i += 1
            symbols = new_symbols
            if len(symbols) == 1:
                break
            pairs = self._get_adjacent_pairs(symbols)
        return symbols


    def encode(self, text: str) -> List[int]:
        if not self.trained:
            raise RuntimeError("Tokenizer has not been trained yet")
        symbols = list(text)
        bpe_tokens = self._apply_bpe(symbols)
        return [self.token2id[token] for token in bpe_tokens]

    def decode(self, token_ids: Sequence[int]) -> str:
        if not self.trained:
            raise RuntimeError("Tokenizer has not been trained yet")
        return "".join(self.id2token[token_id] for token_id in token_ids)

    def vocabulary_size(self) -> int:
        return len(self.id2token)


# Define the repository ID
repo_id = "suniljakkaraju/boole_bpe_tokenizer"

# Download the vocabulary and merge files
vocab_path_downloaded = hf_hub_download(repo_id=repo_id, filename="vocab.json")
merges_path_downloaded = hf_hub_download(repo_id=repo_id, filename="merges.json")

# Load the vocabulary
with open(vocab_path_downloaded, "r") as f:
    token2id_loaded = json.load(f)
id2token_loaded = [token for token, token_id in sorted(token2id_loaded.items(), key=lambda item: item[1])]

# Load the merge rules from the downloaded files and convert them back to the expected format
with open(merges_path_downloaded, "r") as f:
    merges_loaded_str = json.load(f)

merge_rules_loaded = []
for rank, merge_pair_str in enumerate(merges_loaded_str):
    found_split = False
    for i in range(1, len(merge_pair_str)):
        left_str_candidate = merge_pair_str[:i]
        right_str_candidate = merge_pair_str[i:]

        if not left_str_candidate or not right_str_candidate:
            continue

        if left_str_candidate in token2id_loaded and right_str_candidate in token2id_loaded:
            left_id = token2id_loaded[left_str_candidate]
            right_id = token2id_loaded[right_str_candidate]
            merged_token_str = merge_pair_str
            if merged_token_str in token2id_loaded:
                 merged_id = token2id_loaded[merged_token_str]
                 merge_rules_loaded.append((left_id, right_id, merged_id))
                 found_split = True
                 break


# Instantiate the BPETokenizer and set its attributes
loaded_tokenizer = BPETokenizer(target_vocab_size=len(id2token_loaded))
loaded_tokenizer.token2id = token2id_loaded
loaded_tokenizer.id2token = id2token_loaded
loaded_tokenizer.merge_rules = merge_rules_loaded
loaded_tokenizer.trained = True

# Rebuild the merge_to_token and merge_ranks dictionaries for the loaded tokenizer
loaded_tokenizer.merge_to_token = {}
loaded_tokenizer.merge_ranks = {}
for rank, (left_id, right_id, merged_id) in enumerate(loaded_tokenizer.merge_rules):
    left_symbol = loaded_tokenizer.id2token[left_id]
    right_symbol = loaded_tokenizer.id2token[right_id]
    merged_symbol = loaded_tokenizer.id2token[merged_id]
    loaded_tokenizer.merge_to_token[(left_symbol, right_symbol)] = merged_symbol
    loaded_tokenizer.merge_ranks[(left_symbol, right_symbol)] = rank


# Example usage
example_text = "(FEATURE[Model=Sedan] AND FEATURE[Engine=V8] AND (FEATURE[Drive=RWD] OR FEATURE[Drive=eAWD]) AND (FEATURE[Color=Crimson] XOR FEATURE[Color=Sapphire]) AND (PACKAGE[Performance] OR PACKAGE[CitySmart]) AND LOCKS(FEATURE[Tech=SelfPark], FEATURE[Tech=HUD]) AND NOT FEATURE[Upgrade=TowPrep] AND IF FEATURE[Safety=Guardian] THEN ENABLE[Assist=RouteMind] ELSE LOCK[Mode=Season=Summer] AND (FEATURE[Market=Region=West] NAND FEATURE[Market=Season=Winter]) AND TAG[Region=West] AND CONTEXT[Season=Winter])" # Replace with your example text
encoded_output = loaded_tokenizer.encode(example_text)
decoded_output = loaded_tokenizer.decode(encoded_output)

print(f"Original text: {example_text}")
print(f"Encoded output: {encoded_output}")
print(f"Decoded output: {decoded_output}")
print(f"Decoded output matches original: {decoded_output == example_text}")