






# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import copy
import importlib.metadata
import json
import os
from dataclasses import dataclass
from typing import Any, Dict, Union

from packaging import version

from ..utils import is_torch_available, logging


if is_torch_available():
							import torch


A__	:			str			   = logging.get_logger(__name__)
@dataclass
class       _UpperCAmelCase					:

				"""simple docstring"""




				def __init__(     self						:    Tuple,  lowerCamelCase						:    Any=False,  lowerCamelCase						:    Tuple=False,  lowerCamelCase						:    Union[str, Any]=6.0,  lowerCamelCase						:    Any=None,  lowerCamelCase						:    Dict=False,  lowerCamelCase						:    Any=False,  lowerCamelCase						:    Any=None,  lowerCamelCase						:    Tuple="fp4",  lowerCamelCase						:    Optional[Any]=False,  **lowerCamelCase						:    Optional[int],  ):

								'''simple docstring'''
								lowercase__    =			load_in_abit
								lowercase__    =			load_in_abit
								lowercase__    =			llm_inta_threshold
								lowercase__    =			llm_inta_skip_modules
								lowercase__    =			llm_inta_enable_fpaa_cpu_offload
								lowercase__    =			llm_inta_has_fpaa_weight
								lowercase__    =			bnb_abit_quant_type
								lowercase__    =			bnb_abit_use_double_quant

								if bnb_abit_compute_dtype is None:
												lowercase__    =			torch.floataa
								elif isinstance(lowerCamelCase,  lowerCamelCase  ):
												lowercase__    =			getattr(lowerCamelCase,  lowerCamelCase  )
								elif isinstance(lowerCamelCase,  torch.dtype  ):
												lowercase__    =			bnb_abit_compute_dtype
								else:
												raise ValueError('''bnb_4bit_compute_dtype must be a string or a torch.dtype'''  )

								self.post_init()




				def 				lowercase__   (     self						:    str  ):

								'''simple docstring'''
								if not isinstance(self.llm_inta_threshold,  lowerCamelCase  ):
												raise ValueError('''llm_int8_threshold must be a float'''  )

								if self.llm_inta_skip_modules is not None and not isinstance(self.llm_inta_skip_modules,  lowerCamelCase  ):
												raise ValueError('''llm_int8_skip_modules must be a list of strings'''  )
								if not isinstance(self.llm_inta_enable_fpaa_cpu_offload,  lowerCamelCase  ):
												raise ValueError('''llm_int8_enable_fp32_cpu_offload must be a boolean'''  )

								if not isinstance(self.llm_inta_has_fpaa_weight,  lowerCamelCase  ):
												raise ValueError('''llm_int8_has_fp16_weight must be a boolean'''  )

								if self.bnb_abit_compute_dtype is not None and not isinstance(self.bnb_abit_compute_dtype,  torch.dtype  ):
												raise ValueError('''bnb_4bit_compute_dtype must be torch.dtype'''  )

								if not isinstance(self.bnb_abit_quant_type,  lowerCamelCase  ):
												raise ValueError('''bnb_4bit_quant_type must be a string'''  )

								if not isinstance(self.bnb_abit_use_double_quant,  lowerCamelCase  ):
												raise ValueError('''bnb_4bit_use_double_quant must be a boolean'''  )

								if self.load_in_abit and not version.parse(importlib.metadata.version('''bitsandbytes'''  )  ) >= version.parse(
								    '''0.39.0'''  ):
												raise ValueError(
												    '''4 bit quantization requires bitsandbytes>=0.39.0 - please upgrade your bitsandbytes version'''  )




				def 				lowercase__   (     self						:    int  ):

								'''simple docstring'''
								return self.load_in_abit or self.load_in_abit




				def 				lowercase__   (     self						:    int  ):

								'''simple docstring'''
								if self.load_in_abit:
												return "llm_int8"
								elif self.load_in_abit and self.bnb_abit_quant_type == "fp4":
												return "fp4"
								elif self.load_in_abit and self.bnb_abit_quant_type == "nf4":
												return "nf4"
								else:
												return None




				@classmethod
				def 				lowercase__   (     cls						:    int,  lowerCamelCase						:    List[str],  lowerCamelCase						:    Optional[Any],  **lowerCamelCase						:    int  ):

								'''simple docstring'''
								lowercase__    =			cls(**lowerCamelCase  )

								lowercase__    =			[]
								for key, value in kwargs.items():
												if hasattr(lowerCamelCase,  lowerCamelCase  ):
																setattr(lowerCamelCase,  lowerCamelCase,  lowerCamelCase  )
																to_remove.append(lowerCamelCase  )
								for key in to_remove:
												kwargs.pop(lowerCamelCase,  lowerCamelCase  )

								if return_unused_kwargs:
												return config, kwargs
								else:
												return config




				def 				lowercase__   (     self						:    Optional[Any],  lowerCamelCase						:    Union[str, os.PathLike]  ):

								'''simple docstring'''
								with open(lowerCamelCase,  '''w''',  encoding='''utf-8'''  ) as writer:
												lowercase__    =			self.to_dict()
												lowercase__    =			json.dumps(lowerCamelCase,  indent=2,  sort_keys=lowerCamelCase  ) + '''\n'''

												writer.write(lowerCamelCase  )




				def 				lowercase__   (     self						:    Optional[Any]  ):

								'''simple docstring'''
								lowercase__    =			copy.deepcopy(self.__dict__  )
								lowercase__    =			str(output['''bnb_4bit_compute_dtype''']  ).split('''.'''  )[1]

								return output




				def __repr__(     self						:    int  ):

								'''simple docstring'''
								return F"""{self.__class__.__name__} {self.to_json_string()}"""




				def 				lowercase__   (     self						:    str,  lowerCamelCase						:    bool = True  ):

								'''simple docstring'''
								if use_diff is True:
												lowercase__    =			self.to_diff_dict()
								else:
												lowercase__    =			self.to_dict()
								return json.dumps(lowerCamelCase,  indent=2,  sort_keys=lowerCamelCase  ) + "\n"




				def 				lowercase__   (     self						:    Optional[int]  ):

								'''simple docstring'''
								lowercase__    =			self.to_dict()

								# get the default config dict
								lowercase__    =			BitsAndBytesConfig().to_dict()

								lowercase__    =			{}

								# only serialize values that differ from the default config
								for key, value in config_dict.items():
												if value != default_config_dict[key]:
																lowercase__    =			value

								return serializable_config_dict












