




'''simple docstring'''





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:
		_SCREAMING_SNAKE_CASE							:     Union[str, Any]   	=				None

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

_SCREAMING_SNAKE_CASE	,      _SCREAMING_SNAKE_CASE							:     List[Any]   	=				"[INST]", "[/INST]"
_SCREAMING_SNAKE_CASE	,      _SCREAMING_SNAKE_CASE							:     Dict   	=				"<<SYS>>\n", "\n<</SYS>>\n\n"

# fmt: off
_SCREAMING_SNAKE_CASE							:     Union[str, Any]   	=				"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	(			lowercase_							):
			lowerCAmelCase_  :   int		 =					VOCAB_FILES_NAMES
			lowerCAmelCase_  :   Optional[int]		 =					LlamaTokenizer
			lowerCAmelCase_  :   int		 =					"left"
			lowerCAmelCase_  :   Union[str, Any]		 =					["input_ids", "attention_mask"]




			def __init__(      self       ,							a__=None       ,							a__=None       ,							a__=False       ,							a__="<unk>"       ,							a__="<s>"       ,							a__="</s>"       ,							a__=True       ,							a__=False       ,							**a__       ,							)       ->       Any:

										'''simple docstring'''




										super().__init__(
										    vocab_file=a__       ,							tokenizer_file=a__       ,							clean_up_tokenization_spaces=a__       ,							unk_token=a__       ,							bos_token=a__       ,							eos_token=a__       ,							**a__       ,							)
										snake_case_			      =   add_bos_token
										snake_case_			      =   add_eos_token
										self.update_post_processor()

										snake_case_			      =   vocab_file
										snake_case_			      =   False if not self.vocab_file else True




			def      lowerCAmelCase__       (      self							)       ->       Tuple:

										'''simple docstring'''




										snake_case_			      =   self.bos_token
										snake_case_			      =   self.bos_token_id

										snake_case_			      =   self.eos_token
										snake_case_			      =   self.eos_token_id

										snake_case_			      =   F'{(bos+":0 ") * self.add_bos_token}$A:0{(" "+eos+":0") * self.add_eos_token}'
										snake_case_			      =   F'{single}{(" "+bos+":1") * self.add_bos_token} $B:1{(" "+eos+":1") * self.add_eos_token}'

										snake_case_			      =   []
										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_			      =   processors.TemplateProcessing(
										    single=a__       ,							pair=a__       ,							special_tokens=a__							)




			@property
			def      lowerCAmelCase__       (      self							)       ->       Optional[int]:

										'''simple docstring'''




										return self._add_eos_token




			@property
			def      lowerCAmelCase__       (      self							)       ->       List[str]:

										'''simple docstring'''




										return self._add_bos_token




			@add_eos_token.setter
			def      lowerCAmelCase__       (      self       ,							a__							)       ->       Optional[int]:

										'''simple docstring'''




										snake_case_			      =   value
										self.update_post_processor()




			@add_bos_token.setter
			def      lowerCAmelCase__       (      self       ,							a__							)       ->       List[str]:

										'''simple docstring'''




										snake_case_			      =   value
										self.update_post_processor()




			def      lowerCAmelCase__       (      self       ,							a__       ,							a__ = None							)       ->       Tuple[str]:

										'''simple docstring'''




										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(a__							):
																	logger.error(F'Vocabulary path ({save_directory}) should be a directory'							)
																	return
										snake_case_			      =   os.path.join(
										    a__       ,							(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]							)

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

										return (out_vocab_file,)




			def      lowerCAmelCase__       (      self       ,							a__							)       ->       Tuple:

										'''simple docstring'''




										snake_case_			      =   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_			      =   []
										if len(conversation.past_user_inputs							) > 0:
																	if not conversation.past_user_inputs[0].startswith(a__							) or E_SYS not in conversation.past_user_inputs[0]:
																								snake_case_			      =   (
																								    B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
																								)
										elif not dialogue[0][1].startswith(a__							) or E_SYS not in dialogue[0][1]:
																	snake_case_			      =   (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=a__							)
										        + [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=a__							)
										return dialog_tokens










