
"""simple docstring"""


import copy

from ...configuration_utils import PretrainedConfig
from ...utils import add_start_docstrings


_lowercase							      = r'''
    [`RagConfig`] stores the configuration of a *RagModel*. Configuration objects inherit from [`PretrainedConfig`] and
    can be used to control the model outputs. Read the documentation from [`PretrainedConfig`] for more information.

    Args:
        title_sep (`str`, *optional*, defaults to  `" / "`):
            Separator inserted between the title and the text of the retrieved document when calling [`RagRetriever`].
        doc_sep (`str`, *optional*, defaults to  `" // "`):
            Separator inserted between the text of the retrieved document and the original input when calling
            [`RagRetriever`].
        n_docs (`int`, *optional*, defaults to 5):
            Number of documents to retrieve.
        max_combined_length (`int`, *optional*, defaults to 300):
            Max length of contextualized input returned by [`~RagRetriever.__call__`].
        retrieval_vector_size (`int`, *optional*, defaults to 768):
            Dimensionality of the document embeddings indexed by [`RagRetriever`].
        retrieval_batch_size (`int`, *optional*, defaults to 8):
            Retrieval batch size, defined as the number of queries issues concurrently to the faiss index encapsulated
            [`RagRetriever`].
        dataset (`str`, *optional*, defaults to `"wiki_dpr"`):
            A dataset identifier of the indexed dataset in HuggingFace Datasets (list all available datasets and ids
            using `datasets.list_datasets()`).
        dataset_split (`str`, *optional*, defaults to `"train"`)
            Which split of the `dataset` to load.
        index_name (`str`, *optional*, defaults to `"compressed"`)
            The index name of the index associated with the `dataset`. One can choose between `"legacy"`, `"exact"` and
            `"compressed"`.
        index_path (`str`, *optional*)
            The path to the serialized faiss index on disk.
        passages_path (`str`, *optional*):
            A path to text passages compatible with the faiss index. Required if using
            [`~models.rag.retrieval_rag.LegacyIndex`]
        use_dummy_dataset (`bool`, *optional*, defaults to `False`)
            Whether to load a "dummy" variant of the dataset specified by `dataset`.
        label_smoothing (`float`, *optional*, defaults to 0.0):
            Only relevant if `return_loss` is set to `True`. Controls the `epsilon` parameter value for label smoothing
            in the loss calculation. If set to 0, no label smoothing is performed.
        do_marginalize (`bool`, *optional*, defaults to `False`):
            If `True`, the logits are marginalized over all documents by making use of
            `torch.nn.functional.log_softmax`.
        reduce_loss (`bool`, *optional*, defaults to `False`):
            Whether or not to reduce the NLL loss using the `torch.Tensor.sum` operation.
        do_deduplication (`bool`, *optional*, defaults to `True`):
            Whether or not to deduplicate the generations from different context documents for a given input. Has to be
            set to `False` if used while training with distributed backend.
        exclude_bos_score (`bool`, *optional*, defaults to `False`):
            Whether or not to disregard the BOS token when computing the loss.
        output_retrieved(`bool`, *optional*, defaults to `False`):
            If set to `True`, `retrieved_doc_embeds`, `retrieved_doc_ids`, `context_input_ids` and
            `context_attention_mask` are returned. See returned tensors for more detail.
        use_cache (`bool`, *optional*, defaults to `True`):
            Whether or not the model should return the last key/values attentions (not used by all models).
        forced_eos_token_id (`int`, *optional*):
            The id of the token to force as the last generated token when `max_length` is reached. Usually set to
            `eos_token_id`.
'''







