code: 
  
  
  
  
  
  import warnings
  from dataclasses import dataclass, field
  from typing import Optional, Tuple
  
  from .training_args import TrainingArguments
  from .utils import cached_property, is_tf_available, logging, requires_backends
  
  
  _lowerCamelCase				:	str				    =       logging.get_logger(__name__)
  
  if is_tf_available():
  				import tensorflow as tf
  
  
  
  @dataclass
  class 							__UpperCAmelCase  (       lowerCAmelCase_						):
  
  
  
  
  
  
  
  		'''simple docstring'''
  
  
  
  
  
  
  
  		__lowerCAmelCase 				=    'tf'
  		__lowerCAmelCase 				=    field(
  		    default=lowerCAmelCase_		,					metadata={'''help''': '''Name of TPU'''}		,					)
  
  		__lowerCAmelCase 				=    field(
  		    default=lowerCAmelCase_		,					metadata={'''help''': '''Zone of TPU'''}		,					)
  
  		__lowerCAmelCase 				=    field(
  		    default=lowerCAmelCase_		,					metadata={'''help''': '''Name of Cloud TPU-enabled project'''}		,					)
  
  		__lowerCAmelCase 				=    field(
  		    default=1.0		,					metadata={'''help''': '''Power for the Polynomial decay LR scheduler.'''}		,					)
  
  		__lowerCAmelCase 				=    field(default=lowerCAmelCase_		,					metadata={'''help''': '''Whether to activate the XLA compilation or not'''}						)
  
  
  
  
  
  
  
  		@cached_property
  		def        A     (self						:      Tuple      ):
  			requires_backends(self      ,      ["""tf"""]      )
  			logger.info("""Tensorflow: setting up strategy"""      )
  
  			A							  =      tf.config.list_physical_devices("""GPU"""      )
  
  			# Set to float16 at first
  			if self.fpaa:
  				tf.keras.mixed_precision.set_global_policy("""mixed_float16"""      )
  
  			if self.no_cuda:
  				A							  =      tf.distribute.OneDeviceStrategy(device="""/cpu:0"""      )
  			else:
  				try:
  					if self.tpu_name:
  						A							  =      tf.distribute.cluster_resolver.TPUClusterResolver(
  						    self.tpu_name      ,      zone=self.tpu_zone      ,      project=self.gcp_project      )
  					else:
  						A							  =      tf.distribute.cluster_resolver.TPUClusterResolver()
  				except ValueError:
  					if self.tpu_name:
  						raise RuntimeError(F"""Couldn\'t connect to TPU {self.tpu_name}!"""      )
  					else:
  						A							  =      None
  
  				if tpu:
  					# Set to bfloat16 in case of TPU
  					if self.fpaa:
  						tf.keras.mixed_precision.set_global_policy("""mixed_bfloat16"""      )
  
  					tf.config.experimental_connect_to_cluster(lowerCamelCase_      )
  					tf.tpu.experimental.initialize_tpu_system(lowerCamelCase_      )
  
  					A							  =      tf.distribute.TPUStrategy(lowerCamelCase_      )
  
  				elif len(lowerCamelCase_      ) == 0:
  					A							  =      tf.distribute.OneDeviceStrategy(device="""/cpu:0"""      )
  				elif len(lowerCamelCase_      ) == 1:
  					A							  =      tf.distribute.OneDeviceStrategy(device="""/gpu:0"""      )
  				elif len(lowerCamelCase_      ) > 1:
  					# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
  					A							  =      tf.distribute.MirroredStrategy()
  				else:
  					raise ValueError("""Cannot find the proper strategy, please check your environment properties."""      )
  
  			return strategy
  
  
  
  
  
  
  
  		@property
  		def        A     (self						:      str      ):
  			requires_backends(self      ,      ["""tf"""]      )
  			return self._setup_strategy
  
  
  
  
  
  
  
  		@property
  		def        A     (self						:      Tuple      ):
  			requires_backends(self      ,      ["""tf"""]      )
  			return self._setup_strategy.num_replicas_in_sync
  
  
  
  
  
  
  
  		@property
  		def        A     (self						:      Optional[Any]      ):
  			return False  # TF Logging is handled by Keras not the Trainer
  
  
  
  
  
  
  
  		@property
  		def        A     (self						:      Union[str, Any]      ):
  
  			if self.per_gpu_train_batch_size:
  				logger.warning(
  				    """Using deprecated `--per_gpu_train_batch_size` argument which will be removed in a future """
  				    """version. Using `--per_device_train_batch_size` is preferred."""      )
  			A							  =      self.per_gpu_train_batch_size or self.per_device_train_batch_size
  			return per_device_batch_size * self.n_replicas
  
  
  
  
  
  
  
  		@property
  		def        A     (self						:      Any      ):
  
  			if self.per_gpu_eval_batch_size:
  				logger.warning(
  				    """Using deprecated `--per_gpu_eval_batch_size` argument which will be removed in a future """
  				    """version. Using `--per_device_eval_batch_size` is preferred."""      )
  			A							  =      self.per_gpu_eval_batch_size or self.per_device_eval_batch_size
  			return per_device_batch_size * self.n_replicas
  
  
  
  
  
  
  
  		@property
  		def        A     (self						:      List[Any]      ):
  			requires_backends(self      ,      ["""tf"""]      )
  			warnings.warn(
  			    """The n_gpu argument is deprecated and will be removed in a future version, use n_replicas instead."""      ,      lowerCamelCase_      ,      )
  			return self._setup_strategy.num_replicas_in_sync
  
  
