




'''simple docstring'''




from copy import deepcopy
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn
from torch.utils.data import Dataset

from .deepspeed import is_deepspeed_zeroa_enabled
from .generation.configuration_utils import GenerationConfig
from .trainer import Trainer
from .utils import logging


if TYPE_CHECKING:
			from .data.data_collator import DataCollator
			from .modeling_utils import PreTrainedModel
			from .tokenization_utils_base import PreTrainedTokenizerBase
			from .trainer_callback import TrainerCallback
			from .trainer_utils import EvalPrediction, PredictionOutput
			from .training_args import TrainingArguments


snake_case__          =						logging.get_logger(__name__)






class     UpperCamelCase_     (lowerCamelCase_     ):







						"""simple docstring"""


						def __init__(     self      :       Any							, _lowerCamelCase      :       Tuple = None							, _lowerCamelCase      :       Optional[Any] = None							, _lowerCamelCase      :       str = None							, _lowerCamelCase      :       Tuple = None							, _lowerCamelCase      :       Union[str, Any] = None							, _lowerCamelCase      :       Optional[Any] = None							, _lowerCamelCase      :       Tuple = None							, _lowerCamelCase      :       Optional[Any] = None							, _lowerCamelCase      :       Union[str, Any] = None							, _lowerCamelCase      :       Optional[int] = (None, None)							, _lowerCamelCase      :       int = None							, ):





											"""simple docstring"""







											super().__init__(
											    model=_lowerCamelCase							, args=_lowerCamelCase							, data_collator=_lowerCamelCase							, train_dataset=_lowerCamelCase							, eval_dataset=_lowerCamelCase							, tokenizer=_lowerCamelCase							, model_init=_lowerCamelCase							, compute_metrics=_lowerCamelCase							, callbacks=_lowerCamelCase							, optimizers=_lowerCamelCase							, preprocess_logits_for_metrics=_lowerCamelCase							, )

											# Override self.model.generation_config if a GenerationConfig is specified in args.
											# Priority: args.generation_config > model.generation_config > default GenerationConfig.
											if self.args.generation_config is not None:
																A_ :     Any =							self.load_generation_config(self.args.generation_config    )
																A_ :     List[Any] =							gen_config







						@staticmethod
						def       _a    (     _lowerCamelCase      :       Tuple    ):





											"""simple docstring"""







											if isinstance(_lowerCamelCase							, _lowerCamelCase    ):
																return deepcopy(_lowerCamelCase    )

											# str or Path
											A_ :     Dict =							Path(_lowerCamelCase    ) if isinstance(_lowerCamelCase							, _lowerCamelCase    ) else gen_config_arg
											A_ :     Tuple =							None

											# Figuring if it is path pointing to a file, pointing to a directory or else a model id or URL
											# This step is required in order to determine config_file_name
											if pretrained_model_name.is_file():
																A_ :     List[Any] =							pretrained_model_name.name
																A_ :     int =							pretrained_model_name.parent
											# dir path
											elif pretrained_model_name.is_dir():
																pass
											# model id or URL
											else:
																A_ :     List[Any] =							gen_config_arg

											A_ :     Tuple =							GenerationConfig.from_pretrained(_lowerCamelCase							, _lowerCamelCase    )
											return gen_config







						def       _a    (     self      :       Optional[Any]							, _lowerCamelCase      :       Optional[Any] = None							, _lowerCamelCase      :       Tuple = None							, _lowerCamelCase      :       int = "eval"							, **_lowerCamelCase      :       List[str]							, ):





											"""simple docstring"""







											A_ :     Tuple =							gen_kwargs.copy()
											if gen_kwargs.get('''max_length'''    ) is None and gen_kwargs.get('''max_new_tokens'''    ) is None:
																A_ :     Optional[Any] =							self.args.generation_max_length
											A_ :     str =							(
											    gen_kwargs["num_beams"] if gen_kwargs.get('''num_beams'''    ) is not None else self.args.generation_num_beams
											)
											A_ :     Dict =							gen_kwargs

											return super().evaluate(_lowerCamelCase							, ignore_keys=_lowerCamelCase							, metric_key_prefix=_lowerCamelCase    )







						def       _a    (     self      :       Optional[Any]							, _lowerCamelCase      :       Union[str, Any]							, _lowerCamelCase      :       List[Any] = None							, _lowerCamelCase      :       Dict = "test"							, **_lowerCamelCase      :       Tuple							, ):





											"""simple docstring"""







											A_ :     Dict =							gen_kwargs.copy()
											if gen_kwargs.get('''max_length'''    ) is None and gen_kwargs.get('''max_new_tokens'''    ) is None:
																A_ :     Tuple =							self.args.generation_max_length
											A_ :     str =							(
											    gen_kwargs["num_beams"] if gen_kwargs.get('''num_beams'''    ) is not None else self.args.generation_num_beams
											)
											A_ :     Tuple =							gen_kwargs

											return super().predict(_lowerCamelCase							, ignore_keys=_lowerCamelCase							, metric_key_prefix=_lowerCamelCase    )







						def       _a    (     self      :       List[Any]							, _lowerCamelCase      :       int							, _lowerCamelCase      :       Any							, _lowerCamelCase      :       Tuple							, _lowerCamelCase      :       Optional[Any] = None							, **_lowerCamelCase      :       List[Any]							, ):





											"""simple docstring"""







											if not self.args.predict_with_generate or prediction_loss_only:
																return super().prediction_step(
																    _lowerCamelCase							, _lowerCamelCase							, prediction_loss_only=_lowerCamelCase							, ignore_keys=_lowerCamelCase    )

											A_ :     Any =							"labels" in inputs
											A_ :     Dict =							self._prepare_inputs(_lowerCamelCase    )

											# XXX: adapt synced_gpus for fairscale as well
											# Priority (handled in generate):
											# gen_kwargs > model.generation_config > default GenerationConfig()

											if len(_lowerCamelCase    ) == 0 and hasattr(self							, '''_gen_kwargs'''    ):
																A_ :     Optional[Any] =							self._gen_kwargs.copy()

											if gen_kwargs.get('''max_length'''    ) is None and gen_kwargs.get('''max_new_tokens'''    ) is None:
																A_ :     str =							self.model.config.max_length
											A_ :     Optional[int] =							(
											    gen_kwargs["num_beams"] if gen_kwargs.get('''num_beams'''    ) is not None else self.model.config.num_beams
											)
											A_ :     Union[str, Any] =							True if is_deepspeed_zeroa_enabled() else False
											A_ :     int =							(
											    gen_kwargs["synced_gpus"] if gen_kwargs.get('''synced_gpus'''    ) is not None else default_synced_gpus
											)

											# If the `decoder_input_ids` was created from `labels`, evict the former, so that the model can freely generate
											# (otherwise, it would continue generating from the padded `decoder_input_ids`)
											if (
											    "labels" in inputs
											    and "decoder_input_ids" in inputs
											    and inputs["labels"].shape == inputs["decoder_input_ids"].shape
											):
																A_ :     Optional[int] =							{k: v for k, v in inputs.items() if k != "decoder_input_ids"}
											A_ :     List[Any] =							self.model.generate(**_lowerCamelCase							, **_lowerCamelCase    )

											# Temporary hack to ensure the generation config is not initialized for each iteration of the evaluation loop
											# TODO: remove this hack when the legacy code that initializes generation_config from a model config is
											# removed in https://github.com/huggingface/transformers/blob/98d88b23f54e5a23e741833f1e973fdf600cc2c5/src/transformers/generation/utils.py#L1183
											if self.model.generation_config._from_model_config:
																A_ :     str =							False

											# Retrieves GenerationConfig from model.generation_config
											A_ :     Optional[Any] =							self.model.generation_config
											# in case the batch is shorter than max length, the output should be padded
											if generated_tokens.shape[-1] < gen_config.max_length:
																A_ :     str =							self._pad_tensors_to_max_len(_lowerCamelCase							, gen_config.max_length    )
											elif gen_config.max_new_tokens is not None and generated_tokens.shape[-1] < gen_config.max_new_tokens + 1:
																A_ :     str =							self._pad_tensors_to_max_len(_lowerCamelCase							, gen_config.max_new_tokens + 1    )

											with torch.no_grad():
																if has_labels:
																					with self.compute_loss_context_manager():
																										A_ :     Optional[Any] =							model(**_lowerCamelCase    )
																					if self.label_smoother is not None:
																										A_ :     Union[str, Any] =							self.label_smoother(_lowerCamelCase							, inputs['''labels''']    ).mean().detach()
																					else:
																										A_ :     Tuple =							(outputs["loss"] if isinstance(_lowerCamelCase							, _lowerCamelCase    ) else outputs[0]).mean().detach()
																else:
																					A_ :     Dict =							None

											if self.args.prediction_loss_only:
																return loss, None, None

											if has_labels:
																A_ :     List[str] =							inputs["labels"]
																if labels.shape[-1] < gen_config.max_length:
																					A_ :     int =							self._pad_tensors_to_max_len(_lowerCamelCase							, gen_config.max_length    )
																elif gen_config.max_new_tokens is not None and labels.shape[-1] < gen_config.max_new_tokens + 1:
																					A_ :     Optional[Any] =							self._pad_tensors_to_max_len(_lowerCamelCase							, gen_config.max_new_tokens + 1    )
											else:
																A_ :     Union[str, Any] =							None

											return loss, generated_tokens, labels







						def       _a    (     self      :       Optional[int]							, _lowerCamelCase      :       Tuple							, _lowerCamelCase      :       Optional[Any]    ):





											"""simple docstring"""







											if self.tokenizer is not None and hasattr(self.tokenizer							, '''pad_token_id'''    ):
																# If PAD token is not defined at least EOS token has to be defined
																A_ :     List[str] =							(
																    self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else self.tokenizer.eos_token_id
																)
											else:
																if self.model.config.pad_token_id is not None:
																					A_ :     Any =							self.model.config.pad_token_id
																else:
																					raise ValueError('''Pad_token_id must be set in the configuration of the model, in order to pad tensors'''    )

											A_ :     List[str] =							pad_token_id * torch.ones(
											    (tensor.shape[0], max_length)							, dtype=tensor.dtype							, device=tensor.device    )
											A_ :     Union[str, Any] =							tensor
											return padded_tensor









