






'''simple docstring'''







import sys
from collections.abc import Mapping

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter

class _lowercase      (							TensorFormatter[Mapping, np.ndarray, Mapping]						):




							def __init__(				self:							Any     ,				UpperCamelCase__:							Optional[int]=None     ,				**UpperCamelCase__:							Union[str, Any]	):
														super().__init__(features=__UpperCAmelCase	)
														lowerCamelCase__   :      Optional[Any]			    =			np_array_kwargs




							def 						lowerCamelCase_							(				self:							str     ,				UpperCamelCase__:							Optional[Any]	):
														if isinstance(__UpperCAmelCase     ,				__UpperCAmelCase	):
																					if column and all(
																					    isinstance(__UpperCAmelCase     ,				np.ndarray	) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column	):
																												return np.stack(__UpperCAmelCase	)
																					else:
																												# don't use np.array(column, dtype=object)
																												# since it fails in certain cases
																												# see https://stackoverflow.com/q/51005699
																												lowerCamelCase__   :      Dict			    =			np.empty(len(__UpperCAmelCase	)     ,				dtype=__UpperCAmelCase	)
																												lowerCamelCase__   :      List[str]			    =			column
																												return out
														return column




							def 						lowerCamelCase_							(				self:							Dict     ,				UpperCamelCase__:							Optional[Any]	):
														if isinstance(__UpperCAmelCase     ,				(str, bytes, type(__UpperCAmelCase	))	):
																					return value
														elif isinstance(__UpperCAmelCase     ,				(np.character, np.ndarray)	) and np.issubdtype(value.dtype     ,				np.character	):
																					return value
														elif isinstance(__UpperCAmelCase     ,				np.number	):
																					return value

														lowerCamelCase__   :      List[Any]			    =			{}

														if isinstance(__UpperCAmelCase     ,				np.ndarray	) and np.issubdtype(value.dtype     ,				np.integer	):
																					lowerCamelCase__   :      List[Any]			    =			{'dtype': np.intaa}
														elif isinstance(__UpperCAmelCase     ,				np.ndarray	) and np.issubdtype(value.dtype     ,				np.floating	):
																					lowerCamelCase__   :      Dict			    =			{'dtype': np.floataa}
														elif config.PIL_AVAILABLE and "PIL" in sys.modules:
																					import PIL.Image

																					if isinstance(__UpperCAmelCase     ,				PIL.Image.Image	):
																												return np.asarray(__UpperCAmelCase     ,				**self.np_array_kwargs	)

														return np.asarray(__UpperCAmelCase     ,				**{**default_dtype, **self.np_array_kwargs}	)




							def 						lowerCamelCase_							(				self:							Tuple     ,				UpperCamelCase__:							int	):
														# support for torch, tf, jax etc.
														if config.TORCH_AVAILABLE and "torch" in sys.modules:
																					import torch

																					if isinstance(__UpperCAmelCase     ,				torch.Tensor	):
																												return self._tensorize(data_struct.detach().cpu().numpy()[()]	)
														if hasattr(__UpperCAmelCase     ,				"""__array__"""	) and not isinstance(__UpperCAmelCase     ,				(np.ndarray, np.character, np.number)	):
																					lowerCamelCase__   :      int			    =			data_struct.__array__()
														# support for nested types like struct of list of struct
														if isinstance(__UpperCAmelCase     ,				np.ndarray	):
																					if data_struct.dtype == object:
																												return self._consolidate([self.recursive_tensorize(__UpperCAmelCase	) for substruct in data_struct]	)
														if isinstance(__UpperCAmelCase     ,				(list, tuple)	):
																					return self._consolidate([self.recursive_tensorize(__UpperCAmelCase	) for substruct in data_struct]	)
														return self._tensorize(__UpperCAmelCase	)




							def 						lowerCamelCase_							(				self:							List[str]     ,				UpperCamelCase__:							Optional[int]	):
														return map_nested(self._recursive_tensorize     ,				__UpperCAmelCase     ,				map_list=__UpperCAmelCase	)




							def 						lowerCamelCase_							(				self:							Optional[Any]     ,				UpperCamelCase__:							Dict	):
														lowerCamelCase__   :      List[Any]			    =			self.numpy_arrow_extractor().extract_row(__UpperCAmelCase	)
														lowerCamelCase__   :      Any			    =			self.python_features_decoder.decode_row(__UpperCAmelCase	)
														return self.recursive_tensorize(__UpperCAmelCase	)




							def 						lowerCamelCase_							(				self:							Union[str, Any]     ,				UpperCamelCase__:							int	):
														lowerCamelCase__   :      str			    =			self.numpy_arrow_extractor().extract_column(__UpperCAmelCase	)
														lowerCamelCase__   :      List[str]			    =			self.python_features_decoder.decode_column(__UpperCAmelCase     ,				pa_table.column_names[0]	)
														lowerCamelCase__   :      Union[str, Any]			    =			self.recursive_tensorize(__UpperCAmelCase	)
														lowerCamelCase__   :      List[Any]			    =			self._consolidate(__UpperCAmelCase	)
														return column




							def 						lowerCamelCase_							(				self:							Optional[int]     ,				UpperCamelCase__:							List[str]	):
														lowerCamelCase__   :      Any			    =			self.numpy_arrow_extractor().extract_batch(__UpperCAmelCase	)
														lowerCamelCase__   :      str			    =			self.python_features_decoder.decode_batch(__UpperCAmelCase	)
														lowerCamelCase__   :      Union[str, Any]			    =			self.recursive_tensorize(__UpperCAmelCase	)
														for column_name in batch:
																					lowerCamelCase__   :      List[Any]			    =			self._consolidate(batch[column_name]	)
														return batch