@add_start_docstrings(__snake_case			)
class  lowerCAmelCase_ (					__snake_case			):



						'''simple docstring'''


						_lowerCamelCase:			Dict					=			"""rag"""
						_lowerCamelCase:			List[str]					=			True


						def __init__(       self   :					Dict     ,A_   :					Tuple=None     ,A_   :					Any=True     ,A_   :					Any=None     ,A_   :					int=None     ,A_   :					List[str]=None     ,A_   :					Optional[int]=None     ,A_   :					Optional[int]=None     ,A_   :					str=" / "     ,A_   :					str=" // "     ,A_   :					Any=5     ,A_   :					Optional[int]=300     ,A_   :					Any=768     ,A_   :					List[Any]=8     ,A_   :					Dict="wiki_dpr"     ,A_   :					Tuple="train"     ,A_   :					Any="compressed"     ,A_   :					List[Any]=None     ,A_   :					Tuple=None     ,A_   :					List[str]=False     ,A_   :					Union[str, Any]=False     ,A_   :					str=0.0     ,A_   :					List[Any]=True     ,A_   :					Union[str, Any]=False     ,A_   :					int=False     ,A_   :					Dict=False     ,A_   :					Optional[Any]=True     ,A_   :					Any=None     ,**A_   :					Optional[int]     ,)   ->      Tuple:
							super().__init__(
							    bos_token_id=A_     ,pad_token_id=A_     ,eos_token_id=A_     ,decoder_start_token_id=A_     ,forced_eos_token_id=A_     ,is_encoder_decoder=A_     ,prefix=A_     ,vocab_size=A_     ,**A_     ,)
							assert (
							    "question_encoder" in kwargs and "generator" in kwargs
							), "Config has to be initialized with question_encoder and generator config"
							A        =      kwargs.pop('question_encoder'					)
							A        =      question_encoder_config.pop('model_type'					)
							A        =      kwargs.pop('generator'					)
							A        =      decoder_config.pop('model_type'					)

							from ..auto.configuration_auto import AutoConfig

							A        =      AutoConfig.for_model(A_     ,**A_					)
							A        =      AutoConfig.for_model(A_     ,**A_					)

							A        =      reduce_loss
							A        =      label_smoothing
							A        =      exclude_bos_score
							A        =      do_marginalize

							A        =      title_sep
							A        =      doc_sep
							A        =      n_docs
							A        =      max_combined_length

							A        =      dataset
							A        =      dataset_split
							A        =      index_name

							A        =      retrieval_vector_size
							A        =      retrieval_batch_size
							A        =      passages_path
							A        =      index_path
							A        =      use_dummy_dataset

							A        =      output_retrieved

							A        =      do_deduplication

							A        =      use_cache

							if self.forced_eos_token_id is None:
								A        =      getattr(self.generator     ,'forced_eos_token_id'     ,A_					)


						@classmethod
						def   _SCREAMING_SNAKE_CASE					(       cls   :					int     ,A_   :					int     ,A_   :					List[str]     ,**A_   :					Optional[int]					)   ->      Tuple:
							return cls(question_encoder=question_encoder_config.to_dict()     ,generator=generator_config.to_dict()     ,**A_					)




						def   _SCREAMING_SNAKE_CASE					(       self   :					Tuple					)   ->      int:
							A        =      copy.deepcopy(self.__dict__					)
							A        =      self.question_encoder.to_dict()
							A        =      self.generator.to_dict()
							A        =      self.__class__.model_type
							return output


from __future__ import annotations

from collections.abc import Callable
from typing import Generic, TypeVar

snake_case_			  = TypeVar('''T''')
snake_case_			  = TypeVar('''U''')


class       SCREAMING_SNAKE_CASE__					(Generic[T, U]     ):





    def __init__(      self						,   a						,   a):
         lowercase__    :   List[Any]      =       key
         lowercase__    :   List[Any]      =       val
         lowercase__    :   DoubleLinkedListNode[T, U] | None      =       None
         lowercase__    :   DoubleLinkedListNode[T, U] | None      =       None





    def __repr__(      self):
         return (
             f"""Node: key: {self.key}, val: {self.val}, """
             f"""has next: {bool(self.next)}, has prev: {bool(self.prev)}"""
         )