'''simple docstring'''




import logging
import random

import ray

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


snake_case__          =						logging.getLogger(__name__)






class     UpperCamelCase_     :







						"""simple docstring"""


						def __init__(     self      :       Tuple    ):





											"""simple docstring"""







											A_ :     Optional[int] =							False







						def       _a    (     self      :       Any							, _lowerCamelCase      :       List[str]							, _lowerCamelCase      :       Any							, _lowerCamelCase      :       str							, _lowerCamelCase      :       int    ):





											"""simple docstring"""







											if not self.initialized:
																A_ :     Optional[Any] =							RagRetriever(
																    _lowerCamelCase							, question_encoder_tokenizer=_lowerCamelCase							, generator_tokenizer=_lowerCamelCase							, index=_lowerCamelCase							, init_retrieval=_lowerCamelCase							, )
																A_ :     Optional[Any] =							True







						def       _a    (     self      :       int    ):





											"""simple docstring"""







											self.retriever.index.init_index()







						def       _a    (     self      :       Any    ):





											"""simple docstring"""







											del self.retriever
											A_ :     List[Any] =							False







						def       _a    (     self      :       Optional[int]							, _lowerCamelCase      :       Union[str, Any]							, _lowerCamelCase      :       Union[str, Any]    ):





											"""simple docstring"""







											A_ ,A_ :     Dict =							self.retriever._main_retrieve(_lowerCamelCase							, _lowerCamelCase    )
											A_ :     int =							self.retriever.index.get_doc_dicts(_lowerCamelCase    )
											return doc_ids, retrieved_doc_embeds, doc_dicts






