



from collections import OrderedDict
from typing import TYPE_CHECKING, Any, List, Mapping, Optional

from packaging import version


if TYPE_CHECKING:
							from ... import PreTrainedTokenizer, TensorType

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfigWithPast, PatchingSpec
from ...utils import is_torch_available, logging


SCREAMING_SNAKE_CASE__   			=   logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__   			=   {
    '''bigscience/bloom''': '''https://huggingface.co/bigscience/bloom/resolve/main/config.json''',
    '''bigscience/bloom-560m''': '''https://huggingface.co/bigscience/bloom-560m/blob/main/config.json''',
    '''bigscience/bloom-1b1''': '''https://huggingface.co/bigscience/bloom-1b1/blob/main/config.json''',
    '''bigscience/bloom-1b7''': '''https://huggingface.co/bigscience/bloom-1b7/blob/main/config.json''',
    '''bigscience/bloom-3b''': '''https://huggingface.co/bigscience/bloom-3b/blob/main/config.json''',
    '''bigscience/bloom-7b1''': '''https://huggingface.co/bigscience/bloom-7b1/blob/main/config.json''',
}


class   __lowerCAmelCase  (		UpperCAmelCase_   ):
					"""simple docstring"""






					A__					:		Optional[int]      							=     "bloom"
					A__					:		Union[str, Any]      							=     ["past_key_values"]
					A__					:		int      							=     {
					    "num_hidden_layers": "n_layer",
					    "num_attention_heads": "n_head",
					}



					def __init__(  self       :							Dict					,       _snake_case       :							Tuple=25_08_80					,       _snake_case       :							Dict=64					,       _snake_case       :							Optional[int]=2					,       _snake_case       :							int=8					,       _snake_case       :							Optional[int]=1E-5					,       _snake_case       :							List[Any]=0.02					,       _snake_case       :							Optional[Any]=True					,       _snake_case       :							Union[str, Any]=1					,       _snake_case       :							List[Any]=2					,       _snake_case       :							Optional[int]=False					,       _snake_case       :							Union[str, Any]=0.0					,       _snake_case       :							Union[str, Any]=0.0					,       _snake_case       :							int=1					,       _snake_case       :							List[Any]=False					,       **_snake_case       :							str					,       ):






								"""simple docstring"""


								A__	  =     vocab_size
								# Backward compatibility with n_embed kwarg
								A__	  =     kwargs.pop('n_embed'					,       _snake_case       )
								A__	  =     hidden_size if n_embed is None else n_embed
								A__	  =     n_layer
								A__	  =     n_head
								A__	  =     layer_norm_epsilon
								A__	  =     initializer_range
								A__	  =     use_cache
								A__	  =     pretraining_tp
								A__	  =     apply_residual_connection_post_layernorm
								A__	  =     hidden_dropout
								A__	  =     attention_dropout

								A__	  =     bos_token_id
								A__	  =     eos_token_id
								A__	  =     slow_but_exact

								super().__init__(bos_token_id=_snake_case					,       eos_token_id=_snake_case					,       **_snake_case       )







