




'''simple docstring'''

# This test is meant to be run in on an instance with TPUs like this:
#
#   python examples/pytorch/xla_spawn.py --num_cores=8 tests/test_trainer_tpu.py
#
# Replace 8 with the number of TPU cores you have.
#

import sys
from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.utils import logging


a__         =     logging.get_logger(__name__)


if is_torch_available():
		import torch
		from torch import nn
		from torch.utils.data import Dataset

		from transformers import Trainer





		class __magic_name__(						a__      ):






								def __init__(   self						:						str					,   __UpperCamelCase						:						int = 1_0_1       ):



											'''simple docstring'''






											snake_case__          =						length






								def __len__(   self						:						Optional[int]       ):



											'''simple docstring'''






											return self.length







								def __getitem__(   self						:						List[Any]					,   __UpperCamelCase						:						int       ):



											'''simple docstring'''






											return i







		class __magic_name__:






								def __call__(   self						:						int					,   __UpperCamelCase						:						List[Any]       ):



											'''simple docstring'''






											return {"input_ids": torch.tensor(lowercase__       ), "labels": torch.tensor(lowercase__       )}







		class __magic_name__(						nn.Module      ):






								def __init__(   self						:						int       ):



											'''simple docstring'''






											super().__init__()
											# Add some (unused) params otherwise DDP will complain.
											snake_case__          =						nn.Linear(1_2_0					,   8_0       )







								def       __lowerCAmelCase(   self						:						Dict					,   __UpperCamelCase						:						Dict					,   __UpperCamelCase						:						Union[str, Any]=None       ):



											'''simple docstring'''






											if labels is not None:
														return torch.tensor(0.0					,   device=input_ids.device       ), input_ids
											else:
														return input_ids




def    snake_case__     (  )			->       Union[str, Any]:





			'''simple docstring'''






			snake_case__          =						HfArgumentParser((TrainingArguments,)	)
			sys.argv += ["--output_dir", "./examples"]
			snake_case__          =						parser.parse_args_into_dataclasses()[0]

			logger.warning(
			    F"""Process rank: {training_args.local_rank}, device: {training_args.device}, """
			    F"""tpu_num_cores: {training_args.tpu_num_cores}"""       ,      )

			# Essentially, what we want to verify in the distributed case is
			# that we get all samples back, in the right order.
			# (this is crucial for prediction for instance)
			for dataset_length in [1001, 256, 15]:
						snake_case__          =						DummyDataset(_A	)

						def compute_metrics(a	) -> Dict:
									snake_case__          =						list(range(len(_A	)	)	)
									snake_case__          =						p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
									return {"success": success}

						snake_case__          =						Trainer(
						    model=DummyModel()       ,      args=_A       ,      data_collator=DummyDataCollator()       ,      eval_dataset=_A       ,      compute_metrics=_A       ,      )
						snake_case__          =						trainer.evaluate()
						logger.info(_A	)
						if metrics["eval_success"] is not True:
									logger.error(_A	)
									exit(1	)

						snake_case__          =						trainer.predict(_A	)
						logger.info(p.metrics	)
						if p.metrics["test_success"] is not True:
									logger.error(p.metrics	)
									exit(1	)

						snake_case__          =						2

						snake_case__          =						trainer.evaluate()
						logger.info(_A	)
						if metrics["eval_success"] is not True:
									logger.error(_A	)
									exit(1	)

						snake_case__          =						trainer.predict(_A	)
						logger.info(p.metrics	)
						if p.metrics["test_success"] is not True:
									logger.error(p.metrics	)
									exit(1	)

						snake_case__          =						None

			logger.info("""🔥 All distributed tests successful"""	)




def    snake_case__     (  a	)			->       Dict:





			'''simple docstring'''






			main()


if __name__ == "__main__":
		main()





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
           ]       )