class     UpperCamelCase_     (a__     ):







						"""simple docstring"""


						def __init__(     self      :       int							, _lowerCamelCase      :       Optional[Any]							, _lowerCamelCase      :       Dict							, _lowerCamelCase      :       Optional[int]							, _lowerCamelCase      :       Dict							, _lowerCamelCase      :       Tuple=None    ):





											"""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							, )

											A_ :     Tuple =							retrieval_workers
											A_ :     Any =							question_encoder_tokenizer
											A_ :     Dict =							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      :       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       _a    (     self      :       Dict							, _lowerCamelCase      :       Tuple							, _lowerCamelCase      :       int    ):





											"""simple docstring"""







											if len(self.retrieval_workers    ) > 0:
																# Select a random retrieval actor.
																A_ :     Optional[int] =							self.retrieval_workers[random.randint(0							, len(self.retrieval_workers    ) - 1    )]
																A_ ,A_ ,A_ :     Union[str, Any] =							ray.get(
																    random_worker.retrieve.remote(_lowerCamelCase							, _lowerCamelCase    )    )
											else:
																A_ ,A_ :     List[str] =							self._main_retrieve(_lowerCamelCase							, _lowerCamelCase    )
																A_ :     Union[str, Any] =							self.index.get_doc_dicts(_lowerCamelCase    )
											return retrieved_doc_embeds, doc_ids, doc_dicts







						@classmethod
						def       _a    (     cls      :       List[str]							, _lowerCamelCase      :       Union[str, Any]							, _lowerCamelCase      :       Optional[Any]=None							, **_lowerCamelCase      :       str    ):





											"""simple docstring"""







											return super(_lowerCamelCase							, cls    ).get_tokenizers(_lowerCamelCase							, _lowerCamelCase							, **_lowerCamelCase    )







						@classmethod
						def       _a    (     cls      :       Optional[Any]							, _lowerCamelCase      :       Any							, _lowerCamelCase      :       List[Any]							, _lowerCamelCase      :       str=None							, **_lowerCamelCase      :       Tuple    ):





											"""simple docstring"""







											A_ :     Tuple =							kwargs.pop('''config'''							, _lowerCamelCase    ) or RagConfig.from_pretrained(_lowerCamelCase							, **_lowerCamelCase    )
											A_ :     List[Any] =							RagTokenizer.from_pretrained(_lowerCamelCase							, config=_lowerCamelCase    )
											A_ :     Union[str, Any] =							rag_tokenizer.question_encoder
											A_ :     Dict =							rag_tokenizer.generator

											if indexed_dataset is not None:
																A_ :     Any =							'''custom'''
																A_ :     int =							CustomHFIndex(config.retrieval_vector_size							, _lowerCamelCase    )
											else:
																A_ :     Optional[int] =							cls._build_index(_lowerCamelCase    )

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







						def       _a    (     self      :       Optional[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
											A_ :     Dict =							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
											    ]    )



