


'''simple docstring'''






import logging
import os
import sys
from dataclasses import dataclass, field
from typing import List, Optional

import torch
from datasets import Dataset
from torch import nn
from tqdm.auto import tqdm

from transformers import (
    AutoModelForSequenceClassification,
    AutoTokenizer,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    set_seed,
    utils,
)
from transformers.trainer_utils import get_last_checkpoint, is_main_process


a_       :     List[str] 							=     "\nDistills an NLI-based zero-shot classifier to a smaller, more efficient model with a fixed set of candidate class\nnames. Useful for speeding up zero-shot classification in cases where labeled training data is not available, but\nwhen only a single fixed set of classes is needed. Takes a teacher NLI model, student classifier model, unlabeled\ndataset, and set of K possible class names. Yields a single classifier with K outputs corresponding to the provided\nclass names.\n"

a_       :     Dict 							=     logging.getLogger(__name__)




@dataclass
class 			a  :
 _lowerCAmelCase			       =						field(
     default="""roberta-large-mnli"""	,					metadata={"""help""": """The NLI/zero-shot teacher model to be distilled."""}							)
 _lowerCAmelCase			       =						field(
     default="""This example is {}."""	,					metadata={
         """help""": (
             """Template used to turn class names into mock hypotheses for teacher NLI model. Must include {{}}"""
             """where class name is inserted."""
         )
     }	,					)
 _lowerCAmelCase			       =						field(
     default=3_2	,					metadata={"""help""": """Batch size for generating teacher predictions."""}							)
 _lowerCAmelCase			       =						field(
     default=_SCREAMING_SNAKE_CASE	,					metadata={
         """help""": (
             """Allow multiple classes to be true rather than forcing them to sum to 1 (sometimes called"""
             """multi-class multi-label classification)."""
         )
     }	,					)
 _lowerCAmelCase			       =						field(
     default=1.0	,					metadata={"""help""": """Temperature applied to teacher softmax for distillation."""}							)





@dataclass
class 			a  :
 _lowerCAmelCase			       =						field(
     default="""distilbert-base-uncased"""	,					metadata={"""help""": """The NLI/zero-shot teacher model to be distilled."""}							)





@dataclass
class 			a  :
 _lowerCAmelCase			       =						field(metadata={"""help""": """Text file with one unlabeled instance per line."""}							)
 _lowerCAmelCase			       =						field(metadata={"""help""": """Text file with one class name per line."""}							)
 _lowerCAmelCase			       =						field(
     default=_SCREAMING_SNAKE_CASE	,					metadata={"""help""": """Whether to use one of the fast tokenizer (backed by the Rust tokenizers library) or not."""}	,					)





@dataclass
class 			a  (  _SCREAMING_SNAKE_CASE							):
 _lowerCAmelCase			       =						field(
     default=_SCREAMING_SNAKE_CASE	,					metadata={"""help""": """The output directory where the model predictions and checkpoints will be written."""}	,					)
 _lowerCAmelCase			       =						field(
     default=3_2	,					metadata={"""help""": """Batch size per GPU/TPU core/CPU for training."""}							)
 _lowerCAmelCase			       =						field(
     default=1_2_8	,					metadata={"""help""": """Batch size per GPU/TPU core/CPU for evaluation."""}							)
 _lowerCAmelCase			       =						field(default=1.0	,					metadata={"""help""": """Total number of training epochs to perform."""}							)
 _lowerCAmelCase			       =						field(default=_SCREAMING_SNAKE_CASE	,					metadata={"""help""": """Whether to run training of student model."""}							)
 _lowerCAmelCase			       =						field(
     default=_SCREAMING_SNAKE_CASE	,					metadata={
         """help""": (
             """Whether to evaluate the agreement of the final student predictions and the teacher predictions"""
             """after training."""
         )
     }	,					)
 _lowerCAmelCase			       =						field(
     default=0	,					metadata={
         """help""": (
             """Limit the total amount of checkpoints."""
             """Deletes the older checkpoints in the output_dir. Default is 0 (no checkpoints)."""
         )
     }	,					)





