


"""simple docstring"""






# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.auto import AutoModelForSeqaSeqLM, AutoTokenizer
from .base import PipelineTool


lowercase  		=			"""Here is a text containing a lot of information: '''{text}'''.

Can you answer this question about the text: '{question}'"""


class 			SCREAMING_SNAKE_CASE_  (   _lowercase):




							'''simple docstring'''





							__magic_name__					:     List[Any]				  =  '''google/flan-t5-base'''
							__magic_name__					:     Dict				  =  (
							    '''This is a tool that answers questions related to a text. It takes two arguments named `text`, which is the '''
							    '''text where to find the answer, and `question`, which is the question, and returns the answer to the question.'''
							)
							__magic_name__					:     Union[str, Any]				  =  '''text_qa'''
							__magic_name__					:     Union[str, Any]				  =  AutoTokenizer
							__magic_name__					:     List[Any]				  =  AutoModelForSeqaSeqLM

							__magic_name__					:     Tuple				  =  ['''text''', '''text''']
							__magic_name__					:     Tuple				  =  ['''text''']




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__)   ->	Optional[int]:







								'''simple docstring'''




								snake_case__  :						str		=      QA_PROMPT.format(text=lowerCamelCase__	,					question=lowerCamelCase__)
								return self.pre_processor(lowerCamelCase__	,					return_tensors="pt")




							def        UpperCAmelCase (						self	,					lowerCamelCase__)   ->	Optional[int]:







								'''simple docstring'''




								snake_case__  :						Tuple		=      self.model.generate(**lowerCamelCase__)

								snake_case__,					snake_case__  :						Dict		=      inputs["input_ids"].shape
								snake_case__  :						List[Any]		=      output_ids.shape[0]

								return output_ids.reshape(lowerCamelCase__	,					out_b // in_b	,					*output_ids.shape[1:])[0][0]







							def        UpperCAmelCase (						self	,					lowerCamelCase__)   ->	List[Any]:







								'''simple docstring'''




								return self.pre_processor.decode(lowerCamelCase__	,					skip_special_tokens=lowerCamelCase__	,					clean_up_tokenization_spaces=lowerCamelCase__)













import logging
import random

import ray

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


__UpperCAmelCase      =				logging.getLogger(__name__)
class 	a_:



     """simple docstring"""



     def __init__(		self					:	Any)							->	Optional[int]:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		False





     def     __UpperCamelCase						(		self					:	Dict  ,  lowerCAmelCase__					:	Union[str, Any]  ,  lowerCAmelCase__					:	List[Any]  ,  lowerCAmelCase__					:	Optional[Any]  ,  lowerCAmelCase__					:	List[Any])							->	Optional[Any]:

      """simple docstring"""
      if not self.initialized:
       SCREAMING_SNAKE_CASE				    =		RagRetriever(
           lowerCAmelCase__  ,  question_encoder_tokenizer=lowerCAmelCase__  ,  generator_tokenizer=lowerCAmelCase__  ,  index=lowerCAmelCase__  ,  init_retrieval=lowerCAmelCase__  ,  )
       SCREAMING_SNAKE_CASE				    =		True





     def     __UpperCamelCase						(		self					:	List[str])							->	Any:

      """simple docstring"""
      self.retriever.index.init_index()





     def     __UpperCamelCase						(		self					:	int)							->	Dict:

      """simple docstring"""
      # delete the old self.retriever object before assigning the new index
      del self.retriever
      SCREAMING_SNAKE_CASE				    =		False





     def     __UpperCamelCase						(		self					:	List[str]  ,  lowerCAmelCase__					:	Any  ,  lowerCAmelCase__					:	Optional[int])							->	Union[str, Any]:

      """simple docstring"""
      SCREAMING_SNAKE_CASE    ,   SCREAMING_SNAKE_CASE				    =		self.retriever._main_retrieve(lowerCAmelCase__  ,  lowerCAmelCase__)
      SCREAMING_SNAKE_CASE				    =		self.retriever.index.get_doc_dicts(lowerCAmelCase__)
      return doc_ids, retrieved_doc_embeds, doc_dicts

class 	a_(		lowercase__	):



     """simple docstring"""



     def __init__(		self					:	List[str]  ,  lowerCAmelCase__					:	Tuple  ,  lowerCAmelCase__					:	Any  ,  lowerCAmelCase__					:	Union[str, Any]  ,  lowerCAmelCase__					:	Optional[Any]  ,  lowerCAmelCase__					:	Optional[Any]=None)							->	Tuple:

      """simple docstring"""
      if index is not None and index.is_initialized() and len(lowerCAmelCase__) > 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__(
          lowerCAmelCase__  ,  question_encoder_tokenizer=lowerCAmelCase__  ,  generator_tokenizer=lowerCAmelCase__  ,  index=lowerCAmelCase__  ,  init_retrieval=lowerCAmelCase__  ,  )

      SCREAMING_SNAKE_CASE				    =		retrieval_workers
      SCREAMING_SNAKE_CASE				    =		question_encoder_tokenizer
      SCREAMING_SNAKE_CASE				    =		generator_tokenizer
      if len(self.retrieval_workers) > 0:
       ray.get(
           [
               worker.create_rag_retriever.remote(lowerCAmelCase__  ,  lowerCAmelCase__  ,  lowerCAmelCase__  ,  lowerCAmelCase__)
               for worker in self.retrieval_workers
           ])





     def     __UpperCamelCase						(		self					:	Any)							->	Dict:

      """simple docstring"""
      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     __UpperCamelCase						(		self					:	Optional[int]  ,  lowerCAmelCase__					:	List[str]  ,  lowerCAmelCase__					:	Union[str, Any])							->	List[str]:

      """simple docstring"""
      if len(self.retrieval_workers) > 0:
       # Select a random retrieval actor.
       SCREAMING_SNAKE_CASE				    =		self.retrieval_workers[random.randint(0  ,  len(self.retrieval_workers) - 1)]
       SCREAMING_SNAKE_CASE    ,   SCREAMING_SNAKE_CASE    ,   SCREAMING_SNAKE_CASE				    =		ray.get(
           random_worker.retrieve.remote(lowerCAmelCase__  ,  lowerCAmelCase__))
      else:
       SCREAMING_SNAKE_CASE    ,   SCREAMING_SNAKE_CASE				    =		self._main_retrieve(lowerCAmelCase__  ,  lowerCAmelCase__)
       SCREAMING_SNAKE_CASE				    =		self.index.get_doc_dicts(lowerCAmelCase__)
      return retrieved_doc_embeds, doc_ids, doc_dicts





     @classmethod
     def     __UpperCamelCase						(		cls					:	List[str]  ,  lowerCAmelCase__					:	Optional[int]  ,  lowerCAmelCase__					:	Any=None  ,  **lowerCAmelCase__					:	Optional[Any])							->	Union[str, Any]:

      """simple docstring"""
      return super(lowerCAmelCase__  ,  cls).get_tokenizers(lowerCAmelCase__  ,  lowerCAmelCase__  ,  **lowerCAmelCase__)





     @classmethod
     def     __UpperCamelCase						(		cls					:	Any  ,  lowerCAmelCase__					:	Dict  ,  lowerCAmelCase__					:	List[Any]  ,  lowerCAmelCase__					:	Optional[Any]=None  ,  **lowerCAmelCase__					:	str)							->	Dict:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		kwargs.pop('config'  ,  lowerCAmelCase__) or RagConfig.from_pretrained(lowerCAmelCase__  ,  **lowerCAmelCase__)
      SCREAMING_SNAKE_CASE				    =		RagTokenizer.from_pretrained(lowerCAmelCase__  ,  config=lowerCAmelCase__)
      SCREAMING_SNAKE_CASE				    =		rag_tokenizer.question_encoder
      SCREAMING_SNAKE_CASE				    =		rag_tokenizer.generator

      if indexed_dataset is not None:
       SCREAMING_SNAKE_CASE				    =		'custom'
       SCREAMING_SNAKE_CASE				    =		CustomHFIndex(config.retrieval_vector_size  ,  lowerCAmelCase__)
      else:
       SCREAMING_SNAKE_CASE				    =		cls._build_index(lowerCAmelCase__)

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





     def     __UpperCamelCase						(		self					:	List[Any])							->	Union[str, Any]:

      """simple docstring"""
      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
      SCREAMING_SNAKE_CASE				    =		self._build_index(self.config)

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




