



"""simple docstring"""

import argparse
import time

import torch
import transformers
from measures_util import end_measure, log_measures, start_measure
from transformers import AutoConfig, AutoModelForCausalLM, AutoModelForSeqaSeqLM, AutoTokenizer

from accelerate.utils import compute_module_sizes


A_	:						Optional[int]				=			{
    'gpt-j-6b': {'is_causal': True, 'model': 'sgugger/sharded-gpt-j-6B', 'tokenizer': 'EleutherAI/gpt-j-6B'},
    'gpt-neox': {'is_causal': True, 'model': 'EleutherAI/gpt-neox-20b'},
    'opt': {'is_causal': True, 'model': 'facebook/opt-30b'},
    'T0pp': {'is_causal': False, 'model': 'bigscience/T0pp', 'model_revision': 'sharded'},
}

A_	:						List[str]				=			[
    'Hello, my name is',
    'Are unicorns real? Unicorns are',
    'For the first time in several years,',
    'My name is Julien and I am',
    'The goal of life is',
    'Whenever I\'m sad, I like to',
]
def    __snake_case       (  )    ->					Dict:






				'''simple docstring'''
				SCREAMING_SNAKE_CASE	:       int 							=    argparse.ArgumentParser(description='Run and time generations on a big model using Accelerate.'    )
				parser.add_argument('model_name'       ,		type=__A       ,		default=__A       ,		help='The name of the model to try.'    )
				parser.add_argument(
				    '--tokenizer_name'       ,		type=__A       ,		default=__A       ,		help='The name of the tokenizer (if different from the model.'    )
				parser.add_argument('--is_causal'       ,		type=__A       ,		default=__A       ,		help='Whether or not the model is causal.'    )
				parser.add_argument(
				    '--model_revision'       ,		type=__A       ,		default=__A       ,		help='The revision to use for the model checkpoint.'    )
				parser.add_argument('--torch_dtype'       ,		type=__A       ,		default=__A       ,		help='The dtype for the model.'    )
				parser.add_argument('--disk_offload'       ,		action='store_true'    )

				SCREAMING_SNAKE_CASE	:       Optional[Any] 							=    parser.parse_args()

				# Sanitize args
				if args.model_name in DEFAULT_MODELS:
								SCREAMING_SNAKE_CASE	:       Optional[Any] 							=    DEFAULT_MODELS[args.model_name]
								SCREAMING_SNAKE_CASE	:       Tuple 							=    defaults['model']
								if args.tokenizer_name is None:
												SCREAMING_SNAKE_CASE	:       Dict 							=    defaults.get('tokenizer'       ,		args.model_name    )
								if args.is_causal is None:
												SCREAMING_SNAKE_CASE	:       Union[str, Any] 							=    defaults['is_causal']
								if args.model_revision is None:
												SCREAMING_SNAKE_CASE	:       Union[str, Any] 							=    defaults.get('model_revision'       ,		'main'    )

				if args.is_causal is None:
								raise ValueError('Could not infer the default for `--is_causal`, pass either True or False for it.'    )
				if args.tokenizer_name is None:
								SCREAMING_SNAKE_CASE	:       Tuple 							=    args.model_name
				if args.model_revision is None:
								SCREAMING_SNAKE_CASE	:       Optional[int] 							=    'main'

				return args
