



import os
from shutil import copyfile
from typing import TYPE_CHECKING, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging
from ...utils.versions import require_version


if TYPE_CHECKING:
 from transformers.pipelines.conversational import Conversation

require_version("tokenizers>=0.13.3")

if is_sentencepiece_available():
 from .tokenization_llama import LlamaTokenizer
else:
 a_       :Any				  =				None

a_       :Dict				  =				logging.get_logger(__name__)
a_       :Optional[int]				  =				{"vocab_file": "tokenizer.model", "tokenizer_file": "tokenizer.json"}

a_  ,						a_       :List[str]				  =				"[INST]", "[/INST]"
a_  ,						a_       :Optional[Any]				  =				"<<SYS>>\n", "\n<</SYS>>\n\n"

# fmt: off
a_       :Tuple				  =				"You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your\\nanswers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure\\nthat your responses are socially unbiased and positive in nature.\n\nIf a question does not make any sense, or is not factually coherent, explain why instead of answering something not\\ncorrect. If you don't know the answer to a question, please don't share false information."




class  snake_case__						(	lowerCAmelCase_					):






     """simple docstring"""


     _SCREAMING_SNAKE_CASE            =     VOCAB_FILES_NAMES
     _SCREAMING_SNAKE_CASE            =     LlamaTokenizer
     _SCREAMING_SNAKE_CASE            =     """left"""
     _SCREAMING_SNAKE_CASE            =     ["""input_ids""", """attention_mask"""]






     def __init__(	self  :  Union[str, Any],			_snake_case  :  Tuple=None,			_snake_case  :  Optional[Any]=None,			_snake_case  :  Any=False,			_snake_case  :  Union[str, Any]="<unk>",			_snake_case  :  Optional[Any]="<s>",			_snake_case  :  Tuple="</s>",			_snake_case  :  int=True,			_snake_case  :  Any=False,			**_snake_case  :  Tuple,			)       ->int:
            super().__init__(
                vocab_file=_snake_case,			tokenizer_file=_snake_case,			clean_up_tokenization_spaces=_snake_case,			unk_token=_snake_case,			bos_token=_snake_case,			eos_token=_snake_case,			**_snake_case,			)
            snake_case__							:     Union[str, Any]            =     add_bos_token
            snake_case__							:     int            =     add_eos_token
            self.update_post_processor()

            snake_case__							:     Dict            =     vocab_file
            snake_case__							:     Any            =     False if not self.vocab_file else True






     def        lowercase_       (	self  :  Tuple )       ->Dict:
            snake_case__							:     List[str]            =     self.bos_token
            snake_case__							:     str            =     self.bos_token_id

            snake_case__							:     Optional[Any]            =     self.eos_token
            snake_case__							:     Optional[Any]            =     self.eos_token_id

            snake_case__							:     int            =     F'''{(bos+":0 ") * self.add_bos_token}$A:0{(" "+eos+":0") * self.add_eos_token}'''
            snake_case__							:     List[Any]            =     F'''{single}{(" "+bos+":1") * self.add_bos_token} $B:1{(" "+eos+":1") * self.add_eos_token}'''

            snake_case__							:     Optional[Any]            =     []
            if self.add_bos_token:
                   special_tokens.append((bos, bos_token_id) )
            if self.add_eos_token:
                   special_tokens.append((eos, eos_token_id) )
            snake_case__							:     Optional[int]            =     processors.TemplateProcessing(
                single=_snake_case,			pair=_snake_case,			special_tokens=_snake_case )






     @property
     def        lowercase_       (	self  :  Dict )       ->Tuple:
            return self._add_eos_token






     @property
     def        lowercase_       (	self  :  Tuple )       ->Union[str, Any]:
            return self._add_bos_token






     @add_eos_token.setter
     def        lowercase_       (	self  :  Optional[Any],			_snake_case  :  Optional[int] )       ->Optional[Any]:
            snake_case__							:     Tuple            =     value
            self.update_post_processor()






     @add_bos_token.setter
     def        lowercase_       (	self  :  Optional[Any],			_snake_case  :  Optional[int] )       ->str:
            snake_case__							:     Optional[Any]            =     value
            self.update_post_processor()






     def        lowercase_       (	self  :  List[Any],			_snake_case  :  str,			_snake_case  :  Optional[str] = None )       ->Tuple[str]:
            if not self.can_save_slow_tokenizer:
                   raise ValueError(
                       'Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '
                       'tokenizer.' )

            if not os.path.isdir(_snake_case ):
                   logger.error(F'''Vocabulary path ({save_directory}) should be a directory''' )
                   return
            snake_case__							:     Any            =     os.path.join(
                _snake_case,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file'] )

            if os.path.abspath(self.vocab_file ) != os.path.abspath(_snake_case ):
                   copyfile(self.vocab_file,			_snake_case )

            return (out_vocab_file,)






     def        lowercase_       (	self  :  Optional[Any],			_snake_case  :  "Conversation" )       ->int:
            snake_case__							:     Dict            =     list(conversation.iter_texts() )
            if not all([is_user for is_user, msg in dialogue[::2]] ) or not all(
                [not is_user for is_user, msg in dialogue[1::2]] ):
                   raise ValueError(
                       'The model only supports \'user\' and \'assistant\' roles, starting with user and alternating (u/a/u/a/u...)' )

            snake_case__							:     Tuple            =     []
            if len(conversation.past_user_inputs ) > 0:
                   if not conversation.past_user_inputs[0].startswith(_snake_case ) or E_SYS not in conversation.past_user_inputs[0]:
                          snake_case__							:     List[Any]            =     (
                              B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
                          )
            elif not dialogue[0][1].startswith(_snake_case ) or E_SYS not in dialogue[0][1]:
                   snake_case__							:     Optional[Any]            =     (dialogue[0][0], B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + dialogue[0][1])

            dialog_tokens += sum(
                [
                    [self.bos_token_id]
                    + self.encode(
                        F'''{B_INST} {(prompt[1]).strip()} {E_INST} {(answer[1]).strip()} ''',			add_special_tokens=_snake_case )
                    + [self.eos_token_id]
                    for prompt, answer in zip(dialogue[::2],			dialogue[1::2] )
                ],			[],			)
            if not (dialogue[-1][0]):
                   raise ValueError(F'''Last message must be from user, got {dialogue[-1]["role"]}''' )
            dialog_tokens += [self.bos_token_id] + self.encode(
                F'''{B_INST} {(dialogue[-1][1]).strip()} {E_INST}''',			add_special_tokens=_snake_case )
            return dialog_tokens











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_       :Dict				  =				{
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
}

a_       :List[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": 512,
}
def 	lowercase_							(A			:			Dict				):
       snake_case__							:     Optional[int]            =     text.strip()
       if not text:
              return []
       snake_case__							:     List[str]            =     text.split()
       return tokens




class  snake_case__						(	lowerCAmelCase_					):






     """simple docstring"""


     def __init__(	self  :  List[Any],			_snake_case  :  Any=True,			_snake_case  :  Optional[Any]=None,			_snake_case  :  Dict=True,			_snake_case  :  List[str]=None,			_snake_case  :  List[Any]=True,			)       ->Any:
            if never_split is None:
                   snake_case__							:     int            =     []
            snake_case__							:     Any            =     do_lower_case
            snake_case__							:     Union[str, Any]            =     set(_snake_case )
            snake_case__							:     Union[str, Any]            =     tokenize_chinese_chars
            snake_case__							:     str            =     strip_accents
            snake_case__							:     Optional[Any]            =     do_split_on_punc






     def        lowercase_       (	self  :  Dict,			_snake_case  :  Optional[int],			_snake_case  :  List[Any]=None )       ->List[str]:
            snake_case__							:     Tuple            =     self.never_split.union(set(_snake_case ) ) if never_split else self.never_split
            snake_case__							:     Optional[int]            =     self._clean_text(_snake_case )

            # 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:
                   snake_case__							:     List[str]            =     self._tokenize_chinese_chars(_snake_case )
            # prevents treating the same character with different unicode codepoints as different characters
            snake_case__							:     Optional[int]            =     unicodedata.normalize('NFC',			_snake_case )
            snake_case__							:     str            =     whitespace_tokenize(_snake_case )
            snake_case__							:     Optional[Any]            =     []
            for token in orig_tokens:
                   if token not in never_split:
                          if self.do_lower_case:
                                 snake_case__							:     int            =     token.lower()
                                 if self.strip_accents is not False:
                                        snake_case__							:     Any            =     self._run_strip_accents(_snake_case )
                          elif self.strip_accents:
                                 snake_case__							:     Optional[Any]            =     self._run_strip_accents(_snake_case )
                   split_tokens.extend(self._run_split_on_punc(_snake_case,			_snake_case ) )

            snake_case__							:     Dict            =     whitespace_tokenize(' '.join(_snake_case ) )
            return output_tokens






     def        lowercase_       (	self  :  int,			_snake_case  :  List[str] )       ->int:
            snake_case__							:     Dict            =     unicodedata.normalize('NFD',			_snake_case )
            snake_case__							:     Dict            =     []
            for char in text:
                   snake_case__							:     List[Any]            =     unicodedata.category(_snake_case )
                   if cat == "Mn":
                          continue
                   output.append(_snake_case )
            return "".join(_snake_case )






     def        lowercase_       (	self  :  str,			_snake_case  :  Dict,			_snake_case  :  Optional[int]=None )       ->Dict:
            if not self.do_split_on_punc or (never_split is not None and text in never_split):
                   return [text]
            snake_case__							:     Dict            =     list(_snake_case )
            snake_case__							:     List[str]            =     0
            snake_case__							:     Union[str, Any]            =     True
            snake_case__							:     Tuple            =     []
            while i < len(_snake_case ):
                   snake_case__							:     Any            =     chars[i]
                   if _is_punctuation(_snake_case ):
                          output.append([char] )
                          snake_case__							:     int            =     True
                   else:
                          if start_new_word:
                                 output.append([] )
                          snake_case__							:     Union[str, Any]            =     False
                          output[-1].append(_snake_case )
                   i += 1

            return ["".join(_snake_case ) for x in output]






     def        lowercase_       (	self  :  Any,			_snake_case  :  int )       ->Any:
            snake_case__							:     Dict            =     []
            for char in text:
                   snake_case__							:     Union[str, Any]            =     ord(_snake_case )
                   if self._is_chinese_char(_snake_case ):
                          output.append(' ' )
                          output.append(_snake_case )
                          output.append(' ' )
                   else:
                          output.append(_snake_case )
            return "".join(_snake_case )






     def        lowercase_       (	self  :  Any,			_snake_case  :  List[str] )       ->Dict:
            # 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 >= 0X4_E_0_0 and cp <= 0X9_F_F_F)
                or (cp >= 0X3_4_0_0 and cp <= 0X4_D_B_F)  #
                or (cp >= 0X2_0_0_0_0 and cp <= 0X2_A_6_D_F)  #
                or (cp >= 0X2_A_7_0_0 and cp <= 0X2_B_7_3_F)  #
                or (cp >= 0X2_B_7_4_0 and cp <= 0X2_B_8_1_F)  #
                or (cp >= 0X2_B_8_2_0 and cp <= 0X2_C_E_A_F)  #
                or (cp >= 0XF_9_0_0 and cp <= 0XF_A_F_F)
                or (cp >= 0X2_F_8_0_0 and cp <= 0X2_F_A_1_F)  #
            ):  #
                   return True

            return False






     def        lowercase_       (	self  :  List[str],			_snake_case  :  str )       ->Any:
            snake_case__							:     Optional[int]            =     []
            for char in text:
                   snake_case__							:     Any            =     ord(_snake_case )
                   if cp == 0 or cp == 0XF_F_F_D or _is_control(_snake_case ):
                          continue
                   if _is_whitespace(_snake_case ):
                          output.append(' ' )
                   else:
                          output.append(_snake_case )
            return "".join(_snake_case )