class 			a  (  _SCREAMING_SNAKE_CASE							):
 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__   ,					__magic_name__=False    )       ->				List[Any]:
      _a         =				inputs['labels']
      _a         =				model(inputs['input_ids']   ,					attention_mask=inputs['attention_mask']    )
      _a         =				outputs[0]

      _a         =				-torch.sum(target_p * logits.log_softmax(dim=-1    )   ,					axis=-1    ).mean()

      if return_outputs:
           return loss, outputs

      return loss






def 						_A					(lowerCAmelCase__	:Optional[int]		) ->	Dict:




     '''simple docstring'''



     _a         =				[]
     with open(lowerCAmelCase__					,    'r'		) as f:
          for line in f:
               _a         =				line.strip()
               if len(lowerCAmelCase__		) > 0:
                    lines.append(lowerCAmelCase__		)
     return lines






def 						_A					(lowerCAmelCase__	:Union[str, Any]					,    lowerCAmelCase__	:List[str]					,    lowerCAmelCase__	:List[Any]		) ->	Union[str, Any]:




     '''simple docstring'''



     _a         =				[]
     _a         =				[]
     for example in examples:
          for name in class_names:
               premises.append(lowerCAmelCase__		)
               hypotheses.append(hypothesis_template.format(lowerCAmelCase__		)		)
     return premises, hypotheses






def 						_A					(lowerCAmelCase__	:Any		) ->	List[str]:




     '''simple docstring'''



     for label, ind in config.labelaid.items():
          if label.lower().startswith('entail'		):
               return ind
     logger.warning('Could not identify entailment dimension from teacher config label2id. Setting to -1.'		)
     return -1






def 						_A					(lowerCAmelCase__	:str					,    lowerCAmelCase__	:List[str]					,    lowerCAmelCase__	:List[str]					,    lowerCAmelCase__	:str					,    lowerCAmelCase__	:int					,    lowerCAmelCase__	:float					,    lowerCAmelCase__	:bool					,    lowerCAmelCase__	:bool					,    lowerCAmelCase__	:bool					,    lowerCAmelCase__	:bool					,    ) ->	Dict:




     '''simple docstring'''



     _a         =				AutoModelForSequenceClassification.from_pretrained(lowerCAmelCase__		)
     _a         =				model.config
     if not no_cuda and torch.cuda.is_available():
          _a         =				nn.DataParallel(model.cuda()		)
          batch_size *= len(model.device_ids		)
     _a         =				AutoTokenizer.from_pretrained(lowerCAmelCase__					,    use_fast=lowerCAmelCase__		)

     _a   , _a         =				get_premise_hypothesis_pairs(lowerCAmelCase__					,    lowerCAmelCase__					,    lowerCAmelCase__		)
     _a         =				[]

     for i in tqdm(range(0					,    len(lowerCAmelCase__		)					,    lowerCAmelCase__		)		):
          _a         =				premises[i : i + batch_size]
          _a         =				hypotheses[i : i + batch_size]

          _a         =				tokenizer(
              lowerCAmelCase__					,    lowerCAmelCase__					,    padding=lowerCAmelCase__					,    truncation='only_first'					,    return_tensors='pt'					,    )

          with torch.cuda.amp.autocast(enabled=lowerCAmelCase__		):
               with torch.no_grad():
                    _a         =				model(**lowerCAmelCase__		)
          logits.append(outputs.logits.detach().cpu().float()		)

     _a         =				get_entailment_id(lowerCAmelCase__		)
     _a         =				-1 if entail_id == 0 else 0
     _a         =				torch.cat(lowerCAmelCase__					,    dim=0		)  # N*K x 3
     _a         =				logits.reshape(len(lowerCAmelCase__		)					,    len(lowerCAmelCase__		)					,    -1		)[..., [contr_id, entail_id]]  # N x K x 2

     if multi_label:
          # softmax over (contr, entail) logits for each class independently
          _a         =				(nli_logits / temperature).softmax(-1		)
     else:
          # softmax over entail logits across classes s.t. class probabilities sum to 1.
          _a         =				(nli_logits / temperature).softmax(1		)

     return nli_prob[..., 1]  # N x K