style_context: 
  
  
  
  
  
  '''simple docstring'''
  
  
  import logging
  import random
  
  import ray
  
  from transformers import RagConfig, RagRetriever, RagTokenizer
  from transformers.models.rag.retrieval_rag import CustomHFIndex
  
  
  _lowerCamelCase				:	Tuple				    =       logging.getLogger(__name__)
  
  
  
  class 							__UpperCAmelCase  :
  
  
  
  
  
  
  
  		'''simple docstring'''
  
  
  
  
  
  
  
  		def __init__(self						:      Union[str, Any]      ):
  			A							  =      False
  
  
  
  
  
  
  
  		def        A     (self						:      Any      ,      _lowerCAmelCase						:      str      ,      _lowerCAmelCase						:      List[str]      ,      _lowerCAmelCase						:      str      ,      _lowerCAmelCase						:      Any      ):
  			if not self.initialized:
  				A							  =      RagRetriever(
  				    _lowerCAmelCase      ,      question_encoder_tokenizer=_lowerCAmelCase      ,      generator_tokenizer=_lowerCAmelCase      ,      index=_lowerCAmelCase      ,      init_retrieval=_lowerCAmelCase      ,      )
  				A							  =      True
  
  
  
  
  
  
  
  		def        A     (self						:      Optional[int]      ):
  			self.retriever.index.init_index()
  
  
  
  
  
  
  
  		def        A     (self						:      Optional[int]      ):
  			# delete the old self.retriever object before assigning the new index
  			del self.retriever
  			A							  =      False
  
  
  
  
  
  
  
  		def        A     (self						:      List[str]      ,      _lowerCAmelCase						:      int      ,      _lowerCAmelCase						:      List[str]      ):
  			A  ,    A							  =      self.retriever._main_retrieve(_lowerCAmelCase      ,      _lowerCAmelCase      )
  			A							  =      self.retriever.index.get_doc_dicts(_lowerCAmelCase      )
  			return doc_ids, retrieved_doc_embeds, doc_dicts
  
  
  
  
  
  
  class 							__UpperCAmelCase  (       A__						):
  
  
  
  
  
  
  
  		'''simple docstring'''
  
  
  
  
  
  
  
  		def __init__(self						:      Tuple      ,      _lowerCAmelCase						:      int      ,      _lowerCAmelCase						:      int      ,      _lowerCAmelCase						:      List[str]      ,      _lowerCAmelCase						:      Tuple      ,      _lowerCAmelCase						:      str=None      ):
  			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      ,      )
  
  			A							  =      retrieval_workers
  			A							  =      question_encoder_tokenizer
  			A							  =      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        A     (self						:      List[Any]      ):
  			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        A     (self						:      str      ,      _lowerCAmelCase						:      Dict      ,      _lowerCAmelCase						:      Tuple      ):
  			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(_lowerCAmelCase      ,      _lowerCAmelCase      )      )
  			else:
  				A  ,    A							  =      self._main_retrieve(_lowerCAmelCase      ,      _lowerCAmelCase      )
  				A							  =      self.index.get_doc_dicts(_lowerCAmelCase      )
  			return retrieved_doc_embeds, doc_ids, doc_dicts
  
  
  
  
  
  
  
  		@classmethod
  		def        A     (cls						:      Union[str, Any]      ,      _lowerCAmelCase						:      int      ,      _lowerCAmelCase						:      Optional[Any]=None      ,      **_lowerCAmelCase						:      Union[str, Any]      ):
  			return super(_lowerCAmelCase      ,      cls      ).get_tokenizers(_lowerCAmelCase      ,      _lowerCAmelCase      ,      **_lowerCAmelCase      )
  
  
  
  
  
  
  
  		@classmethod
  		def        A     (cls						:      List[Any]      ,      _lowerCAmelCase						:      Optional[int]      ,      _lowerCAmelCase						:      List[str]      ,      _lowerCAmelCase						:      str=None      ,      **_lowerCAmelCase						:      List[str]      ):
  			A							  =      kwargs.pop("""config"""      ,      _lowerCAmelCase      ) or RagConfig.from_pretrained(_lowerCAmelCase      ,      **_lowerCAmelCase      )
  			A							  =      RagTokenizer.from_pretrained(_lowerCAmelCase      ,      config=_lowerCAmelCase      )
  			A							  =      rag_tokenizer.question_encoder
  			A							  =      rag_tokenizer.generator
  
  			if indexed_dataset is not None:
  				A							  =      """custom"""
  				A							  =      CustomHFIndex(config.retrieval_vector_size      ,      _lowerCAmelCase      )
  			else:
  				A							  =      cls._build_index(_lowerCAmelCase      )
  
  			return cls(
  			    _lowerCAmelCase      ,      question_encoder_tokenizer=_lowerCAmelCase      ,      generator_tokenizer=_lowerCAmelCase      ,      retrieval_workers=_lowerCAmelCase      ,      index=_lowerCAmelCase      ,      )
  
  
  
  
  
  
  
  		def        A     (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							  =      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
  			    ]      )
  
  