Upload tokenization_pragya.py

tokenization_pragya.py

@@ -0,0 +1,170 @@
+import json
+import os
+
+import tiktoken
+from transformers import PreTrainedTokenizer
+
+
+def bytes_to_unicode():
+    """Reversible byte -> printable-char map (GPT-2 style). Every byte maps to
+    a unique character, so token strings never lose bytes."""
+    bs = (list(range(ord("!"), ord("~") + 1))
+          + list(range(ord("¡"), ord("¬") + 1))
+          + list(range(ord("®"), ord("ÿ") + 1)))
+    cs = bs[:]
+    n = 0
+    for b in range(256):
+        if b not in bs:
+            bs.append(b)
+            cs.append(256 + n)
+            n += 1
+    return {b: chr(c) for b, c in zip(bs, cs)}
+
+
+_BYTE_ENCODER = bytes_to_unicode()
+_BYTE_DECODER = {v: k for k, v in _BYTE_ENCODER.items()}
+
+
+class PragyaTokenizer(PreTrainedTokenizer):
+    vocab_files_names = {"tokenizer_file": "pragya.tokenizer.json"}
+
+    def __init__(self, tokenizer_file=None, bos_token="<|bos|>",
+                 eos_token="<|bos|>", pad_token="<|bos|>",
+                 unk_token="<|bos|>", **kwargs):
+        if tokenizer_file is None:
+            raise ValueError(
+                "PragyaTokenizer needs tokenizer_file=<path to pragya.tokenizer.json>, "
+                "or load it with from_pretrained()."
+            )
+        with open(tokenizer_file, "r", encoding="utf-8") as f:
+            self._raw = json.load(f)
+
+        mergeable_ranks = {bytes.fromhex(k): int(v)
+                           for k, v in self._raw["mergeable_ranks"].items()}
+        special_tokens = {k: int(v) for k, v in self._raw["special_tokens"].items()}
+        self._enc = tiktoken.Encoding(
+            name=self._raw["name"],
+            pat_str=self._raw["pat_str"],
+            mergeable_ranks=mergeable_ranks,
+            special_tokens=special_tokens,
+        )
+
+        # id <-> token string. Ordinary tokens use the reversible byte map;
+        # special tokens keep their literal string.
+        self._special_set = set(special_tokens)
+        self._id_to_tok = {}
+        for i in range(self._enc.n_vocab):
+            if i in set(special_tokens.values()):
+                continue
+            raw = self._enc.decode_single_token_bytes(i)
+            self._id_to_tok[i] = "".join(_BYTE_ENCODER[b] for b in raw)
+        for name, i in special_tokens.items():
+            self._id_to_tok[i] = name
+        self._tok_to_id = {t: i for i, t in self._id_to_tok.items()}
+
+        kwargs.setdefault("clean_up_tokenization_spaces", False)
+        kwargs.setdefault("padding_side", "left")  # generation needs left padding
+        kwargs.setdefault("model_max_length", 1024)  # CONTEXT in config__1_.py
+        super().__init__(bos_token=bos_token, eos_token=eos_token,
+                         pad_token=pad_token, unk_token=unk_token, **kwargs)
+
+    @classmethod
+    def _from_pretrained(cls, resolved_vocab_files, pretrained_model_name_or_path,
+                         *args, **kwargs):
+        # Transformers 5 can resolve our file to a default name (tokenizer.json)
+        # that we never ship. Work from any resolved file's cache directory,
+        # which also works when pretrained_model_name_or_path is a Hub repo id.
+        if resolved_vocab_files.get("tokenizer_file") is None:
+            for path in resolved_vocab_files.values():
+                if path:
+                    candidate = os.path.join(os.path.dirname(path), "pragya.tokenizer.json")
+                    if os.path.isfile(candidate):
+                        resolved_vocab_files["tokenizer_file"] = candidate
+                        break
+            else:
+                raise FileNotFoundError(
+                    f"pragya.tokenizer.json not found for {pretrained_model_name_or_path}. "
+                    "The repository must ship it alongside the other tokenizer files."
+                )
+        return super()._from_pretrained(resolved_vocab_files,
+                                        pretrained_model_name_or_path,
+                                        *args, **kwargs)
+
+    def _add_tokens(self, new_tokens, special_tokens=False):
+        # Transformers assigns ids past the end of the vocab to unknown tokens,
+        # which the model embedding cannot look up. Refuse them instead.
+        unknown = [t for t in new_tokens
+                   if str(t) not in self._tok_to_id and str(t) not in self.added_tokens_encoder]
+        if unknown:
+            raise ValueError(
+                f"Cannot add tokens {unknown!r}: this tokenizer has a fixed vocab "
+                f"of {self._enc.n_vocab} tokens and the model embedding cannot grow "
+                f"without retraining."
+            )
+        return super()._add_tokens(new_tokens, special_tokens=special_tokens)
+
+    def tokenize(self, text, **kwargs):
+        # Bypass HF's special-token splitting: the training pipeline encoded
+        # "<|bos|>" in text as ordinary characters, so match that exactly.
+        return self._tokenize(text, **kwargs)
+
+    @property
+    def vocab_size(self):
+        return self._enc.n_vocab
+
+    def get_vocab(self):
+        return {**self._tok_to_id, **self.added_tokens_encoder}
+
+    def _tokenize(self, text, **kwargs):
+        return [self._id_to_tok[i] for i in self._enc.encode_ordinary(text)]
+
+    def _convert_token_to_id(self, token):
+        if token in self._tok_to_id:
+            return self._tok_to_id[token]
+        if token in self.added_tokens_encoder:
+            return self.added_tokens_encoder[token]
+        raise ValueError(
+            f"Unknown token {token!r}. This tokenizer ships with a fixed vocab "
+            f"of {self._enc.n_vocab} tokens; new special tokens cannot be added "
+            f"without retraining the model embedding."
+        )
+
+    def build_inputs_with_special_tokens(self, token_ids_0, token_ids_1=None):
+        # Training wrapper prepends <|bos|> by default (add_bos=True).
+        ids = [self.bos_token_id] + list(token_ids_0)
+        if token_ids_1 is not None:
+            ids += list(token_ids_1)
+        return ids
+
+    def _convert_id_to_token(self, index):
+        index = int(index)
+        if index not in self._id_to_tok:
+            raise ValueError(
+                f"Token id {index} is outside the vocab of {self.vocab_size} tokens."
+            )
+        return self._id_to_tok[index]
+
+    def _decode(self, token_ids, skip_special_tokens=False, **kwargs):
+        valid = [int(i) for i in token_ids if 0 <= int(i) < self.vocab_size]
+        return super()._decode(valid, skip_special_tokens=skip_special_tokens, **kwargs)
+
+    def convert_tokens_to_string(self, tokens):
+        special = self._special_set | set(self.added_tokens_encoder)
+        out = bytearray()
+        parts = []
+        for t in tokens:
+            if t in special:
+                parts.append(out.decode("utf-8", errors="replace"))
+                out = bytearray()
+                parts.append(t)
+            else:
+                out.extend(_BYTE_DECODER[c] for c in t)
+        parts.append(out.decode("utf-8", errors="replace"))
+        return "".join(parts)
+
+    def save_vocabulary(self, save_directory, filename_prefix=None):
+        name = (filename_prefix + "-" if filename_prefix else "") + "pragya.tokenizer.json"
+        path = os.path.join(save_directory, name)
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(self._raw, f, indent=2, ensure_ascii=False)
+        return (path,)