
'''simple docstring'''
import os
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ... import AutoTokenizer
from ...utils import logging


A__ :		Union[str, Any]					='block_records.npy'


A__ :		str					=logging.get_logger(__name__)







def 							A_  (    __SCREAMING_SNAKE_CASE    :	str      ,  __SCREAMING_SNAKE_CASE    :	int			)					->							np.ndarray:





  """simple docstring"""

  import tensorflow.compat.va as tf

  __A		:  List[str]       				=		tf.data.TFRecordDataset(__a      ,  buffer_size=512 * 1024 * 1024			)
  __A		:  Optional[int]       				=		blocks_dataset.batch(__a      ,  drop_remainder=__a			)
  __A		:  Tuple       				=		next(blocks_dataset.take(1			).as_numpy_iterator()			)

  return np_record







class 			__A					:
 def __init__( self      :     Dict					, lowerCamelCase      :     Union[str, Any]					, lowerCamelCase      :     Union[str, Any]					, lowerCamelCase      :     Any=2					, lowerCamelCase      :     List[str]=10_00					, lowerCamelCase      :     Any=1_00					, lowerCamelCase      :     List[Any]=10_00_00					, ):



   """simple docstring"""





   from scann.scann_ops.py.scann_ops_pybind import builder as Builder

   __A		:  int       				=		Builder(db=lowerCAmelCase_					, num_neighbors=lowerCAmelCase_					, distance_measure="""dot_product"""							)
   __A		:  Optional[int]       				=		builder.tree(
       num_leaves=lowerCAmelCase_					, num_leaves_to_search=lowerCAmelCase_					, training_sample_size=lowerCAmelCase_							)
   __A		:  Optional[int]       				=		builder.score_ah(dimensions_per_block=lowerCAmelCase_							)

   __A		:  int       				=		builder.build()






 def 		lowercase_( self      :     Optional[Any]					, lowerCamelCase      :     Union[str, Any]							):



   """simple docstring"""





   __A				,       __A		:  Union[str, Any]       				=		self.searcher.search_batched(question_projection.detach().cpu()							)
   return retrieved_block_ids.astype("""int64"""							)







class 			__A					:
 def __init__( self      :     Any					, lowerCamelCase      :     Optional[Any]					, lowerCamelCase      :     Union[str, Any]							):



   """simple docstring"""





   super().__init__()
   __A		:  List[str]       				=		block_records
   __A		:  List[Any]       				=		tokenizer
 def __call__( self      :     List[Any]					, lowerCamelCase      :     Tuple					, lowerCamelCase      :     str					, lowerCamelCase      :     List[Any]					, lowerCamelCase      :     Optional[Any]=None					, lowerCamelCase      :     Tuple="pt"							):



   """simple docstring"""





   __A		:  List[str]       				=		np.take(self.block_records					, indices=lowerCAmelCase_					, axis=0							)

   __A		:  str       				=		self.tokenizer.decode(question_input_ids[0]					, skip_special_tokens=lowerCAmelCase_							)

   __A		:  int       				=		[]
   __A		:  int       				=		[]
   for retrieved_block in retrieved_blocks:
     text.append(lowerCAmelCase_							)
     text_pair.append(retrieved_block.decode()							)

   __A		:  Optional[int]       				=		self.tokenizer(
       lowerCAmelCase_					, lowerCAmelCase_					, padding=lowerCAmelCase_					, truncation=lowerCAmelCase_					, return_special_tokens_mask=lowerCAmelCase_					, max_length=lowerCAmelCase_							)
   __A		:  int       				=		concat_inputs.convert_to_tensors(lowerCAmelCase_							)

   if answer_ids is not None:
     return self.block_has_answer(lowerCAmelCase_					, lowerCAmelCase_							) + (concat_inputs_tensors,)
   else:
     return (None, None, None, concat_inputs_tensors)
 @classmethod
 def 		lowercase_( cls      :     Tuple					, lowerCamelCase      :     Optional[int]					, *lowerCamelCase      :     List[str]					, **lowerCamelCase      :     Optional[int]							):



   """simple docstring"""





   if os.path.isdir(lowerCAmelCase_							):
     __A		:  Optional[int]       				=		os.path.join(lowerCAmelCase_					, _REALM_BLOCK_RECORDS_FILENAME							)
   else:
     __A		:  Tuple       				=		hf_hub_download(
         repo_id=lowerCAmelCase_					, filename=_REALM_BLOCK_RECORDS_FILENAME					, **lowerCAmelCase_							)
   __A		:  Tuple       				=		np.load(lowerCAmelCase_					, allow_pickle=lowerCAmelCase_							)

   __A		:  Optional[Any]       				=		AutoTokenizer.from_pretrained(lowerCAmelCase_					, *lowerCAmelCase_					, **lowerCAmelCase_							)

   return cls(lowerCAmelCase_					, lowerCAmelCase_							)
 def 		lowercase_( self      :     Dict					, lowerCamelCase      :     Dict							):



   """simple docstring"""





   np.save(os.path.join(lowerCAmelCase_					, _REALM_BLOCK_RECORDS_FILENAME							)					, self.block_records							)
   # save tokenizer
   self.tokenizer.save_pretrained(lowerCAmelCase_							)






 def 		lowercase_( self      :     Any					, lowerCamelCase      :     Any					, lowerCamelCase      :     List[str]							):



   """simple docstring"""





   __A		:  Union[str, Any]       				=		[]
   __A		:  Union[str, Any]       				=		[]
   __A		:  List[str]       				=		[]
   __A		:  Optional[Any]       				=		0

   for input_id in concat_inputs.input_ids:
     __A		:  Tuple       				=		input_id.tolist()
     # Check answers between two [SEP] tokens
     __A		:  str       				=		input_id_list.index(self.tokenizer.sep_token_id							)
     __A		:  Union[str, Any]       				=		first_sep_idx + 1 + input_id_list[first_sep_idx + 1 :].index(self.tokenizer.sep_token_id							)

     start_pos.append([]							)
     end_pos.append([]							)
     for answer in answer_ids:
       for idx in range(first_sep_idx + 1					, lowerCAmelCase_							):
         if answer[0] == input_id_list[idx]:
           if input_id_list[idx : idx + len(lowerCAmelCase_							)] == answer:
             start_pos[-1].append(lowerCAmelCase_							)
             end_pos[-1].append(idx + len(lowerCAmelCase_							) - 1							)

     if len(start_pos[-1]							) == 0:
       has_answers.append(lowerCAmelCase_							)
     else:
       has_answers.append(lowerCAmelCase_							)
       if len(start_pos[-1]							) > max_answers:
         __A		:  Any       				=		len(start_pos[-1]							)

        # Pad -1 to max_answers
   for start_pos_, end_pos_ in zip(lowerCAmelCase_					, lowerCAmelCase_							):
     if len(start_pos_							) < max_answers:
       __A		:  int       				=		[-1] * (max_answers - len(start_pos_							))
       start_pos_ += padded
       end_pos_ += padded
   return has_answers, start_pos, end_pos




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:
						lowerCAmelCase__  	=      None

