






'''simple docstring'''







import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


__lowerCAmelCase       		=      '▁'

__lowerCAmelCase       		=      {'vocab_file': 'spiece.model'}

__lowerCAmelCase       		=      {
    'vocab_file': {'google/pegasus-xsum': 'https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model'}
}

__lowerCAmelCase       		=      {
    'google/pegasus-xsum': 512,
}


__lowerCAmelCase       		=      logging.get_logger(__name__)





class 	SCREAMING_SNAKE_CASE					(					SCREAMING_SNAKE_CASE_       ):
			snake_case__  =				VOCAB_FILES_NAMES

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






			def __init__( self      :			List[str]  ,						__SCREAMING_SNAKE_CASE      :			Union[str, Any]  ,						__SCREAMING_SNAKE_CASE      :			Dict="<pad>"  ,						__SCREAMING_SNAKE_CASE      :			List[str]="</s>"  ,						__SCREAMING_SNAKE_CASE      :			Any="<unk>"  ,						__SCREAMING_SNAKE_CASE      :			Dict="<mask_2>"  ,						__SCREAMING_SNAKE_CASE      :			Dict="<mask_1>"  ,						__SCREAMING_SNAKE_CASE      :			str=None  ,						__SCREAMING_SNAKE_CASE      :			Tuple=103  ,						__SCREAMING_SNAKE_CASE      :			Optional[Dict[str, Any]] = None  ,						**__SCREAMING_SNAKE_CASE      :			Optional[int]  ,						)			->			None:
							a_		:  List[str]											=      offset
							if additional_special_tokens is not None:
											if not isinstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					):
															raise TypeError(
															    f'additional_special_tokens should be of type {type(__SCREAMING_SNAKE_CASE					)}, but is'
															    f' {type(__SCREAMING_SNAKE_CASE					)}'					)

											a_		:  List[Any]											=      (
											    ([mask_token_sent] + additional_special_tokens)
											    if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
											    else additional_special_tokens
											)
											# fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
											additional_special_tokens_extended += [
											    f'<unk_{i}>' for i in range(len(__SCREAMING_SNAKE_CASE					)  ,						self.offset - 1					)
											]

											if len(set(__SCREAMING_SNAKE_CASE					)					) != len(__SCREAMING_SNAKE_CASE					):
															raise ValueError(
															    '''Please make sure that the provided additional_special_tokens do not contain an incorrectly'''
															    f' shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}.'					)
											a_		:  Dict											=      additional_special_tokens_extended
							else:
											a_		:  str											=      [mask_token_sent] if mask_token_sent is not None else []
											additional_special_tokens += [f'<unk_{i}>' for i in range(2  ,						self.offset					)]

							a_		:  Union[str, Any]											=      {} if sp_model_kwargs is None else sp_model_kwargs

							super().__init__(
							    eos_token=__SCREAMING_SNAKE_CASE  ,						unk_token=__SCREAMING_SNAKE_CASE  ,						mask_token=__SCREAMING_SNAKE_CASE  ,						pad_token=__SCREAMING_SNAKE_CASE  ,						mask_token_sent=__SCREAMING_SNAKE_CASE  ,						offset=__SCREAMING_SNAKE_CASE  ,						additional_special_tokens=__SCREAMING_SNAKE_CASE  ,						sp_model_kwargs=self.sp_model_kwargs  ,						**__SCREAMING_SNAKE_CASE  ,						)
							a_		:  Optional[int]											=      mask_token_sent
							a_		:  Union[str, Any]											=      vocab_file
							a_		:  Optional[int]											=      spm.SentencePieceProcessor(**self.sp_model_kwargs					)
							self.sp_model.Load(__SCREAMING_SNAKE_CASE					)

							# add special tokens to encoder dict
							a_		:  Dict[int, str]											=      {
							    0: self.pad_token,
							    1: self.eos_token,
							}

							if self.mask_token_sent is not None:
											self.encoder.update(
											    {
											        2: self.mask_token_sent,
											        3: self.mask_token,
											    }					)

							if self.offset > 0:
											# entries 2-104 are only used for pretraining and called <mask_1>, <mask_2>, unk_2, ...unk_102
											# mask_token_sent is already added to list -> so start at 1
											self.encoder.update({i + 3: additional_special_tokens[i] for i in range(1  ,						self.offset - 1					)}					)

							a_		:  Dict[str, int]											=      {v: k for k, v in self.encoder.items()}






			@property
			def 			SCREAMING_SNAKE_CASE      ( self      :			Dict					)			->			int:
							return len(self.sp_model					) + self.offset






			def 			SCREAMING_SNAKE_CASE      ( self      :			Any					)			->			Dict[str, int]:
							a_		:  Any											=      {self.convert_ids_to_tokens(__SCREAMING_SNAKE_CASE					): i for i in range(self.vocab_size					)}
							vocab.update(self.added_tokens_encoder					)
							return vocab






			def __getstate__( self      :			Any					)			->			List[str]:
							a_		:  str											=      self.__dict__.copy()
							a_		:  Tuple											=      None
							return state






			def __setstate__( self      :			Optional[int]  ,						__SCREAMING_SNAKE_CASE      :			List[Any]					)			->			Tuple:
							a_		:  Union[str, Any]											=      d

							# for backward compatibility
							if not hasattr(self  ,						'''sp_model_kwargs'''					):
											a_		:  str											=      {}

							a_		:  Union[str, Any]											=      spm.SentencePieceProcessor(**self.sp_model_kwargs					)
							self.sp_model.Load(self.vocab_file					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			int  ,						__SCREAMING_SNAKE_CASE      :			str					)			->			List[str]:
							return self.sp_model.encode(__SCREAMING_SNAKE_CASE  ,						out_type=__SCREAMING_SNAKE_CASE					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			List[str]  ,						__SCREAMING_SNAKE_CASE      :			str					)			->			int:
							if token in self.decoder:
											return self.decoder[token]
							elif token in self.added_tokens_decoder:
											return self.added_tokens_decoder[token]
							a_		:  Tuple											=      self.sp_model.piece_to_id(__SCREAMING_SNAKE_CASE					)
							return sp_id + self.offset






			def 			SCREAMING_SNAKE_CASE      ( self      :			int  ,						__SCREAMING_SNAKE_CASE      :			int					)			->			str:
							if index in self.encoder:
											return self.encoder[index]
							elif index in self.added_tokens_encoder:
											return self.added_tokens_encoder[index]
							else:
											a_		:  List[Any]											=      self.sp_model.IdToPiece(index - self.offset					)
							return token






			def 			SCREAMING_SNAKE_CASE      ( self      :			Dict  ,						__SCREAMING_SNAKE_CASE      :			Tuple					)			->			Any:
							a_		:  int											=      []
							a_		:  List[str]											=      ''''''
							for token in tokens:
											# make sure that special tokens are not decoded using sentencepiece model
											if token in self.all_special_tokens:
															out_string += self.sp_model.decode(__SCREAMING_SNAKE_CASE					) + token
															a_		:  List[Any]											=      []
											else:
															current_sub_tokens.append(__SCREAMING_SNAKE_CASE					)
							out_string += self.sp_model.decode(__SCREAMING_SNAKE_CASE					)
							return out_string.strip()






			def 			SCREAMING_SNAKE_CASE      ( self      :			Union[str, Any]  ,						__SCREAMING_SNAKE_CASE      :			List[str]=False					)			->			Tuple:
							return 1






			def 			SCREAMING_SNAKE_CASE      ( self      :			List[Any]  ,						__SCREAMING_SNAKE_CASE      :			Optional[Any]					)			->			List[Any]:
							a_		:  Optional[Any]											=      set(self.all_special_ids					)  # call it once instead of inside list comp
							all_special_ids.remove(self.unk_token_id					)  # <unk> is only sometimes special

							return [1 if x in all_special_ids else 0 for x in seq]






			def 			SCREAMING_SNAKE_CASE      ( self      :			Any  ,						__SCREAMING_SNAKE_CASE      :			List  ,						__SCREAMING_SNAKE_CASE      :			Optional[List] = None  ,						__SCREAMING_SNAKE_CASE      :			bool = False					)			->			List[int]:
							if already_has_special_tokens:
											return self._special_token_mask(__SCREAMING_SNAKE_CASE					)
							elif token_ids_a is None:
											return self._special_token_mask(__SCREAMING_SNAKE_CASE					) + [1]
							else:
											return self._special_token_mask(token_ids_a + token_ids_a					) + [1]






			def 			SCREAMING_SNAKE_CASE      ( self      :			Union[str, Any]  ,						__SCREAMING_SNAKE_CASE      :			Optional[int]  ,						__SCREAMING_SNAKE_CASE      :			Optional[int]=None					)			->			List[int]:
							if token_ids_a is None:
											return token_ids_a + [self.eos_token_id]
							# We don't expect to process pairs, but leave the pair logic for API consistency
							return token_ids_a + token_ids_a + [self.eos_token_id]







			def 			SCREAMING_SNAKE_CASE      ( self      :			int  ,						__SCREAMING_SNAKE_CASE      :			str  ,						__SCREAMING_SNAKE_CASE      :			Optional[str] = None					)			->			Tuple[str]:
							if not os.path.isdir(__SCREAMING_SNAKE_CASE					):
											logger.error(f'Vocabulary path ({save_directory}) should be a directory'					)
											return
							a_		:  Any											=      os.path.join(
							    __SCREAMING_SNAKE_CASE  ,						(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']					)

							if os.path.abspath(self.vocab_file					) != os.path.abspath(__SCREAMING_SNAKE_CASE					) and os.path.isfile(self.vocab_file					):
											copyfile(self.vocab_file  ,						__SCREAMING_SNAKE_CASE					)
							elif not os.path.isfile(self.vocab_file					):
											with open(__SCREAMING_SNAKE_CASE  ,						'''wb'''					) as fi:
															a_		:  Optional[int]											=      self.sp_model.serialized_model_proto()
															fi.write(__SCREAMING_SNAKE_CASE					)

							return (out_vocab_file,)












'''simple docstring'''







import json
import os
import pickle
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers import is_faiss_available
from transformers.models.bart.configuration_bart import BartConfig
from transformers.models.bart.tokenization_bart import BartTokenizer
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.dpr.configuration_dpr import DPRConfig
from transformers.models.dpr.tokenization_dpr import DPRContextEncoderTokenizer, DPRQuestionEncoderTokenizer
from transformers.models.rag.configuration_rag import RagConfig
from transformers.models.rag.retrieval_rag import CustomHFIndex, RagRetriever
from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
from transformers.testing_utils import require_faiss, require_sentencepiece, require_tokenizers, require_torch


if is_faiss_available():
		import faiss





@require_faiss
class 	SCREAMING_SNAKE_CASE					(					SCREAMING_SNAKE_CASE_       ):






			def 			SCREAMING_SNAKE_CASE      ( self      :			Dict					)			->			Dict:
							a_		:  Union[str, Any]											=      tempfile.mkdtemp()
							a_		:  Union[str, Any]											=      8

							# DPR tok
							a_		:  Dict											=      [
							    '''[UNK]''',
							    '''[CLS]''',
							    '''[SEP]''',
							    '''[PAD]''',
							    '''[MASK]''',
							    '''want''',
							    '''##want''',
							    '''##ed''',
							    '''wa''',
							    '''un''',
							    '''runn''',
							    '''##ing''',
							    ''',''',
							    '''low''',
							    '''lowest''',
							]
							a_		:  str											=      os.path.join(self.tmpdirname  ,						'''dpr_tokenizer'''					)
							os.makedirs(__SCREAMING_SNAKE_CASE  ,						exist_ok=__SCREAMING_SNAKE_CASE					)
							a_		:  Optional[int]											=      os.path.join(__SCREAMING_SNAKE_CASE  ,						DPR_VOCAB_FILES_NAMES['''vocab_file''']					)
							with open(self.vocab_file  ,						'''w'''  ,						encoding='''utf-8'''					) as vocab_writer:
											vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]					)					)

							# BART tok
							a_		:  Union[str, Any]											=      [
							    '''l''',
							    '''o''',
							    '''w''',
							    '''e''',
							    '''r''',
							    '''s''',
							    '''t''',
							    '''i''',
							    '''d''',
							    '''n''',
							    '''\u0120''',
							    '''\u0120l''',
							    '''\u0120n''',
							    '''\u0120lo''',
							    '''\u0120low''',
							    '''er''',
							    '''\u0120lowest''',
							    '''\u0120newer''',
							    '''\u0120wider''',
							    '''<unk>''',
							]
							a_		:  int											=      dict(zip(__SCREAMING_SNAKE_CASE  ,						range(len(__SCREAMING_SNAKE_CASE					)					)					)					)
							a_		:  int											=      ['''#version: 0.2''', '''\u0120 l''', '''\u0120l o''', '''\u0120lo w''', '''e r''', '''''']
							a_		:  Optional[int]											=      {'''unk_token''': '''<unk>'''}

							a_		:  List[str]											=      os.path.join(self.tmpdirname  ,						'''bart_tokenizer'''					)
							os.makedirs(__SCREAMING_SNAKE_CASE  ,						exist_ok=__SCREAMING_SNAKE_CASE					)
							a_		:  Tuple											=      os.path.join(__SCREAMING_SNAKE_CASE  ,						BART_VOCAB_FILES_NAMES['''vocab_file''']					)
							a_		:  int											=      os.path.join(__SCREAMING_SNAKE_CASE  ,						BART_VOCAB_FILES_NAMES['''merges_file''']					)
							with open(self.vocab_file  ,						'''w'''  ,						encoding='''utf-8'''					) as fp:
											fp.write(json.dumps(__SCREAMING_SNAKE_CASE					) + '''\n'''					)
							with open(self.merges_file  ,						'''w'''  ,						encoding='''utf-8'''					) as fp:
											fp.write('''\n'''.join(__SCREAMING_SNAKE_CASE					)					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Tuple					)			->			DPRQuestionEncoderTokenizer:
							return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname  ,						'''dpr_tokenizer'''					)					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			str					)			->			DPRContextEncoderTokenizer:
							return DPRContextEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname  ,						'''dpr_tokenizer'''					)					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Optional[int]					)			->			BartTokenizer:
							return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname  ,						'''bart_tokenizer'''					)					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			List[Any]					)			->			Union[str, Any]:
							shutil.rmtree(self.tmpdirname					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Dict					)			->			Union[str, Any]:
							a_		:  str											=      Dataset.from_dict(
							    {
							        '''id''': ['''0''', '''1'''],
							        '''text''': ['''foo''', '''bar'''],
							        '''title''': ['''Foo''', '''Bar'''],
							        '''embeddings''': [np.ones(self.retrieval_vector_size					), 2 * np.ones(self.retrieval_vector_size					)],
							    }					)
							dataset.add_faiss_index('''embeddings'''  ,						string_factory='''Flat'''  ,						metric_type=faiss.METRIC_INNER_PRODUCT					)
							return dataset






			def 			SCREAMING_SNAKE_CASE      ( self      :			List[str]					)			->			Tuple:
							a_		:  List[str]											=      self.get_dummy_dataset()
							a_		:  Tuple											=      RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size  ,						question_encoder=DPRConfig().to_dict()  ,						generator=BartConfig().to_dict()  ,						)
							with patch('''transformers.models.rag.retrieval_rag.load_dataset'''					) as mock_load_dataset:
											a_		:  Tuple											=      dataset
											a_		:  Any											=      RagRetriever(
											    __SCREAMING_SNAKE_CASE  ,						question_encoder_tokenizer=self.get_dpr_tokenizer()  ,						generator_tokenizer=self.get_bart_tokenizer()  ,						)
							return retriever






			def 			SCREAMING_SNAKE_CASE      ( self      :			Dict  ,						__SCREAMING_SNAKE_CASE      :			bool					)			->			Dict:
							a_		:  Dict											=      self.get_dummy_dataset()
							a_		:  Dict											=      RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size  ,						question_encoder=DPRConfig().to_dict()  ,						generator=BartConfig().to_dict()  ,						index_name='''custom'''  ,						)
							if from_disk:
											a_		:  Optional[int]											=      os.path.join(self.tmpdirname  ,						'''dataset'''					)
											a_		:  str											=      os.path.join(self.tmpdirname  ,						'''index.faiss'''					)
											dataset.get_index('''embeddings'''					).save(os.path.join(self.tmpdirname  ,						'''index.faiss'''					)					)
											dataset.drop_index('''embeddings'''					)
											dataset.save_to_disk(os.path.join(self.tmpdirname  ,						'''dataset'''					)					)
											del dataset
											a_		:  int											=      RagRetriever(
											    __SCREAMING_SNAKE_CASE  ,						question_encoder_tokenizer=self.get_dpr_tokenizer()  ,						generator_tokenizer=self.get_bart_tokenizer()  ,						)
							else:
											a_		:  Optional[Any]											=      RagRetriever(
											    __SCREAMING_SNAKE_CASE  ,						question_encoder_tokenizer=self.get_dpr_tokenizer()  ,						generator_tokenizer=self.get_bart_tokenizer()  ,						index=CustomHFIndex(config.retrieval_vector_size  ,						__SCREAMING_SNAKE_CASE					)  ,						)
							return retriever






			def 			SCREAMING_SNAKE_CASE      ( self      :			Optional[Any]					)			->			Optional[int]:
							a_		:  str											=      Dataset.from_dict(
							    {
							        '''id''': ['''0''', '''1'''],
							        '''text''': ['''foo''', '''bar'''],
							        '''title''': ['''Foo''', '''Bar'''],
							        '''embeddings''': [np.ones(self.retrieval_vector_size + 1					), 2 * np.ones(self.retrieval_vector_size + 1					)],
							    }					)
							dataset.add_faiss_index('''embeddings'''  ,						string_factory='''Flat'''  ,						metric_type=faiss.METRIC_INNER_PRODUCT					)

							a_		:  Optional[int]											=      os.path.join(self.tmpdirname  ,						'''hf_bert_base.hnswSQ8_correct_phi_128.c_index'''					)
							dataset.save_faiss_index('''embeddings'''  ,						index_file_name + '''.index.dpr'''					)
							pickle.dump(dataset['''id''']  ,						open(index_file_name + '''.index_meta.dpr'''  ,						'''wb'''					)					)

							a_		:  Union[str, Any]											=      os.path.join(self.tmpdirname  ,						'''psgs_w100.tsv.pkl'''					)
							a_		:  Dict											=      {sample['''id''']: [sample['''text'''], sample['''title''']] for sample in dataset}
							pickle.dump(__SCREAMING_SNAKE_CASE  ,						open(__SCREAMING_SNAKE_CASE  ,						'''wb'''					)					)

							a_		:  Optional[Any]											=      RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size  ,						question_encoder=DPRConfig().to_dict()  ,						generator=BartConfig().to_dict()  ,						index_name='''legacy'''  ,						index_path=self.tmpdirname  ,						)
							a_		:  int											=      RagRetriever(
							    __SCREAMING_SNAKE_CASE  ,						question_encoder_tokenizer=self.get_dpr_tokenizer()  ,						generator_tokenizer=self.get_bart_tokenizer()					)
							return retriever






			def 			SCREAMING_SNAKE_CASE      ( self      :			Tuple					)			->			Dict:
							a_		:  Optional[Any]											=      1
							a_		:  Dict											=      self.get_dummy_canonical_hf_index_retriever()
							a_		:  Tuple											=      np.array(
							    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
							a_				,			a_				,			a_		:  str											=      retriever.retrieve(__SCREAMING_SNAKE_CASE  ,						n_docs=__SCREAMING_SNAKE_CASE					)
							self.assertEqual(retrieved_doc_embeds.shape  ,						(2, n_docs, self.retrieval_vector_size)					)
							self.assertEqual(len(__SCREAMING_SNAKE_CASE					)  ,						2					)
							self.assertEqual(sorted(doc_dicts[0]					)  ,						['''embeddings''', '''id''', '''text''', '''title''']					)
							self.assertEqual(len(doc_dicts[0]['''id''']					)  ,						__SCREAMING_SNAKE_CASE					)
							self.assertEqual(doc_dicts[0]['''id'''][0]  ,						'''1'''					)  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['''id'''][0]  ,						'''0'''					)  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()  ,						[[1], [0]]					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Tuple					)			->			Dict:
							a_		:  str											=      self.get_dummy_canonical_hf_index_retriever()
							with tempfile.TemporaryDirectory() as tmp_dirname:
											with patch('''transformers.models.rag.retrieval_rag.load_dataset'''					) as mock_load_dataset:
															a_		:  List[str]											=      self.get_dummy_dataset()
															retriever.save_pretrained(__SCREAMING_SNAKE_CASE					)
															a_		:  Optional[Any]											=      RagRetriever.from_pretrained(__SCREAMING_SNAKE_CASE					)
															self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					)
															a_		:  Union[str, Any]											=      np.array(
															    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
															a_		:  List[str]											=      retriever.retrieve(__SCREAMING_SNAKE_CASE  ,						n_docs=1					)
															self.assertTrue(out is not None					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Dict					)			->			int:
							a_		:  Union[str, Any]											=      1
							a_		:  Optional[Any]											=      self.get_dummy_custom_hf_index_retriever(from_disk=__SCREAMING_SNAKE_CASE					)
							a_		:  List[Any]											=      np.array(
							    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
							a_				,			a_				,			a_		:  Any											=      retriever.retrieve(__SCREAMING_SNAKE_CASE  ,						n_docs=__SCREAMING_SNAKE_CASE					)
							self.assertEqual(retrieved_doc_embeds.shape  ,						(2, n_docs, self.retrieval_vector_size)					)
							self.assertEqual(len(__SCREAMING_SNAKE_CASE					)  ,						2					)
							self.assertEqual(sorted(doc_dicts[0]					)  ,						['''embeddings''', '''id''', '''text''', '''title''']					)
							self.assertEqual(len(doc_dicts[0]['''id''']					)  ,						__SCREAMING_SNAKE_CASE					)
							self.assertEqual(doc_dicts[0]['''id'''][0]  ,						'''1'''					)  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['''id'''][0]  ,						'''0'''					)  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()  ,						[[1], [0]]					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			str					)			->			List[str]:
							a_		:  Dict											=      self.get_dummy_custom_hf_index_retriever(from_disk=__SCREAMING_SNAKE_CASE					)
							with tempfile.TemporaryDirectory() as tmp_dirname:
											retriever.save_pretrained(__SCREAMING_SNAKE_CASE					)
											a_		:  List[str]											=      RagRetriever.from_pretrained(__SCREAMING_SNAKE_CASE					)
											self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					)
											a_		:  Tuple											=      np.array(
											    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
											a_		:  Dict											=      retriever.retrieve(__SCREAMING_SNAKE_CASE  ,						n_docs=1					)
											self.assertTrue(out is not None					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Tuple					)			->			Optional[int]:
							a_		:  Union[str, Any]											=      1
							a_		:  str											=      self.get_dummy_custom_hf_index_retriever(from_disk=__SCREAMING_SNAKE_CASE					)
							a_		:  Optional[int]											=      np.array(
							    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
							a_				,			a_				,			a_		:  Tuple											=      retriever.retrieve(__SCREAMING_SNAKE_CASE  ,						n_docs=__SCREAMING_SNAKE_CASE					)
							self.assertEqual(retrieved_doc_embeds.shape  ,						(2, n_docs, self.retrieval_vector_size)					)
							self.assertEqual(len(__SCREAMING_SNAKE_CASE					)  ,						2					)
							self.assertEqual(sorted(doc_dicts[0]					)  ,						['''embeddings''', '''id''', '''text''', '''title''']					)
							self.assertEqual(len(doc_dicts[0]['''id''']					)  ,						__SCREAMING_SNAKE_CASE					)
							self.assertEqual(doc_dicts[0]['''id'''][0]  ,						'''1'''					)  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['''id'''][0]  ,						'''0'''					)  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()  ,						[[1], [0]]					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			int					)			->			Dict:
							a_		:  List[str]											=      self.get_dummy_custom_hf_index_retriever(from_disk=__SCREAMING_SNAKE_CASE					)
							with tempfile.TemporaryDirectory() as tmp_dirname:
											retriever.save_pretrained(__SCREAMING_SNAKE_CASE					)
											a_		:  Any											=      RagRetriever.from_pretrained(__SCREAMING_SNAKE_CASE					)
											self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					)
											a_		:  Tuple											=      np.array(
											    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
											a_		:  Dict											=      retriever.retrieve(__SCREAMING_SNAKE_CASE  ,						n_docs=1					)
											self.assertTrue(out is not None					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Optional[int]					)			->			Any:
							a_		:  str											=      1
							a_		:  Tuple											=      self.get_dummy_legacy_index_retriever()
							a_		:  Union[str, Any]											=      np.array(
							    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
							a_				,			a_				,			a_		:  Any											=      retriever.retrieve(__SCREAMING_SNAKE_CASE  ,						n_docs=__SCREAMING_SNAKE_CASE					)
							self.assertEqual(retrieved_doc_embeds.shape  ,						(2, n_docs, self.retrieval_vector_size)					)
							self.assertEqual(len(__SCREAMING_SNAKE_CASE					)  ,						2					)
							self.assertEqual(sorted(doc_dicts[0]					)  ,						['''text''', '''title''']					)
							self.assertEqual(len(doc_dicts[0]['''text''']					)  ,						__SCREAMING_SNAKE_CASE					)
							self.assertEqual(doc_dicts[0]['''text'''][0]  ,						'''bar'''					)  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['''text'''][0]  ,						'''foo'''					)  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()  ,						[[1], [0]]					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			int					)			->			Optional[int]:
							a_		:  List[str]											=      self.get_dummy_legacy_index_retriever()
							with tempfile.TemporaryDirectory() as tmp_dirname:
											retriever.save_pretrained(__SCREAMING_SNAKE_CASE					)
											a_		:  Any											=      RagRetriever.from_pretrained(__SCREAMING_SNAKE_CASE					)
											self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					)
											a_		:  List[str]											=      np.array(
											    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
											a_		:  Optional[Any]											=      retriever.retrieve(__SCREAMING_SNAKE_CASE  ,						n_docs=1					)
											self.assertTrue(out is not None					)






			@require_torch
			@require_tokenizers
			@require_sentencepiece
			def 			SCREAMING_SNAKE_CASE      ( self      :			int					)			->			List[str]:
							import torch

							a_		:  Any											=      1
							a_		:  List[Any]											=      self.get_dummy_canonical_hf_index_retriever()
							a_		:  Union[str, Any]											=      [[5, 7], [10, 11]]
							a_		:  Optional[int]											=      np.array(
							    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
							a_		:  str											=      retriever(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE  ,						prefix=retriever.config.generator.prefix  ,						n_docs=__SCREAMING_SNAKE_CASE					)
							a_				,			a_				,			a_		:  List[str]											=      (
							    out['''context_input_ids'''],
							    out['''context_attention_mask'''],
							    out['''retrieved_doc_embeds'''],
							)
							self.assertEqual(retrieved_doc_embeds.shape  ,						(2, n_docs, self.retrieval_vector_size)					)
							self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					)
							self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					)
							self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						np.ndarray					)

							a_		:  Any											=      retriever(
							    __SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE  ,						prefix=retriever.config.generator.prefix  ,						n_docs=__SCREAMING_SNAKE_CASE  ,						return_tensors='''pt'''  ,						)
							a_				,			a_				,			a_				,			a_		:  str											=      (  # noqa: F841
							    out['''context_input_ids'''],
							    out['''context_attention_mask'''],
							    out['''retrieved_doc_embeds'''],
							    out['''doc_ids'''],
							)
							self.assertEqual(retrieved_doc_embeds.shape  ,						(2, n_docs, self.retrieval_vector_size)					)
							self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						torch.Tensor					)
							self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						torch.Tensor					)
							self.assertIsInstance(__SCREAMING_SNAKE_CASE  ,						torch.Tensor					)







			@require_torch
			@require_tokenizers
			@require_sentencepiece
			def 			SCREAMING_SNAKE_CASE      ( self      :			Union[str, Any]					)			->			Any:
							a_		:  str											=      self.get_dpr_ctx_encoder_tokenizer()
							a_		:  Tuple											=      1
							a_		:  Any											=      self.get_dummy_custom_hf_index_retriever(from_disk=__SCREAMING_SNAKE_CASE					)
							retriever.set_ctx_encoder_tokenizer(__SCREAMING_SNAKE_CASE					)

							a_		:  Dict											=      [[5, 7], [10, 11]]
							a_		:  List[str]											=      np.array(
							    [np.ones(self.retrieval_vector_size					), -np.ones(self.retrieval_vector_size					)]  ,						dtype=np.floataa					)
							a_		:  List[Any]											=      retriever(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE  ,						prefix=retriever.config.generator.prefix  ,						n_docs=__SCREAMING_SNAKE_CASE					)

							self.assertEqual(
							    len(__SCREAMING_SNAKE_CASE					)  ,						6					)  # check whether the retriever output consist of 6 attributes including tokenized docs
							self.assertEqual(
							    all(k in out for k in ('''tokenized_doc_ids''', '''tokenized_doc_attention_mask''')					)  ,						__SCREAMING_SNAKE_CASE					)  # check for doc token related keys in dictionary.