def 	lowercase_							(A			:			Dict				):
       snake_case__							:     Dict            =     set()
       snake_case__							:     List[str]            =     word[0]
       for char in word[1:]:
              pairs.add((prev_char, char)				)
              snake_case__							:     str            =     char
       return pairs
def 	lowercase_							(A			:			int				):
       snake_case__							:     Union[str, Any]            =     text.replace('—'		, '-'				)
       snake_case__							:     List[str]            =     text.replace('–'		, '-'				)
       snake_case__							:     List[str]            =     text.replace('―'		, '-'				)
       snake_case__							:     Any            =     text.replace('…'		, '...'				)
       snake_case__							:     Union[str, Any]            =     text.replace('´'		, '\''				)
       snake_case__							:     str            =     re.sub(r'(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'		, r' \1 '		, A				)
       snake_case__							:     int            =     re.sub(r'\s*\n\s*'		, ' \n '		, A				)
       snake_case__							:     Dict            =     re.sub(r'[^\S\n]+'		, ' '		, A				)
       return text.strip()





class  snake_case__						(	lowerCAmelCase_					):






     """simple docstring"""


     _SCREAMING_SNAKE_CASE            =     VOCAB_FILES_NAMES
     _SCREAMING_SNAKE_CASE            =     PRETRAINED_VOCAB_FILES_MAP
     _SCREAMING_SNAKE_CASE            =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _SCREAMING_SNAKE_CASE            =     ["""input_ids""", """attention_mask"""]






     def __init__(	self  :  Optional[Any],			_snake_case  :  Union[str, Any],			_snake_case  :  Tuple,			_snake_case  :  Optional[int]="<unk>",			**_snake_case  :  int )       ->Optional[Any]:
            super().__init__(unk_token=_snake_case,			**_snake_case )

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

                   snake_case__							:     List[Any]            =     English()
                   snake_case__							:     List[Any]            =     _nlp.tokenizer
                   snake_case__							:     List[Any]            =     ftfy.fix_text
            except ImportError:
                   logger.warning('ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.' )
                   snake_case__							:     Union[str, Any]            =     BasicTokenizer(do_lower_case=_snake_case )
                   snake_case__							:     Optional[int]            =     None

            with open(_snake_case,			encoding='utf-8' ) as vocab_handle:
                   snake_case__							:     Union[str, Any]            =     json.load(_snake_case )
            snake_case__							:     List[str]            =     {v: k for k, v in self.encoder.items()}
            with open(_snake_case,			encoding='utf-8' ) as merges_handle:
                   snake_case__							:     int            =     merges_handle.read().split('\n' )[1:-1]
            snake_case__							:     Dict            =     [tuple(merge.split() ) for merge in merges]
            snake_case__							:     Optional[Any]            =     dict(zip(_snake_case,			range(len(_snake_case ) ) ) )
            snake_case__							:     int            =     {}






     @property
     def        lowercase_       (	self  :  Any )       ->str:
            return True






     @property
     def        lowercase_       (	self  :  Dict )       ->str:
            return len(self.encoder )






     def        lowercase_       (	self  :  Dict )       ->Optional[int]:
            return dict(self.encoder,			**self.added_tokens_encoder )






     def        lowercase_       (	self  :  List[Any],			_snake_case  :  Union[str, Any] )       ->Any:
            snake_case__							:     Tuple            =     tuple(token[:-1] ) + (token[-1] + '</w>',)
            if token in self.cache:
                   return self.cache[token]
            snake_case__							:     List[Any]            =     get_pairs(_snake_case )

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

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

                          if word[i] == first and i < len(_snake_case ) - 1 and word[i + 1] == second:
                                 new_word.append(first + second )
                                 i += 2
                          else:
                                 new_word.append(word[i] )
                                 i += 1
                   snake_case__							:     Optional[int]            =     tuple(_snake_case )
                   snake_case__							:     Union[str, Any]            =     new_word
                   if len(_snake_case ) == 1:
                          break
                   else:
                          snake_case__							:     List[Any]            =     get_pairs(_snake_case )
            snake_case__							:     List[Any]            =     ' '.join(_snake_case )
            if word == "\n  </w>":
                   snake_case__							:     Any            =     '\n</w>'
            snake_case__							:     Optional[int]            =     word
            return word






     def        lowercase_       (	self  :  Union[str, Any],			_snake_case  :  Optional[int] )       ->Dict:
            snake_case__							:     Optional[Any]            =     []
            if self.fix_text is None:
                   # Using BERT's BasicTokenizer
                   snake_case__							:     List[str]            =     self.nlp.tokenize(_snake_case )
                   for token in text:
                          split_tokens.extend(list(self.bpe(_snake_case ).split(' ' ) ) )
            else:
                   # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
                   snake_case__							:     Optional[Any]            =     self.nlp(text_standardize(self.fix_text(_snake_case ) ) )
                   for token in text:
                          split_tokens.extend(list(self.bpe(token.text.lower() ).split(' ' ) ) )
            return split_tokens






     def        lowercase_       (	self  :  Tuple,			_snake_case  :  Union[str, Any] )       ->List[Any]:
            return self.encoder.get(_snake_case,			self.encoder.get(self.unk_token ) )






     def        lowercase_       (	self  :  Tuple,			_snake_case  :  Dict )       ->Any:
            return self.decoder.get(_snake_case,			self.unk_token )






     def        lowercase_       (	self  :  str,			_snake_case  :  List[Any] )       ->List[Any]:
            snake_case__							:     Dict            =     ''.join(_snake_case ).replace('</w>',			' ' ).strip()
            return out_string






     def        lowercase_       (	self  :  Dict,			_snake_case  :  str,			_snake_case  :  Optional[str] = None )       ->Tuple[str]:
            if not os.path.isdir(_snake_case ):
                   logger.error(F'''Vocabulary path ({save_directory}) should be a directory''' )
                   return
            snake_case__							:     Union[str, Any]            =     os.path.join(
                _snake_case,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file'] )
            snake_case__							:     List[Any]            =     os.path.join(
                _snake_case,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file'] )

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

            snake_case__							:     Union[str, Any]            =     0
            with open(_snake_case,			'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 _snake_case : 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!' )
                                 snake_case__							:     Dict            =     token_index
                          writer.write(' '.join(_snake_case ) + '\n' )
                          index += 1

            return vocab_file, merge_file