def 						_A					() ->	str:




     '''simple docstring'''



     _a         =				HfArgumentParser(
         (DataTrainingArguments, TeacherModelArguments, StudentModelArguments, DistillTrainingArguments)					,    description=lowerCAmelCase__					,    )

     if len(sys.argv		) == 2 and sys.argv[1].endswith('.json'		):
          # If we pass only one argument to the script and it's the path to a json file,
          # let's parse it to get our arguments.
          _a   , _a   , _a   , _a         =				parser.parse_json_file(
              json_file=os.path.abspath(sys.argv[1]		)		)
     else:
          _a   , _a   , _a   , _a         =				parser.parse_args_into_dataclasses()

     # Detecting last checkpoint.
     _a         =				None
     if os.path.isdir(training_args.output_dir		) and training_args.do_train and not training_args.overwrite_output_dir:
          _a         =				get_last_checkpoint(training_args.output_dir		)
          if last_checkpoint is None and len(os.listdir(training_args.output_dir		)		) > 0:
               raise ValueError(
                   f'Output directory ({training_args.output_dir}) already exists and is not empty. '
                   'Use --overwrite_output_dir to overcome.'		)
          elif last_checkpoint is not None:
               logger.info(
                   f'Checkpoint detected, resuming training at {last_checkpoint}. To avoid this behavior, change '
                   'the `--output_dir` or add `--overwrite_output_dir` to train from scratch.'		)

    # Setup logging
     logging.basicConfig(
         format='%(asctime)s - %(levelname)s - %(name)s - %(message)s'					,    datefmt='%m/%d/%Y %H:%M:%S'					,    handlers=[logging.StreamHandler(sys.stdout		)]					,    )
     logger.setLevel(logging.INFO if is_main_process(training_args.local_rank		) else logging.WARN		)

     # Log on each process the small summary:
     logger.warning(
         f'Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}'
         + f'distributed training: {bool(training_args.local_rank != -1		)}, 16-bits training: {training_args.fpaa}'		)
     # Set the verbosity to info of the Transformers logger (on main process only):
     if is_main_process(training_args.local_rank		):
          utils.logging.set_verbosity_info()
          utils.logging.enable_default_handler()
          utils.logging.enable_explicit_format()

     if training_args.local_rank != -1:
          raise ValueError('Distributed training is not currently supported.'		)
     if training_args.tpu_num_cores is not None:
          raise ValueError('TPU acceleration is not currently supported.'		)

     logger.info(f'Training/evaluation parameters {training_args}'		)

     # Set seed before initializing model.
     set_seed(training_args.seed		)

     # 1. read in data
     _a         =				read_lines(data_args.data_file		)
     _a         =				read_lines(data_args.class_names_file		)

     # 2. get teacher predictions and load into dataset
     logger.info('Generating predictions from zero-shot teacher model'		)
     _a         =				get_teacher_predictions(
         teacher_args.teacher_name_or_path					,    lowerCAmelCase__					,    lowerCAmelCase__					,    teacher_args.hypothesis_template					,    teacher_args.teacher_batch_size					,    teacher_args.temperature					,    teacher_args.multi_label					,    data_args.use_fast_tokenizer					,    training_args.no_cuda					,    training_args.fpaa					,    )
     _a         =				Dataset.from_dict(
         {
             'text': examples,
             'labels': teacher_soft_preds,
         }		)

     # 3. create student
     logger.info('Initializing student model'		)
     _a         =				AutoModelForSequenceClassification.from_pretrained(
         student_args.student_name_or_path					,    num_labels=len(lowerCAmelCase__		)		)
     _a         =				AutoTokenizer.from_pretrained(student_args.student_name_or_path					,    use_fast=data_args.use_fast_tokenizer		)
     _a         =				dict(enumerate(lowerCAmelCase__		)		)
     _a         =				{label: i for i, label in enumerate(lowerCAmelCase__		)}

     # 4. train student on teacher predictions
     _a         =				dataset.map(lowerCAmelCase__					,    input_columns='text'		)
     dataset.set_format('torch'		)

     def compute_metrics(lowerCAmelCase__	:Tuple					,    lowerCAmelCase__	:List[Any]=False		):
          _a         =				p.predictions.argmax(-1		)
          _a         =				p.label_ids.argmax(-1		)  # "label_ids" are actually distributions
          return {"agreement": (preds == proxy_labels).mean().item()}

     _a         =				DistillationTrainer(
         model=lowerCAmelCase__					,    tokenizer=lowerCAmelCase__					,    args=lowerCAmelCase__					,    train_dataset=lowerCAmelCase__					,    compute_metrics=lowerCAmelCase__					,    )

     if training_args.do_train:
          logger.info('Training student model on teacher predictions'		)
          trainer.train()

     if training_args.do_eval:
          _a         =				trainer.evaluate(eval_dataset=lowerCAmelCase__		)['eval_agreement']
          logger.info(f'Agreement of student and teacher predictions: {agreement * 1_00:0.2f}%'		)

     trainer.save_model()