"""simple docstring"""






import logging
import random

import ray

from transformers import RagConfig, RagRetriever, RagTokenizer
from transformers.models.rag.retrieval_rag import CustomHFIndex


SCREAMING_SNAKE_CASE__				:  Optional[Any]						= logging.getLogger(__name__)


class 	A_       :





  """simple docstring"""

  def __init__(						self						)				->					Dict:
     a   :	List[str]						=    False
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					List[str]:
     if not self.initialized:
        a   :	Dict						=    RagRetriever(
            __UpperCAmelCase  ,					question_encoder_tokenizer=__UpperCAmelCase  ,					generator_tokenizer=__UpperCAmelCase  ,					index=__UpperCAmelCase  ,					init_retrieval=__UpperCAmelCase  ,					)
        a   :	Dict						=    True
  def 							lowercase_     (						self						)				->					List[Any]:
     self.retriever.index.init_index()
  def 							lowercase_     (						self						)				->					List[Any]:
     # delete the old self.retriever object before assigning the new index
     del self.retriever
     a   :	Dict						=    False






  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					Optional[int]:
     a	, a   :	List[str]						=    self.retriever._main_retrieve(__UpperCAmelCase  ,					__UpperCAmelCase						)
     a   :	List[Any]						=    self.retriever.index.get_doc_dicts(__UpperCAmelCase						)
     return doc_ids, retrieved_doc_embeds, doc_dicts



class 	A_       (     _UpperCAmelCase   ):





  """simple docstring"""

  def __init__(						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase=None						)				->					Tuple:
     if index is not None and index.is_initialized() and len(__UpperCAmelCase						) > 0:
        raise ValueError(
            'When using Ray for distributed fine-tuning, '
            'you\'ll need to provide the paths instead, '
            'as the dataset and the index are loaded '
            'separately. More info in examples/rag/use_own_knowledge_dataset.py '						)

     super().__init__(
         __UpperCAmelCase  ,					question_encoder_tokenizer=__UpperCAmelCase  ,					generator_tokenizer=__UpperCAmelCase  ,					index=__UpperCAmelCase  ,					init_retrieval=__UpperCAmelCase  ,					)

     a   :	Tuple						=    retrieval_workers
     a   :	List[str]						=    question_encoder_tokenizer
     a   :	Union[str, Any]						=    generator_tokenizer
     if len(self.retrieval_workers						) > 0:
        ray.get(
            [
                worker.create_rag_retriever.remote(__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)
                for worker in self.retrieval_workers
            ]						)
  def 							lowercase_     (						self						)				->					Tuple:
     logger.info('initializing retrieval'						)

     if len(self.retrieval_workers						) > 0:
        ray.get([worker.init_retrieval.remote() for worker in self.retrieval_workers]						)
     else:
        # Non-distributed training. Load index into this same process.
        self.index.init_index()
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					Tuple:
     if len(self.retrieval_workers						) > 0:
        # Select a random retrieval actor.
        a   :	Dict						=    self.retrieval_workers[random.randint(0  ,					len(self.retrieval_workers						) - 1						)]
        a	, a	, a   :	Any						=    ray.get(
            random_worker.retrieve.remote(__UpperCAmelCase  ,					__UpperCAmelCase						)						)
     else:
        a	, a   :	str						=    self._main_retrieve(__UpperCAmelCase  ,					__UpperCAmelCase						)
        a   :	str						=    self.index.get_doc_dicts(__UpperCAmelCase						)
     return retrieved_doc_embeds, doc_ids, doc_dicts
  @classmethod
  def 							lowercase_     (						cls  ,					__UpperCAmelCase  ,					__UpperCAmelCase=None  ,					**__UpperCAmelCase						)				->					int:
     return super(__UpperCAmelCase  ,					cls						).get_tokenizers(__UpperCAmelCase  ,					__UpperCAmelCase  ,					**__UpperCAmelCase						)
  @classmethod
  def 							lowercase_     (						cls  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase=None  ,					**__UpperCAmelCase						)				->					str:
     a   :	Union[str, Any]						=    kwargs.pop('config'  ,					__UpperCAmelCase						) or RagConfig.from_pretrained(__UpperCAmelCase  ,					**__UpperCAmelCase						)
     a   :	Optional[Any]						=    RagTokenizer.from_pretrained(__UpperCAmelCase  ,					config=__UpperCAmelCase						)
     a   :	List[str]						=    rag_tokenizer.question_encoder
     a   :	int						=    rag_tokenizer.generator

     if indexed_dataset is not None:
        a   :	Optional[int]						=    'custom'
        a   :	str						=    CustomHFIndex(config.retrieval_vector_size  ,					__UpperCAmelCase						)
     else:
        a   :	Dict						=    cls._build_index(__UpperCAmelCase						)

     return cls(
         __UpperCAmelCase  ,					question_encoder_tokenizer=__UpperCAmelCase  ,					generator_tokenizer=__UpperCAmelCase  ,					retrieval_workers=__UpperCAmelCase  ,					index=__UpperCAmelCase  ,					)






  def 							lowercase_     (						self						)				->					List[Any]:
     logger.info('re-loading the new dataset with embeddings'						)
     # access from the training loop

     ray.get([worker.clear_object.remote() for worker in self.retrieval_workers]						)

     # build the index object again
     a   :	Optional[Any]						=    self._build_index(self.config						)

     ray.get(
         [
             worker.create_rag_retriever.remote(
                 self.config  ,					self.question_encoder_tokenizer  ,					self.generator_tokenizer  ,					__UpperCAmelCase						)
             for worker in self.retrieval_workers
         ]						)





