





"""simple docstring"""


import logging
import random

import ray

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


lowerCamelCase__						= logging.getLogger(__name__)



class   __SCREAMING_SNAKE_CASE     :







  '''simple docstring'''


  def __init__(		self :							str						)		->					Union[str, Any]:
        _UpperCamelCase   :						Any    =							False

  def  __SCREAMING_SNAKE_CASE					(		self :							int       , __a :							List[str]       , __a :							Tuple       , __a :							Union[str, Any]       , __a :							List[Any]						)		->					Optional[Any]:
        if not self.initialized:
              _UpperCamelCase   :						int    =							RagRetriever(
                  __a       , question_encoder_tokenizer=__a       , generator_tokenizer=__a       , index=__a       , init_retrieval=__a       , )
              _UpperCamelCase   :						List[str]    =							True

  def  __SCREAMING_SNAKE_CASE					(		self :							Tuple						)		->					Tuple:
        self.retriever.index.init_index()


  def  __SCREAMING_SNAKE_CASE					(		self :							List[Any]       , __a :							Dict       , __a :							Any						)		->					Any:
        _UpperCamelCase,	_UpperCamelCase   :						List[Any]    =							self.retriever._main_retrieve(__a       , __a						)
        return doc_ids, retrieved_doc_embeds





class   __SCREAMING_SNAKE_CASE     (    _UpperCamelCase ):







  '''simple docstring'''


  def __init__(		self :							Optional[int]       , __a :							int       , __a :							Tuple       , __a :							List[str]       , __a :							Tuple       , __a :							int=None						)		->					Dict:
        if index is not None and index.is_initialized() and len(__a						) > 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__(
            __a       , question_encoder_tokenizer=__a       , generator_tokenizer=__a       , index=__a       , init_retrieval=__a       , )
        _UpperCamelCase   :						str    =							retrieval_workers
        if len(self.retrieval_workers						) > 0:
              ray.get(
                  [
                      worker.create_rag_retriever.remote(__a       , __a       , __a       , __a						)
                      for worker in self.retrieval_workers
                  ]						)

  def  __SCREAMING_SNAKE_CASE					(		self :							str						)		->					List[str]:
        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  __SCREAMING_SNAKE_CASE					(		self :							Dict       , __a :							List[str]       , __a :							str						)		->					Any:
        if len(self.retrieval_workers						) > 0:
              # Select a random retrieval actor.
              _UpperCamelCase   :						Tuple    =							self.retrieval_workers[random.randint(0       , len(self.retrieval_workers						) - 1						)]
              _UpperCamelCase,	_UpperCamelCase   :						str    =							ray.get(random_worker.retrieve.remote(__a       , __a						)						)
        else:
              _UpperCamelCase,	_UpperCamelCase   :						Optional[int]    =							self._main_retrieve(__a       , __a						)
        return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__a						)

  @classmethod
  def  __SCREAMING_SNAKE_CASE					(		cls :							Any       , __a :							List[str]       , __a :							Optional[Any]=None       , **__a :							List[str]						)		->					Dict:
        return super(__a       , cls						).get_tokenizers(__a       , __a       , **__a						)


  @classmethod
  def  __SCREAMING_SNAKE_CASE					(		cls :							List[Any]       , __a :							Any       , __a :							List[str]       , __a :							str=None       , **__a :							Any						)		->					int:
        _UpperCamelCase   :						Optional[Any]    =							kwargs.pop("config"       , __a						) or RagConfig.from_pretrained(__a       , **__a						)
        _UpperCamelCase   :						Any    =							RagTokenizer.from_pretrained(__a       , config=__a						)
        _UpperCamelCase   :						List[Any]    =							rag_tokenizer.question_encoder
        _UpperCamelCase   :						Any    =							rag_tokenizer.generator
        if indexed_dataset is not None:
              _UpperCamelCase   :						int    =							"custom"
              _UpperCamelCase   :						Tuple    =							CustomHFIndex(config.retrieval_vector_size       , __a						)
        else:
              _UpperCamelCase   :						Optional[Any]    =							cls._build_index(__a						)
        return cls(
            __a       , question_encoder_tokenizer=__a       , generator_tokenizer=__a       , retrieval_workers=__a       , index=__a       , )







"""simple docstring"""


from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding



class   __SCREAMING_SNAKE_CASE     (    _UpperCamelCase ):







  '''simple docstring'''


  SCREAMING_SNAKE_CASE__				:Optional[Any]						  =       ["image_processor", "tokenizer"]
  SCREAMING_SNAKE_CASE__				:Dict						  =       "AutoImageProcessor"
  SCREAMING_SNAKE_CASE__				:Optional[int]						  =       "AutoTokenizer"

  def __init__(		self :							Dict       , __a :							Tuple       , __a :							str						)		->					Optional[Any]:
        super().__init__(__a       , __a						)
        _UpperCamelCase   :						Tuple    =							self.image_processor

  def __call__(		self :							Tuple       , __a :							Tuple=None       , __a :							str=None       , __a :							Any=None       , **__a :							Optional[Any]						)		->					Union[str, Any]:

        if text is None and images is None:
              raise ValueError("You have to specify either text or images. Both cannot be none."						)

        if text is not None:
              _UpperCamelCase   :						List[str]    =							self.tokenizer(__a       , return_tensors=__a       , **__a						)

        if images is not None:
              _UpperCamelCase   :						str    =							self.image_processor(__a       , return_tensors=__a       , **__a						)

        if text is not None and images is not None:
              _UpperCamelCase   :						Optional[int]    =							image_features.pixel_values
              return encoding
        elif text is not None:
              return encoding
        else:
              return BatchEncoding(data=dict(**__a						)       , tensor_type=__a						)

  def  __SCREAMING_SNAKE_CASE					(		self :							Optional[int]       , *__a :							Optional[int]       , **__a :							Optional[int]						)		->					List[Any]:
        return self.tokenizer.batch_decode(*__a       , **__a						)

  def  __SCREAMING_SNAKE_CASE					(		self :							int       , *__a :							Any       , **__a :							List[Any]						)		->					Tuple:
        return self.tokenizer.decode(*__a       , **__a						)


  @property
  def  __SCREAMING_SNAKE_CASE					(		self :							str						)		->					Any:
        return ["input_ids", "attention_mask", "pixel_values"]
