
import json
import os
from functools import lru_cache
from typing import Dict, List, Optional, Tuple, Union

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


__lowerCAmelCase					:			Tuple			      =		logging.get_logger(__name__)


__lowerCAmelCase					:			int			      =		{"vocab_file": "vocab.json", "merges_file": "merges.txt"}

# See all LED models at https://huggingface.co/models?filter=LED
__lowerCAmelCase					:			List[Any]			      =		{
    "vocab_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json",
    },
    "merges_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt",
    },
    "tokenizer_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json",
    },
}

__lowerCAmelCase					:			List[Any]			      =		{
    "allenai/led-base-16384": 16_384,
}


@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def     UpperCAmelCase_  (						) ->						List[Any]:
    __lowercase		:			Union[str, Any]	    =       (
        list(range(ord('''!'''	)	,					ord('''~'''	) + 1	)	) + list(range(ord('''¡'''	)	,					ord('''¬'''	) + 1	)	) + list(range(ord('''®'''	)	,					ord('''ÿ'''	) + 1	)	)
    )
    __lowercase		:			Any	    =       bs[:]
    __lowercase		:			Union[str, Any]	    =       0
    for b in range(2**8	):
        if b not in bs:
            bs.append(__lowerCAmelCase	)
            cs.append(2**8 + n	)
            n += 1
    __lowercase		:			List[str]	    =       [chr(__lowerCAmelCase	) for n in cs]
    return dict(zip(__lowerCAmelCase	,					__lowerCAmelCase	)	)


def     UpperCAmelCase_  (						__lowerCAmelCase	) ->						List[str]:
    __lowercase		:			Optional[int]	    =       set()
    __lowercase		:			Dict	    =       word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)	)
        __lowercase		:			List[Any]	    =       char
    return pairs