def    __snake_case       (  )    ->					Union[str, Any]:






				'''simple docstring'''
				transformers.utils.logging.set_verbosity_error()
				SCREAMING_SNAKE_CASE	:       int 							=    parse_args()

				if args.torch_dtype is None:
								SCREAMING_SNAKE_CASE	:       List[str] 							=    AutoConfig.from_pretrained(args.model_name    )
								SCREAMING_SNAKE_CASE	:       Optional[int] 							=    getattr(__A       ,		'torch_dtype'       ,		torch.floataa    )
				else:
								SCREAMING_SNAKE_CASE	:       Union[str, Any] 							=    getattr(__A       ,		args.torch_dtype    )
				SCREAMING_SNAKE_CASE	:       int 							=    AutoModelForCausalLM if args.is_causal else AutoModelForSeqaSeqLM
				SCREAMING_SNAKE_CASE	:       List[str] 							=    {
				    'torch_dtype': torch_dtype,
				    'revision': args.model_revision,
				}
				if args.disk_offload:
								SCREAMING_SNAKE_CASE	:       Tuple 							=    'tmp_offload'
								SCREAMING_SNAKE_CASE	:       str 							=    True

				SCREAMING_SNAKE_CASE	:       Union[str, Any] 							=    start_measure()
				SCREAMING_SNAKE_CASE	:       Optional[Any] 							=    model_cls.from_pretrained(args.model_name       ,		device_map='auto'       ,		**__A    )
				SCREAMING_SNAKE_CASE	:       Union[str, Any] 							=    end_measure(__A    )
				log_measures(__A       ,		'Model loading'    )

				SCREAMING_SNAKE_CASE	:       Tuple 							=    compute_module_sizes(__A    )
				SCREAMING_SNAKE_CASE	:       List[str] 							=    {v: 0 for v in model.hf_device_map.values()}
				for module, device in model.hf_device_map.items():
								device_size[device] += module_sizes[module]
				SCREAMING_SNAKE_CASE	:       Optional[Any] 							=    '\n'.join([F"""- {device}: {size // 2**20}MiB""" for device, size in device_size.items()]    )
				print(F"""\nTheoretical use:\n{message}"""    )

				SCREAMING_SNAKE_CASE	:       Any 							=    AutoTokenizer.from_pretrained(args.tokenizer_name    )

				SCREAMING_SNAKE_CASE	:       Tuple 							=    start_measure()
				SCREAMING_SNAKE_CASE	:       List[str] 							=    []
				SCREAMING_SNAKE_CASE	:       Tuple 							=    []
				SCREAMING_SNAKE_CASE	:       Optional[int] 							=    []
				for prompt in PROMPTS:
								SCREAMING_SNAKE_CASE	:       Any 							=    tokenizer(__A       ,		return_tensors='pt'    ).to(0    )
								SCREAMING_SNAKE_CASE	:       List[Any] 							=    inputs['input_ids'][0].tolist()
								SCREAMING_SNAKE_CASE	:       Optional[int] 							=    time.time()
								SCREAMING_SNAKE_CASE	:       Optional[int] 							=    model.generate(inputs['input_ids']    )
								SCREAMING_SNAKE_CASE	:       List[Any] 							=    time.time()
								SCREAMING_SNAKE_CASE	:       str 							=    outputs[0].tolist()
								SCREAMING_SNAKE_CASE	:       List[Any] 							=    len(__A    ) if outputs[: len(__A    )] != tokens else len(__A    ) - len(__A    )
								SCREAMING_SNAKE_CASE	:       str 							=    after_generate - before_generate

								SCREAMING_SNAKE_CASE	:       Tuple 							=    tokenizer.decode(__A       ,		skip_special_tokens=__A    )
								texts_outs.append(__A    )
								generation_times.append(__A    )
								gen_tokens.append(__A    )
								print(F"""Prompt: {prompt}\nGeneration {text_out}\nIn {generation_time:.2f}s for {num_gen_tokens} tokens\n"""    )

				SCREAMING_SNAKE_CASE	:       List[Any] 							=    end_measure(__A    )
				log_measures(__A       ,		'Model generation'    )

				SCREAMING_SNAKE_CASE	:       int 							=    [gen / tok for gen, tok in zip(__A       ,		__A    )]
				SCREAMING_SNAKE_CASE	:       List[Any] 							=    sum(__A    ) / len(__A    )
				print(F"""Average time of generation per token: {avg_gen:.2f}s"""    )
				print(F"""First generation (avg time per token): {generation_times_per_token[0]:.2f}s"""    )
				SCREAMING_SNAKE_CASE	:       str 							=    sum(generation_times_per_token[1:]    ) / (len(__A    ) - 1)
				print(F"""Average time of generation per token (excluding the first): {avg_gen:.2f}s"""    )


if __name__ == "__main__":
							main()










"""simple docstring"""

def    __snake_case       (  __A			:	int    )    ->					int:






				'''simple docstring'''
				if not isinstance(__A       ,		__A    ):
								raise ValueError('double_factorial() only accepts integral values'    )
				if num < 0:
								raise ValueError('double_factorial() not defined for negative values'    )
				SCREAMING_SNAKE_CASE	:       List[str] 							=    1
				for i in range(__A       ,		0       ,		-2    ):
								value *= i
				return value


if __name__ == "__main__":
							import doctest

							doctest.testmod()