class   __lowerCAmelCase  (		UpperCAmelCase_   ):
					"""simple docstring"""






					A__					:		List[Any]      							=     version.parse("1.12"   )



					def __init__(  self       :							str					,       _snake_case       :							PretrainedConfig					,       _snake_case       :							str = "default"					,       _snake_case       :							List[PatchingSpec] = None					,       _snake_case       :							bool = False					,       ):






								"""simple docstring"""


								super().__init__(_snake_case					,       task=_snake_case					,       patching_specs=_snake_case					,       use_past=_snake_case       )
								if not getattr(self._config					,       'pad_token_id'					,       _snake_case       ):
											# TODO: how to do that better?
											A__	  =     0



					@property
					def      _a				(  self       :							Union[str, Any]       ):






								"""simple docstring"""


								A__	  =     OrderedDict({'input_ids': {0: 'batch', 1: 'sequence'}}       )
								if self.use_past:
											# BLOOM stores values on dynamic axis 2. For more details see: https://github.com/huggingface/transformers/pull/18344
											self.fill_with_past_key_values_(_snake_case					,       direction='inputs'					,       inverted_values_shape=_snake_case       )
											A__	  =     {0: 'batch', 1: 'past_sequence + sequence'}
								else:
											A__	  =     {0: 'batch', 1: 'sequence'}

								return common_inputs



					@property
					def      _a				(  self       :							List[Any]       ):






								"""simple docstring"""


								return self._config.n_layer



					@property
					def      _a				(  self       :							Union[str, Any]       ):






								"""simple docstring"""


								return self._config.n_head



					@property
					def      _a				(  self       :							Union[str, Any]       ):






								"""simple docstring"""


								return 1E-3



					def      _a				(  self       :							Dict					,       _snake_case       :							"PreTrainedTokenizer"					,       _snake_case       :							int = -1					,       _snake_case       :							int = -1					,       _snake_case       :							bool = False					,       _snake_case       :							Optional["TensorType"] = None					,       ):






								"""simple docstring"""


								A__	  =     super(_snake_case					,       self       ).generate_dummy_inputs(
								    _snake_case					,       batch_size=_snake_case					,       seq_length=_snake_case					,       is_pair=_snake_case					,       framework=_snake_case       )

								# We need to order the input in the way they appears in the forward()
								A__	  =     OrderedDict({'input_ids': common_inputs['input_ids']}       )

								# Need to add the past_keys
								if self.use_past:
											if not is_torch_available():
														raise ValueError('Cannot generate dummy past_keys inputs without PyTorch installed.'       )
											else:
														import torch

														A__  ,			A__	  =     common_inputs['input_ids'].shape
														# Not using the same length for past_key_values
														A__	  =     seqlen + 2
														A__	  =     self._config.hidden_size // self.num_attention_heads
														A__	  =     (
														    batch * self.num_attention_heads,
														    head_dim,
														    past_key_values_length,
														)
														A__	  =     (
														    batch * self.num_attention_heads,
														    past_key_values_length,
														    head_dim,
														)
														A__	  =     [
														    (torch.zeros(_snake_case       ), torch.zeros(_snake_case       )) for _ in range(self.num_layers       )
														]

								A__	  =     common_inputs['attention_mask']
								if self.use_past:
											A__	  =     ordered_inputs['attention_mask'].dtype
											A__	  =     torch.cat(
											    [ordered_inputs['attention_mask'], torch.ones(_snake_case					,       _snake_case					,       dtype=_snake_case       )]					,       dim=1       )

								return ordered_inputs



					@property
					def      _a				(  self       :							Optional[Any]       ):






								"""simple docstring"""


								return 13









import logging
from pathlib import Path

import numpy as np
import pytorch_lightning as pl
import torch
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
from pytorch_lightning.utilities import rank_zero_only
from utils_rag import save_json


def A    (							__UpperCamelCase       )    ->   Optional[int]:
			A__	  =     filter(lambda __UpperCamelCase       : p.requires_grad			,						model.parameters()       )
			A__	  =     sum([np.prod(p.size()       ) for p in model_parameters]       )
			return params


SCREAMING_SNAKE_CASE__   			=   logging.getLogger(__name__)


def A    (							__UpperCamelCase			,						__UpperCamelCase       )    ->   Dict:
			if metric == "rouge2":
						A__	  =     '{val_avg_rouge2:.4f}-{step_count}'
			elif metric == "bleu":
						A__	  =     '{val_avg_bleu:.4f}-{step_count}'
			elif metric == "em":
						A__	  =     '{val_avg_em:.4f}-{step_count}'
			elif metric == "loss":
						A__	  =     '{val_avg_loss:.4f}-{step_count}'
			else:
						raise NotImplementedError(
						    f'''seq2seq callbacks only support rouge2 and bleu, got {metric}, You can make your own by adding to this'''
						    ' function.'       )

			A__	  =     ModelCheckpoint(
			    dirpath=__UpperCamelCase			,						filename=__UpperCamelCase			,						monitor=f'''val_{metric}'''			,						mode='max'			,						save_top_k=1			,						every_n_epochs=1			,						)

			return checkpoint_callback