if __name__ == "__main__":
 main()





'''simple docstring'''






import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


a_       :     Tuple 							=     logging.get_logger(__name__)

a_       :     Optional[Any] 							=     {
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
}

a_       :     str 							=     {
    "vocab_file": {"openai-gpt": "https://huggingface.co/openai-gpt/resolve/main/vocab.json"},
    "merges_file": {"openai-gpt": "https://huggingface.co/openai-gpt/resolve/main/merges.txt"},
}

a_       :     Union[str, Any] 							=     {
    "openai-gpt": 5_1_2,
}






def 						_A					(lowerCAmelCase__	:Any		) ->	int:




     '''simple docstring'''



     _a         =				text.strip()
     if not text:
          return []
     _a         =				text.split()
     return tokens




class 			a  (  _SCREAMING_SNAKE_CASE							):
 def __init__(			self   ,					__magic_name__=True   ,					__magic_name__=None   ,					__magic_name__=True   ,					__magic_name__=None   ,					__magic_name__=True   ,					)       ->				Optional[Any]:
      if never_split is None:
           _a         =				[]
      _a         =				do_lower_case
      _a         =				set(__magic_name__    )
      _a         =				tokenize_chinese_chars
      _a         =				strip_accents
      _a         =				do_split_on_punc
 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__=None    )       ->				List[str]:
      _a         =				self.never_split.union(set(__magic_name__    )    ) if never_split else self.never_split
      _a         =				self._clean_text(__magic_name__    )

      # This was added on November 1st, 2018 for the multilingual and Chinese
      # models. This is also applied to the English models now, but it doesn't
      # matter since the English models were not trained on any Chinese data
      # and generally don't have any Chinese data in them (there are Chinese
      # characters in the vocabulary because Wikipedia does have some Chinese
      # words in the English Wikipedia.).
      if self.tokenize_chinese_chars:
           _a         =				self._tokenize_chinese_chars(__magic_name__    )
      # prevents treating the same character with different unicode codepoints as different characters
      _a         =				unicodedata.normalize('NFC'   ,					__magic_name__    )
      _a         =				whitespace_tokenize(__magic_name__    )
      _a         =				[]
      for token in orig_tokens:
           if token not in never_split:
                if self.do_lower_case:
                     _a         =				token.lower()
                     if self.strip_accents is not False:
                          _a         =				self._run_strip_accents(__magic_name__    )
                elif self.strip_accents:
                     _a         =				self._run_strip_accents(__magic_name__    )
           split_tokens.extend(self._run_split_on_punc(__magic_name__   ,					__magic_name__    )    )

      _a         =				whitespace_tokenize(' '.join(__magic_name__    )    )
      return output_tokens
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				List[str]:
      _a         =				unicodedata.normalize('NFD'   ,					__magic_name__    )
      _a         =				[]
      for char in text:
           _a         =				unicodedata.category(__magic_name__    )
           if cat == "Mn":
                continue
           output.append(__magic_name__    )
      return "".join(__magic_name__    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__=None    )       ->				Dict:
      if not self.do_split_on_punc or (never_split is not None and text in never_split):
           return [text]
      _a         =				list(__magic_name__    )
      _a         =				0
      _a         =				True
      _a         =				[]
      while i < len(__magic_name__    ):
           _a         =				chars[i]
           if _is_punctuation(__magic_name__    ):
                output.append([char]    )
                _a         =				True
           else:
                if start_new_word:
                     output.append([]    )
                _a         =				False
                output[-1].append(__magic_name__    )
           i += 1

      return ["".join(__magic_name__    ) for x in output]
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Tuple:
      _a         =				[]
      for char in text:
           _a         =				ord(__magic_name__    )
           if self._is_chinese_char(__magic_name__    ):
                output.append(' '    )
                output.append(__magic_name__    )
                output.append(' '    )
           else:
                output.append(__magic_name__    )
      return "".join(__magic_name__    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Union[str, Any]:
      # This defines a "chinese character" as anything in the CJK Unicode block:
      #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
      #
      # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
      # despite its name. The modern Korean Hangul alphabet is a different block,
      # as is Japanese Hiragana and Katakana. Those alphabets are used to write
      # space-separated words, so they are not treated specially and handled
      # like the all of the other languages.
      if (
          (cp >= 0x4E00 and cp <= 0x9FFF)
          or (cp >= 0x3400 and cp <= 0x4DBF)  #
          or (cp >= 0x2_0000 and cp <= 0x2_A6DF)  #
          or (cp >= 0x2_A700 and cp <= 0x2_B73F)  #
          or (cp >= 0x2_B740 and cp <= 0x2_B81F)  #
          or (cp >= 0x2_B820 and cp <= 0x2_CEAF)  #
          or (cp >= 0xF900 and cp <= 0xFAFF)
          or (cp >= 0x2_F800 and cp <= 0x2_FA1F)  #
      ):  #
           return True

      return False




 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      _a         =				[]
      for char in text:
           _a         =				ord(__magic_name__    )
           if cp == 0 or cp == 0xFFFD or _is_control(__magic_name__    ):
                continue
           if _is_whitespace(__magic_name__    ):
                output.append(' '    )
           else:
                output.append(__magic_name__    )
      return "".join(__magic_name__    )






def 						_A					(lowerCAmelCase__	:str		) ->	Tuple:




     '''simple docstring'''



     _a         =				set()
     _a         =				word[0]
     for char in word[1:]:
          pairs.add((prev_char, char)		)
          _a         =				char
     return pairs






def 						_A					(lowerCAmelCase__	:Tuple		) ->	Union[str, Any]:




     '''simple docstring'''



     _a         =				text.replace('—'					,    '-'		)
     _a         =				text.replace('–'					,    '-'		)
     _a         =				text.replace('―'					,    '-'		)
     _a         =				text.replace('…'					,    '...'		)
     _a         =				text.replace('´'					,    '\''		)
     _a         =				re.sub(r'(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'					,    r' \1 '					,    lowerCAmelCase__		)
     _a         =				re.sub(r'\s*\n\s*'					,    ' \n '					,    lowerCAmelCase__		)
     _a         =				re.sub(r'[^\S\n]+'					,    ' '					,    lowerCAmelCase__		)
     return text.strip()






class 			a  (  _SCREAMING_SNAKE_CASE							):
 _lowerCAmelCase			       =						VOCAB_FILES_NAMES
 _lowerCAmelCase			       =						PRETRAINED_VOCAB_FILES_MAP
 _lowerCAmelCase			       =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowerCAmelCase			       =						["""input_ids""", """attention_mask"""]
 def __init__(			self   ,					__magic_name__   ,					__magic_name__   ,					__magic_name__="<unk>"   ,					**__magic_name__    )       ->				Any:
      super().__init__(unk_token=__magic_name__   ,					**__magic_name__    )

      try:
           import ftfy
           from spacy.lang.en import English

           _a         =				English()
           _a         =				_nlp.tokenizer
           _a         =				ftfy.fix_text
      except ImportError:
           logger.warning('ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.'    )
           _a         =				BasicTokenizer(do_lower_case=__magic_name__    )
           _a         =				None

      with open(__magic_name__   ,					encoding='utf-8'    ) as vocab_handle:
           _a         =				json.load(__magic_name__    )
      _a         =				{v: k for k, v in self.encoder.items()}
      with open(__magic_name__   ,					encoding='utf-8'    ) as merges_handle:
           _a         =				merges_handle.read().split('\n'    )[1:-1]
      _a         =				[tuple(merge.split()    ) for merge in merges]
      _a         =				dict(zip(__magic_name__   ,					range(len(__magic_name__    )    )    )    )
      _a         =				{}
 @property
 def 				__UpperCAmelCase		(			self    )       ->				Optional[int]:
      return True
 @property
 def 				__UpperCAmelCase		(			self    )       ->				int:
      return len(self.encoder    )
 def 				__UpperCAmelCase		(			self    )       ->				List[str]:
      return dict(self.encoder   ,					**self.added_tokens_encoder    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      _a         =				tuple(token[:-1]    ) + (token[-1] + '</w>',)
      if token in self.cache:
           return self.cache[token]
      _a         =				get_pairs(__magic_name__    )

      if not pairs:
           return token + "</w>"

      while True:
           _a         =				min(__magic_name__   ,					key=lambda __magic_name__    : self.bpe_ranks.get(__magic_name__   ,					float('inf'    )    )    )
           if bigram not in self.bpe_ranks:
                break
           _a   , _a         =				bigram
           _a         =				[]
           _a         =				0
           while i < len(__magic_name__    ):
                try:
                     _a         =				word.index(__magic_name__   ,					__magic_name__    )
                except ValueError:
                     new_word.extend(word[i:]    )
                     break
                else:
                     new_word.extend(word[i:j]    )
                     _a         =				j

                if word[i] == first and i < len(__magic_name__    ) - 1 and word[i + 1] == second:
                     new_word.append(first + second    )
                     i += 2
                else:
                     new_word.append(word[i]    )
                     i += 1
           _a         =				tuple(__magic_name__    )
           _a         =				new_word
           if len(__magic_name__    ) == 1:
                break
           else:
                _a         =				get_pairs(__magic_name__    )
      _a         =				' '.join(__magic_name__    )
      if word == "\n  </w>":
           _a         =				'\n</w>'
      _a         =				word
      return word
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Union[str, Any]:
      _a         =				[]
      if self.fix_text is None:
           # Using BERT's BasicTokenizer
           _a         =				self.nlp.tokenize(__magic_name__    )
           for token in text:
                split_tokens.extend(list(self.bpe(__magic_name__    ).split(' '    )    )    )
      else:
           # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
           _a         =				self.nlp(text_standardize(self.fix_text(__magic_name__    )    )    )
           for token in text:
                split_tokens.extend(list(self.bpe(token.text.lower()    ).split(' '    )    )    )
      return split_tokens
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      return self.encoder.get(__magic_name__   ,					self.encoder.get(self.unk_token    )    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      return self.decoder.get(__magic_name__   ,					self.unk_token    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Dict:
      _a         =				''.join(__magic_name__    ).replace('</w>'   ,					' '    ).strip()
      return out_string




 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__ = None    )       ->				Tuple[str]:
      if not os.path.isdir(__magic_name__    ):
           logger.error(f'Vocabulary path ({save_directory}) should be a directory'    )
           return
      _a         =				os.path.join(
          __magic_name__   ,					(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']    )
      _a         =				os.path.join(
          __magic_name__   ,					(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']    )

      with open(__magic_name__   ,					'w'   ,					encoding='utf-8'    ) as f:
           f.write(json.dumps(self.encoder   ,					indent=2   ,					sort_keys=__magic_name__   ,					ensure_ascii=__magic_name__    ) + '\n'    )

      _a         =				0
      with open(__magic_name__   ,					'w'   ,					encoding='utf-8'    ) as writer:
           writer.write('#version: 0.2\n'    )
           for bpe_tokens, token_index in sorted(self.bpe_ranks.items()   ,					key=lambda __magic_name__    : kv[1]    ):
                if index != token_index:
                     logger.warning(
                         f'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
                         ' Please check that the tokenizer is not corrupted!'    )
                     _a         =				token_index
                writer.write(' '.join(__magic_name__    ) + '\n'    )
                index += 1

      return vocab_file, merge_file

