




from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCAmelCase__    :							str       		=     logging.get_logger(__name__)

UpperCAmelCase__    :							Optional[int]       		=     {
    'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/config.json'
    # See all CPMAnt models at https://huggingface.co/models?filter=cpmant
}

class   UpperCAmelCase    (      a__ ):
   __SCREAMING_SNAKE_CASE		:				str      					=       "cpmant"






   def __init__(						self :							str    ,       UpperCamelCase_ :							int = 3_07_20    ,       UpperCamelCase_ :							int = 40_96    ,       UpperCamelCase_ :							int = 32    ,       UpperCamelCase_ :							int = 1_28    ,       UpperCamelCase_ :							int = 1_02_40    ,       UpperCamelCase_ :							int = 48    ,       UpperCamelCase_ :							int = 0.0    ,       UpperCamelCase_ :							int = 5_12    ,       UpperCamelCase_ :							int = 20_48    ,       UpperCamelCase_ :							int = 1E-6    ,       UpperCamelCase_ :							float = 1.0    ,       UpperCamelCase_ :							int = 32    ,       UpperCamelCase_ :							int = 32    ,       UpperCamelCase_ :							int = 32    ,       UpperCamelCase_ :							bool = True    ,       **UpperCamelCase_ :							Union[str, Any]    ,       )->      Optional[int]:
     super().__init__(**lowercase__  )
     _lowercase:			str      =     prompt_types
     _lowercase:			str      =     prompt_length
     _lowercase:			Optional[int]      =     segment_types
     _lowercase:			int      =     hidden_size
     _lowercase:			List[Any]      =     num_attention_heads
     _lowercase:			List[str]      =     dim_head
     _lowercase:			List[Any]      =     dim_ff
     _lowercase:			Tuple      =     num_hidden_layers
     _lowercase:			str      =     position_bias_num_buckets
     _lowercase:			Dict      =     position_bias_max_distance
     _lowercase:			str      =     dropout_p
     _lowercase:			Optional[int]      =     eps
     _lowercase:			List[str]      =     use_cache
     _lowercase:			List[Any]      =     vocab_size
     _lowercase:			List[Any]      =     init_std






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



     def __init__(  self   :							List[str]       ):
       a_		    =  False



     def 				__magic_name__		(  self   :							Union[str, Any]      ,		lowercase__   :							List[str]      ,		lowercase__   :							Optional[int]      ,		lowercase__   :							Optional[Any]      ,		lowercase__   :							Union[str, Any]       ):
       if not self.initialized:
         a_		    =  RagRetriever(
             lowercase__      ,		question_encoder_tokenizer=lowercase__      ,		generator_tokenizer=lowercase__      ,		index=lowercase__      ,		init_retrieval=lowercase__      ,		)
         a_		    =  True



     def 				__magic_name__		(  self   :							List[Any]       ):
       self.retriever.index.init_index()



     def 				__magic_name__		(  self   :							List[Any]       ):
       # delete the old self.retriever object before assigning the new index
       del self.retriever
       a_		    =  False



     def 				__magic_name__		(  self   :							Optional[Any]      ,		lowercase__   :							List[str]      ,		lowercase__   :							Dict       ):
       a_    , a_		    =  self.retriever._main_retrieve(lowercase__      ,		lowercase__       )
       a_		    =  self.retriever.index.get_doc_dicts(lowercase__       )
       return doc_ids, retrieved_doc_embeds, doc_dicts



class 							__lowercase   (			a__    ):



     def __init__(  self   :							List[Any]      ,		lowercase__   :							Optional[int]      ,		lowercase__   :							List[str]      ,		lowercase__   :							Tuple      ,		lowercase__   :							Dict      ,		lowercase__   :							Dict=None       ):
       if index is not None and index.is_initialized() and len(lowercase__       ) > 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__(
           lowercase__      ,		question_encoder_tokenizer=lowercase__      ,		generator_tokenizer=lowercase__      ,		index=lowercase__      ,		init_retrieval=lowercase__      ,		)

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



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



     @classmethod
     def 				__magic_name__		(  cls   :							Dict      ,		lowercase__   :							Optional[Any]      ,		lowercase__   :							List[str]=None      ,		**lowercase__   :							int       ):
       return super(lowercase__      ,		cls       ).get_tokenizers(lowercase__      ,		lowercase__      ,		**lowercase__       )



     @classmethod
     def 				__magic_name__		(  cls   :							Tuple      ,		lowercase__   :							Any      ,		lowercase__   :							str      ,		lowercase__   :							List[Any]=None      ,		**lowercase__   :							Union[str, Any]       ):
       a_		    =  kwargs.pop('''config'''      ,		lowercase__       ) or RagConfig.from_pretrained(lowercase__      ,		**lowercase__       )
       a_		    =  RagTokenizer.from_pretrained(lowercase__      ,		config=lowercase__       )
       a_		    =  rag_tokenizer.question_encoder
       a_		    =  rag_tokenizer.generator

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

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



     def 				__magic_name__		(  self   :							Optional[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_		    =  self._build_index(self.config       )

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