class 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      A__	:			str        =    VOCAB_FILES_NAMES
      A__	:			Any        =    PRETRAINED_VOCAB_FILES_MAP
      A__	:			int        =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      A__	:			Union[str, Any]        =    ['''input_ids''', '''attention_mask''']




      def __init__(      self		:       Dict				,       _snake_case		:       List[Any]				,       _snake_case		:       Dict				,       _snake_case		:       int="replace"				,       _snake_case		:       Optional[int]="<s>"				,       _snake_case		:       List[Any]="</s>"				,       _snake_case		:       str="</s>"				,       _snake_case		:       Optional[int]="<s>"				,       _snake_case		:       List[Any]="<unk>"				,       _snake_case		:       Optional[Any]="<pad>"				,       _snake_case		:       List[Any]="<mask>"				,       _snake_case		:       Optional[Any]=False				,       **_snake_case		:       Dict				,       ):
          __lowercase		:			str	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else bos_token
          __lowercase		:			List[Any]	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else eos_token
          __lowercase		:			Union[str, Any]	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else sep_token
          __lowercase		:			List[str]	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else cls_token
          __lowercase		:			Any	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else unk_token
          __lowercase		:			Optional[Any]	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else pad_token

          # Mask token behave like a normal word, i.e. include the space before it
          __lowercase		:			Any	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else mask_token

          super().__init__(
              errors=_snake_case				,       bos_token=_snake_case				,       eos_token=_snake_case				,       unk_token=_snake_case				,       sep_token=_snake_case				,       cls_token=_snake_case				,       pad_token=_snake_case				,       mask_token=_snake_case				,       add_prefix_space=_snake_case				,       **_snake_case				,       )

          with open(_snake_case				,       encoding='''utf-8'''				) as vocab_handle:
              __lowercase		:			List[str]	    =       json.load(_snake_case				)
          __lowercase		:			Optional[Any]	    =       {v: k for k, v in self.encoder.items()}
          __lowercase		:			Dict	    =       errors  # how to handle errors in decoding
          __lowercase		:			str	    =       bytes_to_unicode()
          __lowercase		:			Dict	    =       {v: k for k, v in self.byte_encoder.items()}
          with open(_snake_case				,       encoding='''utf-8'''				) as merges_handle:
              __lowercase		:			Any	    =       merges_handle.read().split('''\n'''				)[1:-1]
          __lowercase		:			Tuple	    =       [tuple(merge.split()				) for merge in bpe_merges]
          __lowercase		:			Dict	    =       dict(zip(_snake_case				,       range(len(_snake_case				)				)				)				)
          __lowercase		:			Dict	    =       {}
          __lowercase		:			str	    =       add_prefix_space

          # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
          __lowercase		:			str	    =       re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''				)




      @property
      # Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
      def 		snake_case_			(      self		:       Dict				):
          return len(self.encoder				)




      def 		snake_case_			(      self		:       List[str]				):
          return dict(self.encoder				,       **self.added_tokens_encoder				)




      def 		snake_case_			(      self		:       Tuple				,       _snake_case		:       str				):
          if token in self.cache:
              return self.cache[token]
          __lowercase		:			Optional[Any]	    =       tuple(_snake_case				)
          __lowercase		:			str	    =       get_pairs(_snake_case				)

          if not pairs:
              return token

          while True:
              __lowercase		:			Dict	    =       min(_snake_case				,       key=lambda _snake_case				: self.bpe_ranks.get(_snake_case				,       float('''inf'''				)				)				)
              if bigram not in self.bpe_ranks:
                  break
              __lowercase    ,      __lowercase		:			Tuple	    =       bigram
              __lowercase		:			Dict	    =       []
              __lowercase		:			int	    =       0
              while i < len(_snake_case				):
                  try:
                      __lowercase		:			List[Any]	    =       word.index(_snake_case				,       _snake_case				)
                  except ValueError:
                      new_word.extend(word[i:]				)
                      break
                  else:
                      new_word.extend(word[i:j]				)
                      __lowercase		:			Optional[int]	    =       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
              __lowercase		:			str	    =       tuple(_snake_case				)
              __lowercase		:			Tuple	    =       new_word
              if len(_snake_case				) == 1:
                  break
              else:
                  __lowercase		:			Any	    =       get_pairs(_snake_case				)
          __lowercase		:			Union[str, Any]	    =       ''' '''.join(_snake_case				)
          __lowercase		:			Union[str, Any]	    =       word
          return word




      def 		snake_case_			(      self		:       Tuple				,       _snake_case		:       List[Any]				):
          __lowercase		:			Optional[int]	    =       []
          for token in re.findall(self.pat				,       _snake_case				):
              __lowercase		:			Dict	    =       ''''''.join(
                  self.byte_encoder[b] for b in token.encode('''utf-8'''				)				)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
              bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_snake_case				).split(''' '''				)				)
          return bpe_tokens




      def 		snake_case_			(      self		:       Union[str, Any]				,       _snake_case		:       List[Any]				):
          return self.encoder.get(_snake_case				,       self.encoder.get(self.unk_token				)				)




      def 		snake_case_			(      self		:       Dict				,       _snake_case		:       List[Any]				):
          return self.decoder.get(_snake_case				)




      def 		snake_case_			(      self		:       int				,       _snake_case		:       Dict				):
          __lowercase		:			List[Any]	    =       ''''''.join(_snake_case				)
          __lowercase		:			Any	    =       bytearray([self.byte_decoder[c] for c in text]				).decode('''utf-8'''				,       errors=self.errors				)
          return text




      def 		snake_case_			(      self		:       Optional[Any]				,       _snake_case		:       str				,       _snake_case		:       Optional[str] = None				):
          if not os.path.isdir(_snake_case				):
              logger.error(F'Vocabulary path ({save_directory}) should be a directory'				)
              return
          __lowercase		:			Union[str, Any]	    =       os.path.join(
              _snake_case				,       (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']				)
          __lowercase		:			str	    =       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'''				)

          __lowercase		:			List[str]	    =       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!'''				)
                      __lowercase		:			List[str]	    =       token_index
                  writer.write(''' '''.join(_snake_case				) + '''\n'''				)
                  index += 1

          return vocab_file, merge_file




      def 		snake_case_			(      self		:       Dict				,       _snake_case		:       List[int]				,       _snake_case		:       Optional[List[int]] = None				):
          if token_ids_a is None:
              return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
          __lowercase		:			int	    =       [self.cls_token_id]
          __lowercase		:			int	    =       [self.sep_token_id]
          return cls + token_ids_a + sep + sep + token_ids_a + sep




      def 		snake_case_			(      self		:       List[Any]				,       _snake_case		:       List[int]				,       _snake_case		:       Optional[List[int]] = None				,       _snake_case		:       bool = False				):
          if already_has_special_tokens:
              return super().get_special_tokens_mask(
                  token_ids_a=_snake_case				,       token_ids_a=_snake_case				,       already_has_special_tokens=_snake_case				)

          if token_ids_a is None:
              return [1] + ([0] * len(_snake_case				)) + [1]
          return [1] + ([0] * len(_snake_case				)) + [1, 1] + ([0] * len(_snake_case				)) + [1]




      def 		snake_case_			(      self		:       Union[str, Any]				,       _snake_case		:       List[int]				,       _snake_case		:       Optional[List[int]] = None				):
          __lowercase		:			Dict	    =       [self.sep_token_id]
          __lowercase		:			List[str]	    =       [self.cls_token_id]

          if token_ids_a is None:
              return len(cls + token_ids_a + sep				) * [0]
          return len(cls + token_ids_a + sep + sep + token_ids_a + sep				) * [0]




      def 		snake_case_			(      self		:       List[str]				,       _snake_case		:       Union[str, Any]				,       _snake_case		:       Optional[int]=False				,       **_snake_case		:       List[Any]				):
          __lowercase		:			int	    =       kwargs.pop('''add_prefix_space'''				,       self.add_prefix_space				)
          if (is_split_into_words or add_prefix_space) and (len(_snake_case				) > 0 and not text[0].isspace()):
              __lowercase		:			Optional[Any]	    =       ''' ''' + text
          return (text, kwargs)




      def 		snake_case_			(      self		:       Tuple				,       _snake_case		:       Union[Dict[str, EncodedInput], BatchEncoding]				,       _snake_case		:       Optional[int] = None				,       _snake_case		:       PaddingStrategy = PaddingStrategy.DO_NOT_PAD				,       _snake_case		:       Optional[int] = None				,       _snake_case		:       Optional[bool] = None				,       ):
          __lowercase		:			Optional[Any]	    =       super()._pad(
              encoded_inputs=_snake_case				,       max_length=_snake_case				,       padding_strategy=_snake_case				,       pad_to_multiple_of=_snake_case				,       return_attention_mask=_snake_case				,       )

          # Load from model defaults
          if return_attention_mask is None:
              __lowercase		:			List[str]	    =       '''attention_mask''' in self.model_input_names

          if return_attention_mask and "global_attention_mask" in encoded_inputs:
              __lowercase		:			Dict	    =       encoded_inputs[self.model_input_names[0]]
              # `global_attention_mask` need to have the same length as other (sequential) inputs.
              __lowercase		:			Tuple	    =       len(encoded_inputs['''global_attention_mask''']				) != len(_snake_case				)

              if needs_to_be_padded:
                  __lowercase		:			List[Any]	    =       len(_snake_case				) - len(encoded_inputs['''global_attention_mask''']				)

                  if self.padding_side == "right":
                      # Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
                      __lowercase		:			Union[str, Any]	    =       (
                          encoded_inputs['''global_attention_mask'''] + [-1] * difference
                      )
                  elif self.padding_side == "left":
                      __lowercase		:			Union[str, Any]	    =       [-1] * difference + encoded_inputs[
                          '''global_attention_mask'''
                      ]
                  else:
                      raise ValueError('''Invalid padding strategy:''' + str(self.padding_side				)				)

          return encoded_inputs




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 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      def 		snake_case_			(      self		:       Optional[int]				):
          __lowercase		:			Dict	    =       tempfile.mkdtemp()
          __lowercase		:			Tuple	    =       8

          # DPR tok
          __lowercase		:			Optional[int]	    =       [
              '''[UNK]''',
              '''[CLS]''',
              '''[SEP]''',
              '''[PAD]''',
              '''[MASK]''',
              '''want''',
              '''##want''',
              '''##ed''',
              '''wa''',
              '''un''',
              '''runn''',
              '''##ing''',
              ''',''',
              '''low''',
              '''lowest''',
          ]
          __lowercase		:			Optional[Any]	    =       os.path.join(self.tmpdirname				,       '''dpr_tokenizer'''				)
          os.makedirs(_snake_case				,       exist_ok=_snake_case				)
          __lowercase		:			str	    =       os.path.join(_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
          __lowercase		:			List[str]	    =       [
              '''l''',
              '''o''',
              '''w''',
              '''e''',
              '''r''',
              '''s''',
              '''t''',
              '''i''',
              '''d''',
              '''n''',
              '''\u0120''',
              '''\u0120l''',
              '''\u0120n''',
              '''\u0120lo''',
              '''\u0120low''',
              '''er''',
              '''\u0120lowest''',
              '''\u0120newer''',
              '''\u0120wider''',
              '''<unk>''',
          ]
          __lowercase		:			Optional[Any]	    =       dict(zip(_snake_case				,       range(len(_snake_case				)				)				)				)
          __lowercase		:			int	    =       ['''#version: 0.2''', '''\u0120 l''', '''\u0120l o''', '''\u0120lo w''', '''e r''', '''''']
          __lowercase		:			Any	    =       {'''unk_token''': '''<unk>'''}

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




      def 		snake_case_			(      self		:       List[Any]				):
          return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname				,       '''dpr_tokenizer'''				)				)




      def 		snake_case_			(      self		:       Dict				):
          return DPRContextEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname				,       '''dpr_tokenizer'''				)				)




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




      def 		snake_case_			(      self		:       str				):
          shutil.rmtree(self.tmpdirname				)




      def 		snake_case_			(      self		:       Tuple				):
          __lowercase		:			Dict	    =       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 		snake_case_			(      self		:       Union[str, Any]				):
          __lowercase		:			Union[str, Any]	    =       self.get_dummy_dataset()
          __lowercase		:			Optional[int]	    =       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:
              __lowercase		:			List[Any]	    =       dataset
              __lowercase		:			str	    =       RagRetriever(
                  _snake_case				,       question_encoder_tokenizer=self.get_dpr_tokenizer()				,       generator_tokenizer=self.get_bart_tokenizer()				,       )
          return retriever




      def 		snake_case_			(      self		:       int				,       _snake_case		:       bool				):
          __lowercase		:			Dict	    =       self.get_dummy_dataset()
          __lowercase		:			List[Any]	    =       RagConfig(
              retrieval_vector_size=self.retrieval_vector_size				,       question_encoder=DPRConfig().to_dict()				,       generator=BartConfig().to_dict()				,       index_name='''custom'''				,       )
          if from_disk:
              __lowercase		:			List[Any]	    =       os.path.join(self.tmpdirname				,       '''dataset'''				)
              __lowercase		:			Union[str, Any]	    =       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
              __lowercase		:			Optional[Any]	    =       RagRetriever(
                  _snake_case				,       question_encoder_tokenizer=self.get_dpr_tokenizer()				,       generator_tokenizer=self.get_bart_tokenizer()				,       )
          else:
              __lowercase		:			Optional[Any]	    =       RagRetriever(
                  _snake_case				,       question_encoder_tokenizer=self.get_dpr_tokenizer()				,       generator_tokenizer=self.get_bart_tokenizer()				,       index=CustomHFIndex(config.retrieval_vector_size				,       _snake_case				)				,       )
          return retriever




      def 		snake_case_			(      self		:       str				):
          __lowercase		:			List[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				)

          __lowercase		:			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'''				)				)

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

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




      def 		snake_case_			(      self		:       Optional[Any]				):
          __lowercase		:			List[str]	    =       1
          __lowercase		:			Tuple	    =       self.get_dummy_canonical_hf_index_retriever()
          __lowercase		:			List[Any]	    =       np.array(
              [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
          __lowercase    ,      __lowercase    ,      __lowercase		:			str	    =       retriever.retrieve(_snake_case				,       n_docs=_snake_case				)
          self.assertEqual(retrieved_doc_embeds.shape				,       (2, n_docs, self.retrieval_vector_size)				)
          self.assertEqual(len(_snake_case				)				,       2				)
          self.assertEqual(sorted(doc_dicts[0]				)				,       ['''embeddings''', '''id''', '''text''', '''title''']				)
          self.assertEqual(len(doc_dicts[0]['''id''']				)				,       _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 		snake_case_			(      self		:       int				):
          __lowercase		:			int	    =       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:
                  __lowercase		:			Optional[Any]	    =       self.get_dummy_dataset()
                  retriever.save_pretrained(_snake_case				)
                  __lowercase		:			str	    =       RagRetriever.from_pretrained(_snake_case				)
                  self.assertIsInstance(_snake_case				,       _snake_case				)
                  __lowercase		:			List[str]	    =       np.array(
                      [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
                  __lowercase		:			Union[str, Any]	    =       retriever.retrieve(_snake_case				,       n_docs=1				)
                  self.assertTrue(out is not None				)




      def 		snake_case_			(      self		:       str				):
          __lowercase		:			List[str]	    =       1
          __lowercase		:			Dict	    =       self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case				)
          __lowercase		:			Optional[Any]	    =       np.array(
              [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
          __lowercase    ,      __lowercase    ,      __lowercase		:			Tuple	    =       retriever.retrieve(_snake_case				,       n_docs=_snake_case				)
          self.assertEqual(retrieved_doc_embeds.shape				,       (2, n_docs, self.retrieval_vector_size)				)
          self.assertEqual(len(_snake_case				)				,       2				)
          self.assertEqual(sorted(doc_dicts[0]				)				,       ['''embeddings''', '''id''', '''text''', '''title''']				)
          self.assertEqual(len(doc_dicts[0]['''id''']				)				,       _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 		snake_case_			(      self		:       Any				):
          __lowercase		:			List[str]	    =       self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case				)
          with tempfile.TemporaryDirectory() as tmp_dirname:
              retriever.save_pretrained(_snake_case				)
              __lowercase		:			Optional[Any]	    =       RagRetriever.from_pretrained(_snake_case				)
              self.assertIsInstance(_snake_case				,       _snake_case				)
              __lowercase		:			Optional[Any]	    =       np.array(
                  [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
              __lowercase		:			List[Any]	    =       retriever.retrieve(_snake_case				,       n_docs=1				)
              self.assertTrue(out is not None				)




      def 		snake_case_			(      self		:       List[Any]				):
          __lowercase		:			Any	    =       1
          __lowercase		:			List[Any]	    =       self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case				)
          __lowercase		:			Any	    =       np.array(
              [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
          __lowercase    ,      __lowercase    ,      __lowercase		:			Union[str, Any]	    =       retriever.retrieve(_snake_case				,       n_docs=_snake_case				)
          self.assertEqual(retrieved_doc_embeds.shape				,       (2, n_docs, self.retrieval_vector_size)				)
          self.assertEqual(len(_snake_case				)				,       2				)
          self.assertEqual(sorted(doc_dicts[0]				)				,       ['''embeddings''', '''id''', '''text''', '''title''']				)
          self.assertEqual(len(doc_dicts[0]['''id''']				)				,       _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 		snake_case_			(      self		:       Any				):
          __lowercase		:			str	    =       self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case				)
          with tempfile.TemporaryDirectory() as tmp_dirname:
              retriever.save_pretrained(_snake_case				)
              __lowercase		:			Optional[int]	    =       RagRetriever.from_pretrained(_snake_case				)
              self.assertIsInstance(_snake_case				,       _snake_case				)
              __lowercase		:			Union[str, Any]	    =       np.array(
                  [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
              __lowercase		:			Tuple	    =       retriever.retrieve(_snake_case				,       n_docs=1				)
              self.assertTrue(out is not None				)




      def 		snake_case_			(      self		:       Tuple				):
          __lowercase		:			Optional[int]	    =       1
          __lowercase		:			str	    =       self.get_dummy_legacy_index_retriever()
          __lowercase		:			Tuple	    =       np.array(
              [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
          __lowercase    ,      __lowercase    ,      __lowercase		:			Tuple	    =       retriever.retrieve(_snake_case				,       n_docs=_snake_case				)
          self.assertEqual(retrieved_doc_embeds.shape				,       (2, n_docs, self.retrieval_vector_size)				)
          self.assertEqual(len(_snake_case				)				,       2				)
          self.assertEqual(sorted(doc_dicts[0]				)				,       ['''text''', '''title''']				)
          self.assertEqual(len(doc_dicts[0]['''text''']				)				,       _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 		snake_case_			(      self		:       Union[str, Any]				):
          __lowercase		:			Tuple	    =       self.get_dummy_legacy_index_retriever()
          with tempfile.TemporaryDirectory() as tmp_dirname:
              retriever.save_pretrained(_snake_case				)
              __lowercase		:			Tuple	    =       RagRetriever.from_pretrained(_snake_case				)
              self.assertIsInstance(_snake_case				,       _snake_case				)
              __lowercase		:			Optional[int]	    =       np.array(
                  [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
              __lowercase		:			Tuple	    =       retriever.retrieve(_snake_case				,       n_docs=1				)
              self.assertTrue(out is not None				)




      @require_torch
      @require_tokenizers
      @require_sentencepiece
      def 		snake_case_			(      self		:       Optional[Any]				):
          import torch

          __lowercase		:			Tuple	    =       1
          __lowercase		:			Any	    =       self.get_dummy_canonical_hf_index_retriever()
          __lowercase		:			str	    =       [[5, 7], [10, 11]]
          __lowercase		:			List[Any]	    =       np.array(
              [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
          __lowercase		:			Any	    =       retriever(_snake_case				,       _snake_case				,       prefix=retriever.config.generator.prefix				,       n_docs=_snake_case				)
          __lowercase    ,      __lowercase    ,      __lowercase		:			Tuple	    =       (
              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(_snake_case				,       _snake_case				)
          self.assertIsInstance(_snake_case				,       _snake_case				)
          self.assertIsInstance(_snake_case				,       np.ndarray				)

          __lowercase		:			Optional[Any]	    =       retriever(
              _snake_case				,       _snake_case				,       prefix=retriever.config.generator.prefix				,       n_docs=_snake_case				,       return_tensors='''pt'''				,       )
          __lowercase    ,      __lowercase    ,      __lowercase    ,      __lowercase		:			Optional[Any]	    =       (  # 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(_snake_case				,       torch.Tensor				)
          self.assertIsInstance(_snake_case				,       torch.Tensor				)
          self.assertIsInstance(_snake_case				,       torch.Tensor				)




      @require_torch
      @require_tokenizers
      @require_sentencepiece
      def 		snake_case_			(      self		:       List[Any]				):
          __lowercase		:			Tuple	    =       self.get_dpr_ctx_encoder_tokenizer()
          __lowercase		:			str	    =       1
          __lowercase		:			int	    =       self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case				)
          retriever.set_ctx_encoder_tokenizer(_snake_case				)

          __lowercase		:			Tuple	    =       [[5, 7], [10, 11]]
          __lowercase		:			int	    =       np.array(
              [np.ones(self.retrieval_vector_size				), -np.ones(self.retrieval_vector_size				)]				,       dtype=np.floataa				)
          __lowercase		:			Any	    =       retriever(_snake_case				,       _snake_case				,       prefix=retriever.config.generator.prefix				,       n_docs=_snake_case				)

          self.assertEqual(
              len(_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''')				)				,       _snake_case				)  # check for doc token related keys in dictionary.