'''simple docstring'''





import random
from typing import List, Optional, Union

import numpy as np
import torch

from ..state import AcceleratorState
from .constants import CUDA_DISTRIBUTED_TYPES
from .dataclasses import DistributedType, RNGType
from .imports import is_npu_available, is_tpu_available, is_xpu_available


if is_tpu_available(check_device=False):
		import torch_xla.core.xla_model as xm


def   UpperCamelCase_(			snake_case							: int							,       snake_case							: bool = False  ):

							'''simple docstring'''






							if device_specific:
														seed += AcceleratorState().process_index
							random.seed(snake_case  )
							np.random.seed(snake_case  )
							torch.manual_seed(snake_case  )
							if is_xpu_available():
														torch.xpu.manual_seed_all(snake_case  )
							elif is_npu_available():
														torch.npu.manual_seed_all(snake_case  )
							else:
														torch.cuda.manual_seed_all(snake_case  )
							# ^^ safe to call this function even if cuda is not available
							if is_tpu_available():
														xm.set_rng_state(snake_case  )


def   UpperCamelCase_(			snake_case							: Optional[RNGType] = None							,       snake_case							: Optional[torch.Generator] = None  ):

							'''simple docstring'''






							if rng_type == RNGType.TORCH:
														snake_case_			      =   torch.get_rng_state()
							elif rng_type == RNGType.CUDA:
														snake_case_			      =   torch.cuda.get_rng_state()
							elif rng_type == RNGType.XLA:
														assert is_tpu_available(), "Can't synchronize XLA seeds on an environment without TPUs."
														snake_case_			      =   torch.tensor(xm.get_rng_state()  )
							elif rng_type == RNGType.NPU:
														assert is_npu_available(), "Can't synchronize NPU seeds on an environment without NPUs."
														snake_case_			      =   torch.npu.get_rng_state()
							elif rng_type == RNGType.XPU:
														assert is_xpu_available(), "Can't synchronize XPU seeds on an environment without XPUs."
														snake_case_			      =   torch.xpu.get_rng_state()
							elif rng_type == RNGType.GENERATOR:
														assert generator is not None, "Need a generator to synchronize its seed."
														snake_case_			      =   generator.get_state()

							# Broadcast the rng state from device 0 to other devices
							snake_case_			      =   AcceleratorState()
							if state.distributed_type == DistributedType.TPU:
														snake_case_			      =   rng_state.to(xm.xla_device()  )
														xm.collective_broadcast([rng_state]  )
														xm.mark_step()
														snake_case_			      =   rng_state.cpu()
							elif (
							    state.distributed_type in CUDA_DISTRIBUTED_TYPES
							    or state.distributed_type == DistributedType.MULTI_NPU
							    or state.distributed_type == DistributedType.MULTI_XPU
							):
														snake_case_			      =   rng_state.to(state.device  )
														torch.distributed.broadcast(snake_case							,       0  )
														snake_case_			      =   rng_state.cpu()
							elif state.distributed_type == DistributedType.MULTI_CPU:
														torch.distributed.broadcast(snake_case							,       0  )

							# Set the broadcast rng state
							if rng_type == RNGType.TORCH:
														torch.set_rng_state(snake_case  )
							elif rng_type == RNGType.CUDA:
														torch.cuda.set_rng_state(snake_case  )
							elif rng_type == RNGType.NPU:
														torch.npu.set_rng_state(snake_case  )
							elif rng_type == RNGType.XPU:
														torch.xpu.set_rng_state(snake_case  )
							elif rng_type == RNGType.XLA:
														xm.set_rng_state(rng_state.item()  )
							elif rng_type == RNGType.GENERATOR:
														generator.set_state(snake_case  )




def   UpperCamelCase_(			snake_case							: List[Union[str, RNGType]]							,       snake_case							: Optional[torch.Generator] = None  ):

							'''simple docstring'''






							for rng_type in rng_types:
														synchronize_rng_state(RNGType(snake_case  )							,       generator=snake_case  )




