Upload app.py with huggingface_hub

app.py

@@ -0,0 +1,226 @@
+
+import gradio as gr
+import os
+import json
+from collections import Counter, deque
+from functools import lru_cache
+
+
+class BPETokenizerSimple:
+    def __init__(self):
+        self.vocab = {}
+        self.inverse_vocab = {}
+        self.bpe_merges = {}
+
+    def train(self, text, vocab_size, allowed_special={"<|endoftext|>"}):
+        processed_text = []
+        for i, char in enumerate(text):
+            if char == " " and i != 0:
+                processed_text.append("Ġ")
+            if char != " ":
+                processed_text.append(char)
+        processed_text = "".join(processed_text)
+
+        unique_chars = [chr(i) for i in range(256)]
+        unique_chars.extend(char for char in sorted(set(processed_text)) if char not in unique_chars)
+        if 'Ġ' not in unique_chars:
+            unique_chars.append('Ġ')
+
+        self.vocab = {i: char for i, char in enumerate(unique_chars)}
+        self.inverse_vocab = {char: i for i, char in self.vocab.items()}
+
+        if allowed_special:
+            for token in allowed_special:
+                if token not in self.inverse_vocab:
+                    new_id = len(self.vocab)
+                    self.vocab[new_id] = token
+                    self.inverse_vocab[token] = new_id
+
+        token_ids = [self.inverse_vocab[char] for char in processed_text]
+
+        for new_id in range(len(self.vocab), vocab_size):
+            if len(token_ids) < 2:
+                break
+            pair_id = self.find_freq_pair(token_ids, mode="most")
+            if pair_id is None:
+                break
+
+            updated = self.replace_pair(token_ids, pair_id, new_id)
+            if updated == token_ids:
+                break
+
+            token_ids = updated
+            self.bpe_merges[pair_id] = new_id
+
+        for (p0, p1), new_id in self.bpe_merges.items():
+            merged_token = self.vocab[p0] + self.vocab[p1]
+            self.vocab[new_id] = merged_token
+            self.inverse_vocab[merged_token] = new_id
+
+    def encode(self, text):
+        tokens = []
+        words = text.replace("\n", " \n ").split()
+
+        for i, word in enumerate(words):
+            if i > 0 and not word.startswith("\n"):
+                tokens.append("Ġ" + word)
+            else:
+                tokens.append(word)
+
+        token_ids = []
+        for token in tokens:
+            if token in self.inverse_vocab:
+                token_id = self.inverse_vocab[token]
+                token_ids.append(token_id)
+            else:
+                sub_token_ids = self.tokenize_with_bpe(token)
+                token_ids.extend(sub_token_ids)
+
+        return token_ids
+
+    def tokenize_with_bpe(self, token):
+        token_ids = [self.inverse_vocab.get(char, None) for char in token]
+        if None in token_ids:
+            missing_chars = [char for char, tid in zip(token, token_ids) if tid is None]
+            raise ValueError(f"Characters not found in vocab: {missing_chars}")
+
+        can_merge = True
+        while can_merge and len(token_ids) > 1:
+            can_merge = False
+            new_tokens = []
+            i = 0
+            while i < len(token_ids) - 1:
+                pair = (token_ids[i], token_ids[i + 1])
+                if pair in self.bpe_merges:
+                    merged_token_id = self.bpe_merges[pair]
+                    new_tokens.append(merged_token_id)
+                    i += 2
+                    can_merge = True
+                else:
+                    new_tokens.append(token_ids[i])
+                    i += 1
+            if i < len(token_ids):
+                new_tokens.append(token_ids[i])
+            token_ids = new_tokens
+
+        return token_ids
+
+    def decode(self, token_ids):
+        decoded_string = ""
+        for token_id in token_ids:
+            if token_id not in self.vocab:
+                raise ValueError(f"Token ID {token_id} not found in vocab.")
+            token = self.vocab[token_id]
+            if token.startswith("Ġ"):
+                decoded_string += " " + token[1:]
+            else:
+                decoded_string += token
+        return decoded_string
+
+    def save_tokenizer(self, file_path):
+        tokenizer_data = {
+            "vocab": {str(k): v for k, v in self.vocab.items()},
+            "inverse_vocab": self.inverse_vocab,
+            "bpe_merges": {str(k): v for k, v in self.bpe_merges.items()}
+        }
+        with open(file_path, "w", encoding="utf-8") as f:
+            json.dump(tokenizer_data, f, ensure_ascii=False, indent=4)
+
+    @classmethod
+    def load_tokenizer(cls, file_path):
+        with open(file_path, "r", encoding="utf-8") as f:
+            tokenizer_data = json.load(f)
+
+        tokenizer = cls()
+        tokenizer.vocab = {int(k): v for k, v in tokenizer_data["vocab"].items()}
+        tokenizer.inverse_vocab = tokenizer_data["inverse_vocab"]
+        tokenizer.bpe_merges = {eval(k): v for k, v in tokenizer_data["bpe_merges"].items()}
+        return tokenizer
+
+    @lru_cache(maxsize=None)
+    def get_special_token_id(self, token):
+        return self.inverse_vocab.get(token, None)
+
+    @staticmethod
+    def find_freq_pair(token_ids, mode="most"):
+        if len(token_ids) < 2:
+            return None
+        pairs = Counter(zip(token_ids, token_ids[1:]))
+        if not pairs:
+            return None
+
+        if mode == "most":
+            return max(pairs.items(), key=lambda x: x[1])[0]
+        elif mode == "least":
+            return min(pairs.items(), key=lambda x: x[1])[0]
+        else:
+            raise ValueError("Invalid mode. Choose 'most' or 'least'.")
+
+    @staticmethod
+    def replace_pair(token_ids, pair_id, new_id):
+        dq = deque(token_ids)
+        replaced = []
+
+        while dq:
+            current = dq.popleft()
+            if dq and (current, dq[0]) == pair_id:
+                replaced.append(new_id)
+                dq.popleft()
+            else:
+                replaced.append(current)
+
+        return replaced
+
+
+# Load the tokenizer
+TOKENIZER_FILE = "bpe_tokenizer_artifacts/bpe_tokenizer_simple.json"
+
+try:
+    bpe_tokenizer = BPETokenizerSimple.load_tokenizer(TOKENIZER_FILE)
+    print("Tokenizer loaded successfully!")
+except FileNotFoundError:
+    print(f"Error: Tokenizer file not found at {TOKENIZER_FILE}. Please ensure it's uploaded to the Space.")
+    # Fallback or error handling for missing tokenizer file
+    bpe_tokenizer = BPETokenizerSimple()
+    # You might want to train a dummy tokenizer or exit if the file is critical
+
+def encode_text(text):
+    if not text:
+        return ""
+    token_ids = bpe_tokenizer.encode(text)
+    return str(token_ids)
+
+def decode_ids(id_string):
+    if not id_string:
+        return ""
+    try:
+        token_ids = eval(id_string) # Be cautious with eval in production, but for simple list of ints it's fine.
+        if not isinstance(token_ids, list) or not all(isinstance(x, int) for x in token_ids):
+            return "Invalid input: Please provide a list of integers, e.g., [424, 256, 654]"
+        decoded_text = bpe_tokenizer.decode(token_ids)
+        return decoded_text
+    except Exception as e:
+        return f"Error decoding: {e}. Please provide a valid Python list of integers."
+
+
+# Gradio Interface
+with gr.Blocks() as demo:
+    gr.Markdown("# BPE Tokenizer Demo")
+    gr.Markdown("Encode text to BPE token IDs and decode token IDs back to text.")
+
+    with gr.Row():
+        text_input = gr.Textbox(lines=5, label="Input Text for Encoding")
+        encoded_output = gr.Textbox(lines=5, label="Encoded Token IDs")
+
+    encode_button = gr.Button("Encode Text")
+    encode_button.click(encode_text, inputs=text_input, outputs=encoded_output)
+
+    with gr.Row():
+        ids_input = gr.Textbox(lines=5, label="Input Token IDs for Decoding (e.g., [1, 2, 3])")
+        decoded_output = gr.Textbox(lines=5, label="Decoded Text")
+
+    decode_button = gr.Button("Decode IDs")
+    decode_button.click(decode_ids, inputs=ids_input, outputs=decoded_output)
+
+
+demo.launch()