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


lowercase		:						Union[str, Any] =					logging.get_logger(__name__)




class    _snake_case  (						_lowercase       ):






							def __init__(					self						:			Union[str, Any]						, snake_case__						:			Union["PreTrainedModel", nn.Module] = None						, snake_case__						:			"TrainingArguments" = None						, snake_case__						:			Optional["DataCollator"] = None						, snake_case__						:			Optional[Dataset] = None						, snake_case__						:			Optional[Union[Dataset, Dict[str, Dataset]]] = None						, snake_case__						:			Optional["PreTrainedTokenizerBase"] = None						, snake_case__						:			Optional[Callable[[], "PreTrainedModel"]] = None						, snake_case__						:			Optional[Callable[["EvalPrediction"], Dict]] = None						, snake_case__						:			Optional[List["TrainerCallback"]] = None						, snake_case__						:			Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (None, None)						, snake_case__						:			Optional[Callable[[torch.Tensor, torch.Tensor], torch.Tensor]] = None						, ):






												"""simple docstring"""



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

												# 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:						str		    =					self.load_generation_config(self.args.generation_config    )
																	__a:						Dict		    =					gen_config






							@staticmethod
							def 							__lowercase  (					snake_case__						:			Union[str, GenerationConfig]    ):






												"""simple docstring"""



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

												# str or Path
												__a:						Any		    =					Path(snake_case__    ) if isinstance(snake_case__						, snake_case__    ) else gen_config_arg
												__a:						List[str]		    =					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:						Optional[Any]		    =					pretrained_model_name.name
																	__a:						Tuple		    =					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:						Optional[int]		    =					GenerationConfig.from_pretrained(snake_case__						, snake_case__    )
												return gen_config






							def 							__lowercase  (					self						:			str						, snake_case__						:			Optional[Dataset] = None						, snake_case__						:			Optional[List[str]] = None						, snake_case__						:			str = "eval"						, **snake_case__						:			Optional[Any]						, ):






												"""simple docstring"""



												__a:						Optional[int]		    =					gen_kwargs.copy()
												if gen_kwargs.get("""max_length"""    ) is None and gen_kwargs.get("""max_new_tokens"""    ) is None:
																	__a:						List[str]		    =					self.args.generation_max_length
												__a:						Optional[Any]		    =					(
												    gen_kwargs["""num_beams"""] if gen_kwargs.get("""num_beams"""    ) is not None else self.args.generation_num_beams
												)
												__a:						int		    =					gen_kwargs

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






							def 							__lowercase  (					self						:			Any						, snake_case__						:			Dataset						, snake_case__						:			Optional[List[str]] = None						, snake_case__						:			str = "test"						, **snake_case__						:			int						, ):






												"""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:						Optional[Any]		    =					(
												    gen_kwargs["""num_beams"""] if gen_kwargs.get("""num_beams"""    ) is not None else self.args.generation_num_beams
												)
												__a:						List[Any]		    =					gen_kwargs

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






							def 							__lowercase  (					self						:			Any						, snake_case__						:			nn.Module						, snake_case__						:			Dict[str, Union[torch.Tensor, Any]]						, snake_case__						:			bool						, snake_case__						:			Optional[List[str]] = None						, **snake_case__						:			str						, ):






												"""simple docstring"""



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

												__a:						List[Any]		    =					"""labels""" in inputs
												__a:						Union[str, Any]		    =					self._prepare_inputs(snake_case__    )

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

												if len(snake_case__    ) == 0 and hasattr(self						, """_gen_kwargs"""    ):
																	__a:						int		    =					self._gen_kwargs.copy()

												if gen_kwargs.get("""max_length"""    ) is None and gen_kwargs.get("""max_new_tokens"""    ) is None:
																	__a:						int		    =					self.model.config.max_length
												__a:						str		    =					(
												    gen_kwargs["""num_beams"""] if gen_kwargs.get("""num_beams"""    ) is not None else self.model.config.num_beams
												)
												__a:						Optional[Any]		    =					True if is_deepspeed_zeroa_enabled() else False
												__a:						Optional[Any]		    =					(
												    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:						Tuple		    =					{k: v for k, v in inputs.items() if k != """decoder_input_ids"""}
												__a:						Dict		    =					self.model.generate(**snake_case__						, **snake_case__    )

												# 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:						Optional[int]		    =					False

												# Retrieves GenerationConfig from model.generation_config
												__a:						int		    =					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(snake_case__						, 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:						int		    =					self._pad_tensors_to_max_len(snake_case__						, gen_config.max_new_tokens + 1    )

												with torch.no_grad():
																	if has_labels:
																						with self.compute_loss_context_manager():
																											__a:						Any		    =					model(**snake_case__    )
																						if self.label_smoother is not None:
																											__a:						List[Any]		    =					self.label_smoother(snake_case__						, inputs["""labels"""]    ).mean().detach()
																						else:
																											__a:						Tuple		    =					(outputs["""loss"""] if isinstance(snake_case__						, snake_case__    ) else outputs[0]).mean().detach()
																	else:
																						__a:						str		    =					None

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

												if has_labels:
																	__a:						Optional[int]		    =					inputs["""labels"""]
																	if labels.shape[-1] < gen_config.max_length:
																						__a:						Dict		    =					self._pad_tensors_to_max_len(snake_case__						, gen_config.max_length    )
																	elif gen_config.max_new_tokens is not None and labels.shape[-1] < gen_config.max_new_tokens + 1:
																						__a:						Tuple		    =					self._pad_tensors_to_max_len(snake_case__						, gen_config.max_new_tokens + 1    )
												else:
																	__a:						Tuple		    =					None

												return loss, generated_tokens, labels






							def 							__lowercase  (					self						:			Dict						, snake_case__						:			Any						, snake_case__						:			str    ):






												"""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:						Dict		    =					(
																	    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:						Union[str, 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:						Union[str, Any]		    =					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"""
from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_tokenizers_available,
    is_torch_available,
)


lowercase		:						List[str] =					{
    'configuration_open_llama': ['OPEN_LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP', 'OpenLlamaConfig'],
}

try:
			if not is_sentencepiece_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowercase		:						Tuple =					['LlamaTokenizer']

try:
			if not is_tokenizers_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowercase		:						Dict =					['LlamaTokenizerFast']

try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowercase		:						List[Any] =					[
			    'OpenLlamaForCausalLM',
			    'OpenLlamaModel',
			    'OpenLlamaPreTrainedModel',
			    'OpenLlamaForSequenceClassification',
			]


if TYPE_CHECKING:
			from .configuration_open_llama import OPEN_LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP, OpenLlamaConfig

			try:
						if not is_sentencepiece_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from transformers import LlamaTokenizer

			try:
						if not is_tokenizers_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from transformers import LlamaTokenizerFast

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_open_llama import (
						    OpenLlamaForCausalLM,
						    OpenLlamaForSequenceClassification,
						    OpenLlamaModel,
						    OpenLlamaPreTrainedModel,
						)


else:
			import sys

			lowercase		:						Dict =					_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)
