Address as_target_tokenizer deprecation in Transformers v5.

tokenization_indictrans.py

@@ -1,55 +1,54 @@
 import os
 import json
 
+from transformers.utils import logging
 from typing import Dict, List, Optional, Union, Tuple
 
-from transformers.utils import logging
 from sentencepiece import SentencePieceProcessor
 from transformers.tokenization_utils import PreTrainedTokenizer
 
 
 logger = logging.get_logger(__name__)
 
-SPIECE_UNDERLINE = "▁"
-
-SPECIAL_TAGS = {
-    "_bt_",
-    "_ft_",
-    "asm_Beng",
-    "awa_Deva",
-    "ben_Beng",
-    "bho_Deva",
-    "brx_Deva",
-    "doi_Deva",
-    "eng_Latn",
-    "gom_Deva",
-    "gon_Deva",
-    "guj_Gujr",
-    "hin_Deva",
-    "hne_Deva",
-    "kan_Knda",
-    "kas_Arab",
-    "kas_Deva",
-    "kha_Latn",
-    "lus_Latn",
-    "mag_Deva",
-    "mai_Deva",
-    "mal_Mlym",
-    "mar_Deva",
-    "mni_Beng",
-    "mni_Mtei",
-    "npi_Deva",
-    "ory_Orya",
-    "pan_Guru",
-    "san_Deva",
-    "sat_Olck",
-    "snd_Arab",
-    "snd_Deva",
-    "tam_Taml",
-    "tel_Telu",
-    "urd_Arab",
-    "unr_Deva",
-}
+# Convert LANGUAGE_TAGS to a frozen set for faster lookups
+LANGUAGE_TAGS = frozenset(
+    {
+        "asm_Beng",
+        "awa_Deva",
+        "ben_Beng",
+        "bho_Deva",
+        "brx_Deva",
+        "doi_Deva",
+        "eng_Latn",
+        "gom_Deva",
+        "gon_Deva",
+        "guj_Gujr",
+        "hin_Deva",
+        "hne_Deva",
+        "kan_Knda",
+        "kas_Arab",
+        "kas_Deva",
+        "kha_Latn",
+        "lus_Latn",
+        "mag_Deva",
+        "mai_Deva",
+        "mal_Mlym",
+        "mar_Deva",
+        "mni_Beng",
+        "mni_Mtei",
+        "npi_Deva",
+        "ory_Orya",
+        "pan_Guru",
+        "san_Deva",
+        "sat_Olck",
+        "snd_Arab",
+        "snd_Deva",
+        "tam_Taml",
+        "tel_Telu",
+        "urd_Arab",
+        "unr_Deva",
+    }
+)
 
 VOCAB_FILES_NAMES = {
     "src_vocab_fp": "dict.SRC.json",
@@ -60,9 +59,8 @@ VOCAB_FILES_NAMES = {
 
 
 class IndicTransTokenizer(PreTrainedTokenizer):
-    _added_tokens_encoder = {}
-    _added_tokens_decoder = {}
-
+    _added_tokens_encoder: Dict[str, int] = {}
+    _added_tokens_decoder: Dict[str, int] = {}
     vocab_files_names = VOCAB_FILES_NAMES
     model_input_names = ["input_ids", "attention_mask"]
 
@@ -79,47 +77,55 @@ class IndicTransTokenizer(PreTrainedTokenizer):
         do_lower_case=False,
         **kwargs,
     ):
-
-        self.src = True
-
         self.src_vocab_fp = src_vocab_fp
         self.tgt_vocab_fp = tgt_vocab_fp
         self.src_spm_fp = src_spm_fp
         self.tgt_spm_fp = tgt_spm_fp
 
-        self.unk_token = unk_token.content
-        self.pad_token = pad_token.content
-        self.eos_token = eos_token.content
-        self.bos_token = bos_token.content
+        # Store token content directly instead of accessing .content
+        self.unk_token = (
+            hasattr(unk_token, "content") and unk_token.content or unk_token
+        )
+        self.pad_token = (
+            hasattr(pad_token, "content") and pad_token.content or pad_token
+        )
+        self.eos_token = (
+            hasattr(eos_token, "content") and eos_token.content or eos_token
+        )
+        self.bos_token = (
+            hasattr(bos_token, "content") and bos_token.content or bos_token
+        )
+
+        # Load vocabularies
+        self.src_encoder = self._load_json(self.src_vocab_fp)
+        self.tgt_encoder = self._load_json(self.tgt_vocab_fp)
 
-        self.encoder = self._load_json(self.src_vocab_fp)
-        if self.unk_token not in self.encoder:
+        # Validate tokens
+        if self.unk_token not in self.src_encoder:
             raise KeyError("<unk> token must be in vocab")
-        assert self.pad_token in self.encoder
-        self.encoder_rev = {v: k for k, v in self.encoder.items()}
+        if self.pad_token not in self.src_encoder:
+            raise KeyError("<pad> token must be in vocab")
 
-        self.decoder = self._load_json(self.tgt_vocab_fp)
-        if self.unk_token not in self.encoder:
-            raise KeyError("<unk> token must be in vocab")
-        assert self.pad_token in self.encoder
-        self.decoder_rev = {v: k for k, v in self.decoder.items()}
+        # Pre-compute reverse mappings
+        self.src_decoder = {v: k for k, v in self.src_encoder.items()}
+        self.tgt_decoder = {v: k for k, v in self.tgt_encoder.items()}
 
-        # load SentencePiece model for pre-processing
+        # Load SPM models
         self.src_spm = self._load_spm(self.src_spm_fp)
         self.tgt_spm = self._load_spm(self.tgt_spm_fp)
 
-        self.current_spm = self.src_spm
-        self.current_encoder = self.encoder
-        self.current_encoder_rev = self.encoder_rev
+        # Initialize current settings
+        self._switch_to_input_mode()
 
-        self.unk_token_id = self.encoder[self.unk_token]
-        self.pad_token_id = self.encoder[self.pad_token]
-        self.eos_token_id = self.encoder[self.eos_token]
-        self.bos_token_id = self.encoder[self.bos_token]
+        # Cache token IDs
+        self.unk_token_id = self.src_encoder[self.unk_token]
+        self.pad_token_id = self.src_encoder[self.pad_token]
+        self.eos_token_id = self.src_encoder[self.eos_token]
+        self.bos_token_id = self.src_encoder[self.bos_token]
 
         super().__init__(
             src_vocab_file=self.src_vocab_fp,
-            tgt_vocab_file=self.src_vocab_fp,
+            tgt_vocab_file=self.tgt_vocab_fp,
             do_lower_case=do_lower_case,
             unk_token=unk_token,
             bos_token=bos_token,
@@ -128,134 +134,118 @@ class IndicTransTokenizer(PreTrainedTokenizer):
             **kwargs,
         )
 
-    def add_new_special_tags(self, new_tags: List[str]):
-        SPECIAL_TAGS.update(new_tags)
+    def add_new_language_tags(self, new_tags: List[str]) -> None:
+        global LANGUAGE_TAGS
+        LANGUAGE_TAGS = frozenset(LANGUAGE_TAGS | set(new_tags))
 
-    def _switch_to_input_mode(self):
-        self.src = True
+    def _switch_to_input_mode(self) -> None:
+        self.spm = self.src_spm
         self.padding_side = "left"
-        self.current_spm = self.src_spm
-        self.current_encoder = self.encoder
-        self.current_encoder_rev = self.encoder_rev
+        self.encoder = self.src_encoder
+        self.decoder = self.src_decoder
+        self._tokenize = self._src_tokenize
 
-    def _switch_to_target_mode(self):
-        self.src = False
+    def _switch_to_target_mode(self) -> None:
+        self.spm = self.tgt_spm
         self.padding_side = "right"
-        self.current_spm = self.tgt_spm
-        self.current_encoder = self.decoder
-        self.current_encoder_rev = self.decoder_rev
+        self.encoder = self.tgt_encoder
+        self.decoder = self.tgt_decoder
+        self._tokenize = self._tgt_tokenize
 
-    def _load_spm(self, path: str) -> SentencePieceProcessor:
+    @staticmethod
+    def _load_spm(path: str) -> SentencePieceProcessor:
         return SentencePieceProcessor(model_file=path)
 
-    def _save_json(self, data, path: str) -> None:
+    @staticmethod
+    def _save_json(data: Union[Dict, List], path: str) -> None:
         with open(path, "w", encoding="utf-8") as f:
             json.dump(data, f, indent=2)
 
-    def _load_json(self, path: str) -> Union[Dict, List]:
+    @staticmethod
+    def _load_json(path: str) -> Union[Dict, List]:
         with open(path, "r", encoding="utf-8") as f:
             return json.load(f)
 
-    def _split_tags(self, tokens: List[str]) -> Tuple[List[str], List[str]]:
-        tags = [token for token in tokens if token in SPECIAL_TAGS]
-        tokens = [token for token in tokens if token not in SPECIAL_TAGS]
-        return tags, tokens
-
-    def _split_pads(self, tokens: List[str]) -> Tuple[List[str], List[str]]:
-        pads = [token for token in tokens if token == self.pad_token]
-        tokens = [token for token in tokens if token != self.pad_token]
-        return pads, tokens
-
     @property
     def src_vocab_size(self) -> int:
-        return len(self.encoder)
+        return len(self.src_encoder)
 
     @property
     def tgt_vocab_size(self) -> int:
-        return len(self.decoder)
+        return len(self.tgt_encoder)
 
     def get_src_vocab(self) -> Dict[str, int]:
-        return dict(self.encoder, **self.added_tokens_encoder)
+        return dict(self.src_encoder, **self.added_tokens_encoder)
 
     def get_tgt_vocab(self) -> Dict[str, int]:
-        return dict(self.decoder, **self.added_tokens_decoder)
+        return dict(self.tgt_encoder, **self.added_tokens_decoder)
 
-    # hack override
     def get_vocab(self) -> Dict[str, int]:
         return self.get_src_vocab()
 
-    # hack override
     @property
     def vocab_size(self) -> int:
         return self.src_vocab_size
 
     def _convert_token_to_id(self, token: str) -> int:
-        """Converts an token (str) into an index (integer) using the source/target vocabulary map."""
-        return self.current_encoder.get(token, self.current_encoder[self.unk_token])
+        return self.encoder.get(token, self.unk_token_id)
 
     def _convert_id_to_token(self, index: int) -> str:
-        """Converts an index (integer) into a token (str) using the source/target vocabulary map."""
-        return self.current_encoder_rev.get(index, self.unk_token)
+        return self.decoder.get(index, self.unk_token)
 
     def convert_tokens_to_string(self, tokens: List[str]) -> str:
-        """Uses sentencepiece model for detokenization"""
-        pads, tokens = self._split_pads(tokens)
-
-        if self.src:
+        return "".join(tokens).replace("▁", " ").strip()
 
-            tags, non_tags = self._split_tags(tokens)
+    def _src_tokenize(self, text: str) -> List[str]:
+        src_lang, tgt_lang, text = text.split(" ", 2)
+        assert src_lang in LANGUAGE_TAGS, f"Invalid source language tag: {src_lang}"
+        assert tgt_lang in LANGUAGE_TAGS, f"Invalid target language tag: {tgt_lang}"
+        return [src_lang, tgt_lang] + self.spm.EncodeAsPieces(text)
 
-            return (
-                " ".join(pads)
-                + " "
-                + " ".join(tags)
-                + " "
-                + "".join(non_tags).replace(SPIECE_UNDERLINE, " ").strip()
-            )
+    def _tgt_tokenize(self, text: str) -> List[str]:
+        return self.spm.EncodeAsPieces(text)
 
-        return (
-            "".join(tokens).replace(SPIECE_UNDERLINE, " ").strip()
-            + " "
-            + " ".join(pads)
+    def _decode(
+        self,
+        token_ids: Union[int, List[int]],
+        skip_special_tokens: bool = False,
+        clean_up_tokenization_spaces: bool = None,
+        spaces_between_special_tokens: bool = True,
+        **kwargs,
+    ) -> str:
+        self._switch_to_target_mode()
+        decoded_token_ids = super()._decode(
+            token_ids=token_ids,
+            skip_special_tokens=skip_special_tokens,
+            clean_up_tokenization_spaces=clean_up_tokenization_spaces,
+            spaces_between_special_tokens=spaces_between_special_tokens,
+            **kwargs,
         )
-
-    def _tokenize(self, text) -> List[str]:
-        if self.src:
-            tokens = text.split(" ")
-            tags, non_tags = self._split_tags(tokens)
-            text = " ".join(non_tags)
-            tokens = self.current_spm.EncodeAsPieces(text)
-            return tags + tokens
-        else:
-            return self.current_spm.EncodeAsPieces(text)
+        self._switch_to_input_mode()
+        return decoded_token_ids
 
     def build_inputs_with_special_tokens(
         self, token_ids_0: List[int], token_ids_1: Optional[List[int]] = None
     ) -> List[int]:
-        if token_ids_1 is None:
-            return token_ids_0 + [self.eos_token_id]
-        # We don't expect to process pairs, but leave the pair logic for API consistency
-        return token_ids_0 + [self.eos_token_id] + token_ids_1 + [self.eos_token_id]
+        return token_ids_0 + [self.eos_token_id]
 
     def save_vocabulary(
         self, save_directory: str, filename_prefix: Optional[str] = None
-    ) -> Tuple[str]:
+    ) -> Tuple[str, ...]:
         if not os.path.isdir(save_directory):
             logger.error(f"Vocabulary path ({save_directory}) should be a directory")
-            return
+            return ()
 
         src_spm_fp = os.path.join(save_directory, "model.SRC")
         tgt_spm_fp = os.path.join(save_directory, "model.TGT")
         src_vocab_fp = os.path.join(save_directory, "dict.SRC.json")
         tgt_vocab_fp = os.path.join(save_directory, "dict.TGT.json")
 
-        self._save_json(self.encoder, src_vocab_fp)
-        self._save_json(self.decoder, tgt_vocab_fp)
-
-        with open(src_spm_fp, "wb") as f:
-            f.write(self.src_spm.serialized_model_proto())
+        self._save_json(self.src_encoder, src_vocab_fp)
+        self._save_json(self.tgt_encoder, tgt_vocab_fp)
 
-        with open(tgt_spm_fp, "wb") as f:
-            f.write(self.tgt_spm.serialized_model_proto())
+        for fp, spm in [(src_spm_fp, self.src_spm), (tgt_spm_fp, self.tgt_spm)]:
+            with open(fp, "wb") as f:
+                f.write(spm.serialized_model_proto())
 
-        return src_vocab_fp, tgt_vocab_fp, src_spm_fp, tgt_spm_fp
+        return src_vocab_fp, tgt_vocab_fp, src_spm_fp, tgt_spm_fp