class       SCREAMING_SNAKE_CASE__					(Generic[T, U]     ):





    def __init__(      self):
         lowercase__    :   DoubleLinkedListNode[T, U]      =       DoubleLinkedListNode(a						,   a)
         lowercase__    :   DoubleLinkedListNode[T, U]      =       DoubleLinkedListNode(a						,   a)
         lowercase__      , lowercase__    :   Union[str, Any]      =       self.rear, self.head





    def __repr__(      self):
         lowercase__    :   Any      =       ['DoubleLinkedList']
         lowercase__    :   List[str]      =       self.head
         while node.next is not None:
              rep.append(str(a))
              lowercase__    :   Tuple      =       node.next
         rep.append(str(self.rear))
         return ",\n    ".join(a)





    def       snake_case_				(      self						,   a):
         lowercase__    :   Optional[Any]      =       self.rear.prev

         # All nodes other than self.head are guaranteed to have non-None previous
         assert previous is not None

         lowercase__    :   Dict      =       node
         lowercase__    :   int      =       previous
         lowercase__    :   Union[str, Any]      =       node
         lowercase__    :   Optional[int]      =       self.rear





    def       snake_case_				(      self						,   a):

         if node.prev is None or node.next is None:
              return None

         lowercase__    :   Union[str, Any]      =       node.next
         lowercase__    :   Tuple      =       node.prev
         lowercase__    :   Union[str, Any]      =       None
         lowercase__    :   List[Any]      =       None
         return node



class       SCREAMING_SNAKE_CASE__					(Generic[T, U]     ):
    __lowerCamelCase						:       dict[Callable[[T], U], LRUCache[T, U]]														=     {}





    def __init__(      self						,   a):
         lowercase__    :   DoubleLinkedList[T, U]      =       DoubleLinkedList()
         lowercase__    :   Optional[Any]      =       capacity
         lowercase__    :   Union[str, Any]      =       0
         lowercase__    :   Tuple      =       0
         lowercase__    :   int      =       0
         lowercase__    :   dict[T, DoubleLinkedListNode[T, U]]      =       {}





    def __repr__(      self):
         return (
             f"""CacheInfo(hits={self.hits}, misses={self.miss}, """
             f"""capacity={self.capacity}, current size={self.num_keys})"""
         )





    def __contains__(      self						,   a):
         return key in self.cache





    def       snake_case_				(      self						,   a):
         # Note: pythonic interface would throw KeyError rather than return None

         if key in self.cache:
              self.hits += 1
              lowercase__    :   DoubleLinkedListNode[T, U]      =       self.cache[key]
              lowercase__    :   str      =       self.list.remove(self.cache[key])
              assert node == value_node

              # node is guaranteed not None because it is in self.cache
              assert node is not None
              self.list.add(a)
              return node.val
         self.miss += 1
         return None





    def       snake_case_				(      self						,   a						,   a):

         if key not in self.cache:
              if self.num_keys >= self.capacity:
                   # delete first node (oldest) when over capacity
                   lowercase__    :   Optional[int]      =       self.list.head.next

                   # guaranteed to have a non-None first node when num_keys > 0
                   # explain to type checker via assertions
                   assert first_node is not None
                   assert first_node.key is not None
                   assert (
                       self.list.remove(a) is not None
                   )  # node guaranteed to be in list assert node.key is not None

                   del self.cache[first_node.key]
                   self.num_keys -= 1
              lowercase__    :   Optional[Any]      =       DoubleLinkedListNode(a						,   a)
              self.list.add(self.cache[key])
              self.num_keys += 1

         else:
              # bump node to the end of the list, update value
              lowercase__    :   Any      =       self.list.remove(self.cache[key])
              assert node is not None  # node guaranteed to be in list
              lowercase__    :   Union[str, Any]      =       value
              self.list.add(a)





    @classmethod
    def       snake_case_				(      cls						,   a = 128):

         def cache_decorator_inner(a) -> Callable[..., U]:
              def cache_decorator_wrapper(*a) -> U:
                   if func not in cls.decorator_function_to_instance_map:
                        lowercase__    :   Dict      =       LRUCache(a)

                   lowercase__    :   str      =       cls.decorator_function_to_instance_map[func].get(args[0])
                   if result is None:
                        lowercase__    :   str      =       func(*a)
                        cls.decorator_function_to_instance_map[func].put(args[0]						,   a)
                   return result

              def cache_info() -> LRUCache[T, U]:
                   return cls.decorator_function_to_instance_map[func]

              setattr(a						,   'cache_info'						,   a)  # noqa: B010

              return cache_decorator_wrapper

         return cache_decorator_inner


if __name__ == "__main__":
       import doctest

       doctest.testmod()