def A    (							__UpperCamelCase			,						__UpperCamelCase       )    ->   Any:
			return EarlyStopping(
			    monitor=f'''val_{metric}'''			,						mode='min' if 'loss' in metric else 'max'			,						patience=__UpperCamelCase			,						verbose=__UpperCamelCase			,						)






class   __lowerCAmelCase  (		pl.Callback   ):
					"""simple docstring"""






					def      _a				(  self       :							Dict					,       _snake_case       :							Union[str, Any]					,       _snake_case       :							str       ):






								"""simple docstring"""


								A__	  =     {F'''lr_group_{i}''': param['lr'] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups       )}
								pl_module.logger.log_metrics(_snake_case       )



					@rank_zero_only
					def      _a				(  self       :							Union[str, Any]					,       _snake_case       :							pl.Trainer					,       _snake_case       :							pl.LightningModule					,       _snake_case       :							str					,       _snake_case       :							Optional[Any]=True       ):






								"""simple docstring"""


								logger.info(F'''***** {type_path} results at step {trainer.global_step:05d} *****'''       )
								A__	  =     trainer.callback_metrics
								trainer.logger.log_metrics({k: v for k, v in metrics.items() if k not in ['log', 'progress_bar', 'preds']}       )
								# Log results
								A__	  =     Path(pl_module.hparams.output_dir       )
								if type_path == "test":
											A__	  =     od / 'test_results.txt'
											A__	  =     od / 'test_generations.txt'
								else:
											# this never gets hit. I prefer not to save intermediate generations, and results are in metrics.json
											# If people want this it will be easy enough to add back.
											A__	  =     od / F'''{type_path}_results/{trainer.global_step:05d}.txt'''
											A__	  =     od / F'''{type_path}_generations/{trainer.global_step:05d}.txt'''
											results_file.parent.mkdir(exist_ok=_snake_case       )
											generations_file.parent.mkdir(exist_ok=_snake_case       )
								with open(_snake_case					,       'a+'       ) as writer:
											for key in sorted(_snake_case       ):
														if key in ["log", "progress_bar", "preds"]:
																	continue
														A__	  =     metrics[key]
														if isinstance(_snake_case					,       torch.Tensor       ):
																	A__	  =     val.item()
														A__	  =     F'''{key}: {val:.6f}\n'''
														writer.write(_snake_case       )

								if not save_generations:
											return

								if "preds" in metrics:
											A__	  =     '\n'.join(metrics['preds']       )
											generations_file.open('w+'       ).write(_snake_case       )



					@rank_zero_only
					def      _a				(  self       :							Dict					,       _snake_case       :							List[str]					,       _snake_case       :							List[Any]       ):






								"""simple docstring"""


								try:
											A__	  =     pl_module.model.model.num_parameters()
								except AttributeError:
											A__	  =     pl_module.model.num_parameters()

								A__	  =     count_trainable_parameters(_snake_case       )
								# mp stands for million parameters
								trainer.logger.log_metrics({'n_params': npars, 'mp': npars / 1E6, 'grad_mp': n_trainable_pars / 1E6}       )



					@rank_zero_only
					def      _a				(  self       :							int					,       _snake_case       :							pl.Trainer					,       _snake_case       :							pl.LightningModule       ):






								"""simple docstring"""


								save_json(pl_module.metrics					,       pl_module.metrics_save_path       )
								return self._write_logs(_snake_case					,       _snake_case					,       'test'       )



					@rank_zero_only
					def      _a				(  self       :							Optional[Any]					,       _snake_case       :							pl.Trainer					,       _snake_case       :							List[Any]       ):






								"""simple docstring"""


								save_json(pl_module.metrics					,       pl_module.metrics_save_path       )
								# Uncommenting this will save val generations
								# return self._write_logs(trainer, pl_module, "valid")