from typing import Any, Dict, List, Optional, Tuple, Union

import torch
from torch import nn
from torch.utils.data import DistributedSampler, RandomSampler

from transformers import PreTrainedModel, Trainer, logging
from transformers.integrations import is_fairscale_available
from transformers.models.fsmt.configuration_fsmt import FSMTConfig
from transformers.optimization import (
    Adafactor,
    AdamW,
    get_constant_schedule,
    get_constant_schedule_with_warmup,
    get_cosine_schedule_with_warmup,
    get_cosine_with_hard_restarts_schedule_with_warmup,
    get_linear_schedule_with_warmup,
    get_polynomial_decay_schedule_with_warmup,
)
from transformers.trainer_pt_utils import get_tpu_sampler
from transformers.training_args import ParallelMode
from transformers.utils import is_torch_tpu_available


if is_fairscale_available():
							from fairscale.optim import OSS


A__	:			Tuple			   = logging.get_logger(__name__)

A__	:			int			   = {
    'linear': get_linear_schedule_with_warmup,
    'cosine': get_cosine_schedule_with_warmup,
    'cosine_w_restarts': get_cosine_with_hard_restarts_schedule_with_warmup,
    'polynomial': get_polynomial_decay_schedule_with_warmup,
    'constant': get_constant_schedule,
    'constant_w_warmup': get_constant_schedule_with_warmup,
}
class       _UpperCAmelCase					(       A__	):

				"""simple docstring"""




				def __init__(     self						:    Optional[int],  lowerCamelCase						:    int=None,  lowerCamelCase						:    int=None,  *lowerCamelCase						:    List[Any],  **lowerCamelCase						:    Any  ):

								'''simple docstring'''
								super().__init__(*lowerCamelCase,  **lowerCamelCase  )

								if config is None:
												assert isinstance(self.model,  lowerCamelCase  ), (
												    "If no `config` is passed the model to be trained has to be of type `PreTrainedModel`, but is"
												    F""" {self.model.__class__}"""
												)
												lowercase__    =			self.model.config
								else:
												lowercase__    =			config

								lowercase__    =			data_args
								lowercase__    =			self.config.tgt_vocab_size if isinstance(self.config,  lowerCamelCase  ) else self.config.vocab_size

								if self.args.label_smoothing != 0 or (self.data_args is not None and self.data_args.ignore_pad_token_for_loss):
												assert self.config.pad_token_id is not None, (
												    "Make sure that `config.pad_token_id` is correcly defined when ignoring `pad_token` for loss"
												    " calculation or doing label smoothing."
												)

								if self.config.pad_token_id is None and self.config.eos_token_id is not None:
												logger.warning(
												    F"""The `config.pad_token_id` is `None`. Using `config.eos_token_id` = {self.config.eos_token_id} for"""
												    ''' padding..'''  )

								if self.args.label_smoothing == 0:
												lowercase__    =			torch.nn.CrossEntropyLoss(ignore_index=self.config.pad_token_id  )
								else:
												# dynamically import label_smoothed_nll_loss
												from utils import label_smoothed_nll_loss

												lowercase__    =			label_smoothed_nll_loss




				def 				lowercase__   (     self						:    List[Any],  lowerCamelCase						:    int  ):

								'''simple docstring'''
								if self.optimizer is None:
												lowercase__    =			['''bias''', '''LayerNorm.weight''']
												lowercase__    =			[
												    {
												        '''params''': [p for n, p in self.model.named_parameters() if not any(nd in n for nd in no_decay  )],
												        '''weight_decay''': self.args.weight_decay,
												    },
												    {
												        '''params''': [p for n, p in self.model.named_parameters() if any(nd in n for nd in no_decay  )],
												        '''weight_decay''': 0.0,
												    },
												]
												lowercase__    =			Adafactor if self.args.adafactor else AdamW
												if self.args.adafactor:
																lowercase__    =			Adafactor
																lowercase__    =			{'''scale_parameter''': False, '''relative_step''': False}
												else:
																lowercase__    =			AdamW
																lowercase__    =			{
																    '''betas''': (self.args.adam_betaa, self.args.adam_betaa),
																    '''eps''': self.args.adam_epsilon,
																}
												lowercase__    =			self.args.learning_rate
												if self.sharded_ddp:
																lowercase__    =			OSS(
																    params=lowerCamelCase,  optim=lowerCamelCase,  **lowerCamelCase,  )
												else:
																lowercase__    =			optimizer_cls(lowerCamelCase,  **lowerCamelCase  )

								if self.lr_scheduler is None:
												lowercase__    =			self._get_lr_scheduler(lowerCamelCase  )
								else:  # ignoring --lr_scheduler
												logger.warning('''scheduler is passed to `Seq2SeqTrainer`, `--lr_scheduler` arg is ignored.'''  )




				def 				lowercase__   (     self						:    List[str],  lowerCamelCase						:    Optional[int]  ):

								'''simple docstring'''
								lowercase__    =			arg_to_scheduler[self.args.lr_scheduler]
								if self.args.lr_scheduler == "constant":
												lowercase__    =			schedule_func(self.optimizer  )
								elif self.args.lr_scheduler == "constant_w_warmup":
												lowercase__    =			schedule_func(self.optimizer,  num_warmup_steps=self.args.warmup_steps  )
								else:
												lowercase__    =			schedule_func(
												    self.optimizer,  num_warmup_steps=self.args.warmup_steps,  num_training_steps=lowerCamelCase  )
								return scheduler




				def 				lowercase__   (     self						:    List[Any]  ):

								'''simple docstring'''
								if isinstance(self.train_dataset,  torch.utils.data.IterableDataset  ):
												return None
								elif is_torch_tpu_available():
												return get_tpu_sampler(self.train_dataset  )
								else:
												if self.args.sortish_sampler:
																self.train_dataset.make_sortish_sampler(
																    self.args.per_device_train_batch_size,  distributed=(self.args.parallel_mode == ParallelMode.DISTRIBUTED),  )

												return (
												    RandomSampler(self.train_dataset  )
												    if self.args.local_rank == -1
												    else DistributedSampler(self.train_dataset  )
												)




				def 				lowercase__   (     self						:    Any,  lowerCamelCase						:    List[Any],  lowerCamelCase						:    str,  lowerCamelCase						:    Union[str, Any]  ):

								'''simple docstring'''
								if self.args.label_smoothing == 0:
												if self.data_args is not None and self.data_args.ignore_pad_token_for_loss:
																# force training to ignore pad token
																lowercase__    =			model(**lowerCamelCase,  use_cache=lowerCamelCase  )[0]
																lowercase__    =			self.loss_fn(logits.view(-1,  logits.shape[-1]  ),  labels.view(-1  )  )
												else:
																# compute usual loss via models
																lowercase__ ,						lowercase__    =			model(**lowerCamelCase,  labels=lowerCamelCase,  use_cache=lowerCamelCase  )[:2]
								else:
												# compute label smoothed loss
												lowercase__    =			model(**lowerCamelCase,  use_cache=lowerCamelCase  )[0]
												lowercase__    =			torch.nn.functional.log_softmax(lowerCamelCase,  dim=-1  )
												lowercase__ ,						lowercase__    =			self.loss_fn(lowerCamelCase,  lowerCamelCase,  self.args.label_smoothing,  ignore_index=self.config.pad_token_id  )
								return loss, logits




				def 				lowercase__   (     self						:    List[str],  lowerCamelCase						:    Optional[Any],  lowerCamelCase						:    Tuple  ):

								'''simple docstring'''
								lowercase__    =			inputs.pop('''labels'''  )
								lowercase__ ,						lowercase__    =			self._compute_loss(lowerCamelCase,  lowerCamelCase,  lowerCamelCase  )
								return loss




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

								'''simple docstring'''
								lowercase__    =			self._prepare_inputs(lowerCamelCase  )

								lowercase__    =			{
								    '''max_length''': self.data_args.val_max_target_length
								    if self.data_args is not None
								    else self.config.max_length,
								    '''num_beams''': self.data_args.eval_beams if self.data_args is not None else self.config.num_beams,
								}

								if self.args.predict_with_generate and not self.args.prediction_loss_only:
												lowercase__    =			self.model.generate(
												    inputs['''input_ids'''],  attention_mask=inputs['''attention_mask'''],  **lowerCamelCase,  )
												# in case the batch is shorter than max length, the output should be padded
												if generated_tokens.shape[-1] < gen_kwargs["max_length"]:
																lowercase__    =			self._pad_tensors_to_max_len(lowerCamelCase,  gen_kwargs['''max_length''']  )

								lowercase__    =			inputs.pop('''labels'''  )
								with torch.no_grad():
												# compute loss on predict data
												lowercase__ ,						lowercase__    =			self._compute_loss(lowerCamelCase,  lowerCamelCase,  lowerCamelCase  )

								lowercase__    =			loss.mean().detach()
								if self.args.prediction_loss_only:
												return (loss, None, None)

								lowercase__    =			generated_tokens if self.args.predict_with_generate else logits

								if labels.shape[-1] < gen_kwargs["max_length"]:
												lowercase__    =			self._pad_tensors_to_max_len(lowerCamelCase,  gen_kwargs['''max_length''']  )

								return (loss, logits, labels)




				def 				lowercase__   (     self						:    List[Any],  lowerCamelCase						:    str,  lowerCamelCase						:    Any  ):

								'''simple docstring'''
								# If PAD token is not defined at least EOS token has to be defined
								lowercase__    =			self.config.pad_token_id if self.config.pad_token_id is not None else self.config.eos_token_id

								if pad_token_id is None:
												raise ValueError(
												    '''Make sure that either `config.pad_token_id` or `config.eos_token_id` is defined if tensor has to be'''
												    F""" padded to `max_length`={max_length}"""  )

								lowercase__    =			pad_token_id * torch.ones(
								    (tensor.shape[0], max_length),  dtype=tensor.dtype,  device=tensor.device  )
								lowercase__    =			tensor
								return padded_tensor