lowerCAmelCase__  	=      logging.get_logger(__name__)
lowerCAmelCase__  	=      {"vocab_file": "tokenizer.model", "tokenizer_file": "tokenizer.json"}

lowerCAmelCase__	,					lowerCAmelCase__  	=      "[INST]", "[/INST]"
lowerCAmelCase__	,					lowerCAmelCase__  	=      "<<SYS>>\n", "\n<</SYS>>\n\n"

# fmt: off
lowerCAmelCase__  	=      "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 	_a  (							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       ,  lowerCAmelCase_=None       ,  lowerCAmelCase_=None       ,  lowerCAmelCase_=False       ,  lowerCAmelCase_="<unk>"       ,  lowerCAmelCase_="<s>"       ,  lowerCAmelCase_="</s>"       ,  lowerCAmelCase_=True       ,  lowerCAmelCase_=False       ,  **lowerCAmelCase_       ,  ):
		super().__init__(
		    vocab_file=lowerCAmelCase_       ,  tokenizer_file=lowerCAmelCase_       ,  clean_up_tokenization_spaces=lowerCAmelCase_       ,  unk_token=lowerCAmelCase_       ,  bos_token=lowerCAmelCase_       ,  eos_token=lowerCAmelCase_       ,  **lowerCAmelCase_       ,  )
		_lowercase   =add_bos_token
		_lowercase   =add_eos_token
		self.update_post_processor()

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


	def __lowerCAmelCase      (   self       ):
		_lowercase   =self.bos_token
		_lowercase   =self.bos_token_id

		_lowercase   =self.eos_token
		_lowercase   =self.eos_token_id

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

		_lowercase   =[]
		if self.add_bos_token:
			special_tokens.append((bos, bos_token_id)       )
		if self.add_eos_token:
			special_tokens.append((eos, eos_token_id)       )
		_lowercase   =processors.TemplateProcessing(
		    single=lowerCAmelCase_       ,  pair=lowerCAmelCase_       ,  special_tokens=lowerCAmelCase_       )


	@property
	def __lowerCAmelCase      (   self       ):
		return self._add_eos_token


	@property
	def __lowerCAmelCase      (   self       ):
		return self._add_bos_token


	@add_eos_token.setter
	def __lowerCAmelCase      (   self       ,  lowerCAmelCase_       ):
		_lowercase   =value
		self.update_post_processor()


	@add_bos_token.setter
	def __lowerCAmelCase      (   self       ,  lowerCAmelCase_       ):
		_lowercase   =value
		self.update_post_processor()


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

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

		return (out_vocab_file,)


	def __lowerCAmelCase      (   self       ,  lowerCAmelCase_       ):
		_lowercase   =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...)"       )

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




