
import unittest

from parameterized import parameterized

from transformers import AutoTokenizer, GPTNeoXConfig, is_torch_available, set_seed
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
							import torch

							from transformers import (
							    GPTNeoXForCausalLM,
							    GPTNeoXForQuestionAnswering,
							    GPTNeoXForSequenceClassification,
							    GPTNeoXForTokenClassification,
							    GPTNeoXModel,
							)

class     _UpperCamelCase			:






			'''simple docstring'''

			def __init__(    self	:    Union[str, Any]  ,		snake_case_	:    int  ,		snake_case_	:    Any=13  ,		snake_case_	:    List[Any]=7  ,		snake_case_	:    Union[str, Any]=True  ,		snake_case_	:    Dict=True  ,		snake_case_	:    Optional[Any]=True  ,		snake_case_	:    str=True  ,		snake_case_	:    Any=99  ,		snake_case_	:    Optional[Any]=64  ,		snake_case_	:    str=5  ,		snake_case_	:    Optional[Any]=4  ,		snake_case_	:    Any=37  ,		snake_case_	:    Dict="gelu"  ,		snake_case_	:    List[str]=0.1  ,		snake_case_	:    Tuple=0.1  ,		snake_case_	:    List[Any]=512  ,		snake_case_	:    str=16  ,		snake_case_	:    Union[str, Any]=2  ,		snake_case_	:    Any=0.02  ,		snake_case_	:    str=3  ,		snake_case_	:    List[Any]=4  ,		snake_case_	:    List[Any]=None  ,		):
							UpperCamelCase_:							List[str]		=  parent
							UpperCamelCase_:							List[Any]		=  batch_size
							UpperCamelCase_:							List[str]		=  seq_length
							UpperCamelCase_:							Union[str, Any]		=  is_training
							UpperCamelCase_:							int		=  use_input_mask
							UpperCamelCase_:							Tuple		=  use_token_type_ids
							UpperCamelCase_:							List[Any]		=  use_labels
							UpperCamelCase_:							List[Any]		=  vocab_size
							UpperCamelCase_:							List[Any]		=  hidden_size
							UpperCamelCase_:							List[str]		=  num_hidden_layers
							UpperCamelCase_:							Optional[Any]		=  num_attention_heads
							UpperCamelCase_:							Optional[Any]		=  intermediate_size
							UpperCamelCase_:							int		=  hidden_act
							UpperCamelCase_:							Dict		=  hidden_dropout_prob
							UpperCamelCase_:							Optional[Any]		=  attention_probs_dropout_prob
							UpperCamelCase_:							Tuple		=  max_position_embeddings
							UpperCamelCase_:							Optional[int]		=  type_vocab_size
							UpperCamelCase_:							Dict		=  type_sequence_label_size
							UpperCamelCase_:							Tuple		=  initializer_range
							UpperCamelCase_:							Any		=  num_labels
							UpperCamelCase_:							List[Any]		=  num_choices
							UpperCamelCase_:							Optional[int]		=  scope
							UpperCamelCase_:							Tuple		=  vocab_size - 1

			def  lowerCAmelCase__		(    self	:    List[str] ):
							UpperCamelCase_:							Union[str, Any]		=  ids_tensor([self.batch_size, self.seq_length]  ,		self.vocab_size )

							UpperCamelCase_:							int		=  None
							if self.use_input_mask:
											UpperCamelCase_:							int		=  random_attention_mask([self.batch_size, self.seq_length] )

							UpperCamelCase_:							List[Any]		=  None
							if self.use_labels:
											UpperCamelCase_:							List[str]		=  ids_tensor([self.batch_size, self.seq_length]  ,		self.num_labels )

							UpperCamelCase_:							Dict		=  self.get_config()

							return config, input_ids, input_mask, token_labels

			def  lowerCAmelCase__		(    self	:    Tuple ):
							return GPTNeoXConfig(
							    vocab_size=self.vocab_size  ,		hidden_size=self.hidden_size  ,		num_hidden_layers=self.num_hidden_layers  ,		num_attention_heads=self.num_attention_heads  ,		intermediate_size=self.intermediate_size  ,		hidden_act=self.hidden_act  ,		hidden_dropout_prob=self.hidden_dropout_prob  ,		attention_probs_dropout_prob=self.attention_probs_dropout_prob  ,		max_position_embeddings=self.max_position_embeddings  ,		type_vocab_size=self.type_vocab_size  ,		is_decoder=snake_case_  ,		initializer_range=self.initializer_range  ,		pad_token_id=self.pad_token_id  ,		)

			def  lowerCAmelCase__		(    self	:    Union[str, Any] ):
							UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_:							Dict		=  self.prepare_config_and_inputs()

							UpperCamelCase_:							List[Any]		=  True

							return config, input_ids, input_mask, token_labels

			def  lowerCAmelCase__		(    self	:    Dict  ,		snake_case_	:    Any  ,		snake_case_	:    Optional[int]  ,		snake_case_	:    Tuple ):
							UpperCamelCase_:							Tuple		=  GPTNeoXModel(config=snake_case_ )
							model.to(snake_case_ )
							model.eval()
							UpperCamelCase_:							Any		=  model(snake_case_  ,		attention_mask=snake_case_ )
							UpperCamelCase_:							Optional[int]		=  model(snake_case_ )
							self.parent.assertEqual(result.last_hidden_state.shape  ,		(self.batch_size, self.seq_length, self.hidden_size) )

			def  lowerCAmelCase__		(    self	:    Any  ,		snake_case_	:    Dict  ,		snake_case_	:    Optional[Any]  ,		snake_case_	:    Dict ):
							UpperCamelCase_:							List[Any]		=  True
							UpperCamelCase_:							Optional[Any]		=  GPTNeoXModel(snake_case_ )
							model.to(snake_case_ )
							model.eval()
							UpperCamelCase_:							Dict		=  model(snake_case_  ,		attention_mask=snake_case_ )
							self.parent.assertEqual(result.last_hidden_state.shape  ,		(self.batch_size, self.seq_length, self.hidden_size) )

			def  lowerCAmelCase__		(    self	:    Any  ,		snake_case_	:    Optional[int]  ,		snake_case_	:    Dict  ,		snake_case_	:    Optional[Any]  ,		snake_case_	:    Dict ):
							UpperCamelCase_:							Optional[int]		=  GPTNeoXForCausalLM(config=snake_case_ )
							model.to(snake_case_ )
							model.eval()
							UpperCamelCase_:							Optional[int]		=  model(snake_case_  ,		attention_mask=snake_case_  ,		labels=snake_case_ )
							self.parent.assertEqual(result.logits.shape  ,		(self.batch_size, self.seq_length, self.vocab_size) )

			def  lowerCAmelCase__		(    self	:    Union[str, Any]  ,		snake_case_	:    int  ,		snake_case_	:    Optional[Any]  ,		snake_case_	:    List[Any]  ,		snake_case_	:    Optional[Any] ):
							UpperCamelCase_:							Optional[Any]		=  self.num_labels
							UpperCamelCase_:							Union[str, Any]		=  GPTNeoXForQuestionAnswering(snake_case_ )
							model.to(snake_case_ )
							model.eval()
							UpperCamelCase_:							Tuple		=  model(snake_case_  ,		attention_mask=snake_case_ )
							self.parent.assertEqual(result.start_logits.shape  ,		(self.batch_size, self.seq_length) )
							self.parent.assertEqual(result.end_logits.shape  ,		(self.batch_size, self.seq_length) )

			def  lowerCAmelCase__		(    self	:    str  ,		snake_case_	:    Optional[Any]  ,		snake_case_	:    Tuple  ,		snake_case_	:    int  ,		snake_case_	:    Optional[int] ):
							UpperCamelCase_:							List[Any]		=  self.num_labels
							UpperCamelCase_:							List[str]		=  GPTNeoXForSequenceClassification(snake_case_ )
							model.to(snake_case_ )
							model.eval()
							UpperCamelCase_:							List[Any]		=  ids_tensor([self.batch_size]  ,		self.type_sequence_label_size )
							UpperCamelCase_:							Union[str, Any]		=  model(snake_case_  ,		attention_mask=snake_case_  ,		labels=snake_case_ )
							self.parent.assertEqual(result.logits.shape  ,		(self.batch_size, self.num_labels) )

			def  lowerCAmelCase__		(    self	:    Union[str, Any]  ,		snake_case_	:    Tuple  ,		snake_case_	:    Tuple  ,		snake_case_	:    List[str]  ,		snake_case_	:    int ):
							UpperCamelCase_:							List[Any]		=  self.num_labels
							UpperCamelCase_:							Tuple		=  GPTNeoXForTokenClassification(snake_case_ )
							model.to(snake_case_ )
							model.eval()
							UpperCamelCase_:							Dict		=  model(snake_case_  ,		attention_mask=snake_case_  ,		labels=snake_case_ )
							self.parent.assertEqual(result.logits.shape  ,		(self.batch_size, self.seq_length, self.num_labels) )

			def  lowerCAmelCase__		(    self	:    List[str]  ,		snake_case_	:    Optional[int]  ,		snake_case_	:    List[Any]  ,		snake_case_	:    List[Any] ):
							UpperCamelCase_:							int		=  True
							UpperCamelCase_:							Optional[Any]		=  GPTNeoXForCausalLM(config=snake_case_ )
							model.to(snake_case_ )
							model.eval()

							# first forward pass
							UpperCamelCase_:							List[str]		=  model(snake_case_  ,		attention_mask=snake_case_  ,		use_cache=snake_case_ )
							UpperCamelCase_:							Tuple		=  outputs.past_key_values

							# create hypothetical multiple next token and extent to next_input_ids
							UpperCamelCase_:							Optional[Any]		=  ids_tensor((self.batch_size, 3)  ,		config.vocab_size )
							UpperCamelCase_:							Any		=  ids_tensor((self.batch_size, 3)  ,		vocab_size=2 )

							# append to next input_ids and
							UpperCamelCase_:							Tuple		=  torch.cat([input_ids, next_tokens]  ,		dim=-1 )
							UpperCamelCase_:							Union[str, Any]		=  torch.cat([input_mask, next_mask]  ,		dim=-1 )

							UpperCamelCase_:							int		=  model(snake_case_  ,		attention_mask=snake_case_  ,		output_hidden_states=snake_case_ )
							UpperCamelCase_:							Any		=  output_from_no_past["""hidden_states"""][0]
							UpperCamelCase_:							List[Any]		=  model(
							    snake_case_  ,		attention_mask=snake_case_  ,		past_key_values=snake_case_  ,		output_hidden_states=snake_case_  ,		)["""hidden_states"""][0]

							# select random slice
							UpperCamelCase_:							List[Any]		=  ids_tensor((1,)  ,		output_from_past.shape[-1] ).item()
							UpperCamelCase_:							Optional[int]		=  output_from_no_past[:, -3:, random_slice_idx].detach()
							UpperCamelCase_:							Tuple		=  output_from_past[:, :, random_slice_idx].detach()

							self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1] )

							# test that outputs are equal for slice
							self.parent.assertTrue(torch.allclose(snake_case_  ,		snake_case_  ,		atol=1e-3 ) )




			def  lowerCAmelCase__		(    self	:    int ):
							UpperCamelCase_:							List[Any]		=  self.prepare_config_and_inputs()
							UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_:							Optional[Any]		=  config_and_inputs
							UpperCamelCase_:							Any		=  {"""input_ids""": input_ids, """attention_mask""": input_mask}
							return config, inputs_dict




@require_torch
class     _UpperCamelCase			(   _A       ,    _A       ,    _A       ,    unittest.TestCase					):






			'''simple docstring'''

			__UpperCamelCase :  int    =	(
			    (
			        GPTNeoXModel,
			        GPTNeoXForCausalLM,
			        GPTNeoXForQuestionAnswering,
			        GPTNeoXForSequenceClassification,
			        GPTNeoXForTokenClassification,
			    )
			    if is_torch_available()
			    else ()
			)
			__UpperCamelCase :  Dict    =	(GPTNeoXForCausalLM,) if is_torch_available() else ()
			__UpperCamelCase :  int    =	(
			    {
			        """feature-extraction""": GPTNeoXModel,
			        """question-answering""": GPTNeoXForQuestionAnswering,
			        """text-classification""": GPTNeoXForSequenceClassification,
			        """text-generation""": GPTNeoXForCausalLM,
			        """token-classification""": GPTNeoXForTokenClassification,
			        """zero-shot""": GPTNeoXForSequenceClassification,
			    }
			    if is_torch_available()
			    else {}
			)
			__UpperCamelCase :  Optional[Any]    =	False
			__UpperCamelCase :  Union[str, Any]    =	False
			__UpperCamelCase :  Optional[int]    =	False
			__UpperCamelCase :  Tuple    =	False

			def  lowerCAmelCase__		(    self	:    Optional[int] ):
							UpperCamelCase_:							Optional[int]		=  GPTNeoXModelTester(self )
							UpperCamelCase_:							Optional[Any]		=  ConfigTester(self  ,		config_class=snake_case_  ,		hidden_size=64  ,		num_attention_heads=8 )

			def  lowerCAmelCase__		(    self	:    int ):
							self.config_tester.run_common_tests()

			def  lowerCAmelCase__		(    self	:    str ):
							UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_:							Tuple		=  self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(snake_case_  ,		snake_case_  ,		snake_case_ )

			def  lowerCAmelCase__		(    self	:    Optional[Any] ):
							UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_:							int		=  self.model_tester.prepare_config_and_inputs_for_decoder()
							self.model_tester.create_and_check_model_as_decoder(snake_case_  ,		snake_case_  ,		snake_case_ )

			def  lowerCAmelCase__		(    self	:    List[str] ):
							# This regression test was failing with PyTorch < 1.3
							UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_:							Optional[int]		=  self.model_tester.prepare_config_and_inputs_for_decoder()

							UpperCamelCase_:							Dict		=  None

							self.model_tester.create_and_check_model_as_decoder(snake_case_  ,		snake_case_  ,		snake_case_ )

			def  lowerCAmelCase__		(    self	:    List[str] ):
							UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_,				UpperCamelCase_:							Any		=  self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_decoder_model_past_large_inputs(snake_case_  ,		snake_case_  ,		snake_case_ )

			def  lowerCAmelCase__		(    self	:    List[str] ):
							UpperCamelCase_:							Optional[Any]		=  self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_causal_lm(*snake_case_ )

			def  lowerCAmelCase__		(    self	:    Union[str, Any] ):
							UpperCamelCase_:							Optional[int]		=  self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_question_answering(*snake_case_ )

			def  lowerCAmelCase__		(    self	:    Dict ):
							UpperCamelCase_:							Optional[int]		=  self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_sequence_classification(*snake_case_ )

			def  lowerCAmelCase__		(    self	:    Optional[int] ):
							UpperCamelCase_:							Any		=  self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_token_classification(*snake_case_ )

			@unittest.skip(reason="""Feed forward chunking is not implemented""" )
			def  lowerCAmelCase__		(    self	:    Optional[Any] ):
							pass




			@parameterized.expand([("""linear""",), ("""dynamic""",)] )
			def  lowerCAmelCase__		(    self	:    List[str]  ,		snake_case_	:    Any ):
							UpperCamelCase_,				UpperCamelCase_:							Any		=  self.model_tester.prepare_config_and_inputs_for_common()
							UpperCamelCase_:							Optional[int]		=  ids_tensor([1, 10]  ,		config.vocab_size )
							UpperCamelCase_:							Dict		=  ids_tensor([1, int(config.max_position_embeddings * 1.5 )]  ,		config.vocab_size )

							set_seed(42 )  # Fixed seed at init time so the two models get the same random weights
							UpperCamelCase_:							List[Any]		=  GPTNeoXModel(snake_case_ )
							original_model.to(snake_case_ )
							original_model.eval()
							UpperCamelCase_:							List[Any]		=  original_model(snake_case_ ).last_hidden_state
							UpperCamelCase_:							List[Any]		=  original_model(snake_case_ ).last_hidden_state

							set_seed(42 )  # Fixed seed at init time so the two models get the same random weights
							UpperCamelCase_:							str		=  {"""type""": scaling_type, """factor""": 10.0}
							UpperCamelCase_:							Optional[int]		=  GPTNeoXModel(snake_case_ )
							scaled_model.to(snake_case_ )
							scaled_model.eval()
							UpperCamelCase_:							List[Any]		=  scaled_model(snake_case_ ).last_hidden_state
							UpperCamelCase_:							Dict		=  scaled_model(snake_case_ ).last_hidden_state

							# Dynamic scaling does not change the RoPE embeddings until it receives an input longer than the original
							# maximum sequence length, so the outputs for the short input should match.
							if scaling_type == "dynamic":
											self.assertTrue(torch.allclose(snake_case_  ,		snake_case_  ,		atol=1e-5 ) )
							else:
											self.assertFalse(torch.allclose(snake_case_  ,		snake_case_  ,		atol=1e-5 ) )

							# The output should be different for long inputs
							self.assertFalse(torch.allclose(snake_case_  ,		snake_case_  ,		atol=1e-5 ) )




@require_torch
class     _UpperCamelCase			(   unittest.TestCase					):






			'''simple docstring'''

			@slow
			def  lowerCAmelCase__		(    self	:    Optional[Any] ):
							UpperCamelCase_:							Optional[int]		=  AutoTokenizer.from_pretrained("""EleutherAI/pythia-410m-deduped""" )
							for checkpointing in [True, False]:
											UpperCamelCase_:							Dict		=  GPTNeoXForCausalLM.from_pretrained("""EleutherAI/pythia-410m-deduped""" )

											if checkpointing:
															model.gradient_checkpointing_enable()
											else:
															model.gradient_checkpointing_disable()
											model.to(snake_case_ )

											UpperCamelCase_:							Optional[int]		=  tokenizer("""My favorite food is"""  ,		return_tensors="""pt""" ).to(snake_case_ )
											# The hub repo. is updated on 2023-04-04, resulting in poor outputs.
											# See: https://github.com/huggingface/transformers/pull/24193
											UpperCamelCase_:							str		=  """My favorite food is a good old-fashioned, old-fashioned, old-fashioned.\n\nI'm not sure"""

											UpperCamelCase_:							Union[str, Any]		=  model.generate(**snake_case_  ,		do_sample=snake_case_  ,		max_new_tokens=20 )
											UpperCamelCase_:							Any		=  tokenizer.batch_decode(snake_case_ )[0]

											self.assertEqual(snake_case_  ,		snake_case_ )








import json
import multiprocessing
import os
import re
from collections import defaultdict

import torch
from accelerate import Accelerator
from accelerate.utils import set_seed
from arguments import HumanEvalArguments
from datasets import load_dataset, load_metric
from torch.utils.data import IterableDataset
from torch.utils.data.dataloader import DataLoader
from tqdm import tqdm

import transformers
from transformers import AutoModelForCausalLM, AutoTokenizer, HfArgumentParser, StoppingCriteria, StoppingCriteriaList


lowerCamelCase_  :	str 				=					["""\nclass""", """\ndef""", """\n#""", """\n@""", """\nprint""", """\nif"""]

class     _UpperCamelCase			(   _A					):






			'''simple docstring'''

			def __init__(    self	:    Optional[int]  ,		snake_case_	:    List[str]  ,		snake_case_	:    Union[str, Any]  ,		snake_case_	:    Optional[Any]=None  ,		snake_case_	:    str=1 ):
							UpperCamelCase_:							List[str]		=  tokenizer
							UpperCamelCase_:							str		=  dataset
							UpperCamelCase_:							List[str]		=  len(snake_case_ ) if n_tasks is None else n_tasks
							UpperCamelCase_:							str		=  n_copies




			def __iter__(    self	:    Tuple ):
							UpperCamelCase_:							Optional[int]		=  []
							for task in range(self.n_tasks ):
											# without strip, the model generate commented codes ...
											prompts.append(self.tokenizer.eos_token + self.dataset[task]["""prompt"""].strip() )
							UpperCamelCase_:							List[str]		=  self.tokenizer(snake_case_  ,		padding=snake_case_  ,		return_tensors="""pt""" )
							for task in range(self.n_tasks ):
											for _ in range(self.n_copies ):
															yield {
															    "ids": outputs.input_ids[task],
															    "task_id": task,
															    "input_len": outputs.attention_mask[task].sum(),
															}




class     _UpperCamelCase			(   _A					):






			'''simple docstring'''

			def __init__(    self	:    Any  ,		snake_case_	:    List[str]  ,		snake_case_	:    Dict  ,		snake_case_	:    int ):
							UpperCamelCase_:							Union[str, Any]		=  start_length
							UpperCamelCase_:							Dict		=  eof_strings
							UpperCamelCase_:							List[str]		=  tokenizer




			def __call__(    self	:    Tuple  ,		snake_case_	:    List[str]  ,		snake_case_	:    Optional[Any]  ,		**snake_case_	:    int ):
							UpperCamelCase_:							Union[str, Any]		=  self.tokenizer.batch_decode(input_ids[:, self.start_length :] )
							UpperCamelCase_:							Dict		=  []
							for decoded_generation in decoded_generations:
											done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings ) )
							return all(snake_case_ )







def     A__   (  lowerCamelCase  )					->		Optional[int]:
				UpperCamelCase_:							str		=  re.split("""(%s)""" % """|""".join(lowerCamelCase  )       ,      lowerCamelCase  )
				# last string should be ""
				return "".join(string_list[:-2]  )







def     A__   (  lowerCamelCase       ,      lowerCamelCase       ,      lowerCamelCase       ,      lowerCamelCase       ,      lowerCamelCase       ,      lowerCamelCase=20       ,      **lowerCamelCase  )					->		int:
				UpperCamelCase_:							str		=  defaultdict(lowerCamelCase  )  # dict of list of generated tokens
				for step, batch in tqdm(enumerate(lowerCamelCase  )  ):
								with torch.no_grad():
												UpperCamelCase_:							Optional[int]		=  batch["""ids"""].shape[-1]
												UpperCamelCase_:							Dict		=  accelerator.unwrap_model(lowerCamelCase  ).generate(
												    input_ids=batch["""ids"""][:, : batch["""input_len"""]]       ,      num_return_sequences=lowerCamelCase       ,      **lowerCamelCase  )
												# each task is generated batch_size times
												UpperCamelCase_:							Optional[int]		=  batch["""task_id"""].repeat(lowerCamelCase  )
												UpperCamelCase_:							int		=  accelerator.pad_across_processes(
												    lowerCamelCase       ,      dim=1       ,      pad_index=tokenizer.pad_token_id  )

												UpperCamelCase_,				UpperCamelCase_:							Tuple		=  accelerator.gather((generated_tokens, generated_tasks)  )
												UpperCamelCase_:							Tuple		=  generated_tokens.cpu().numpy()
												UpperCamelCase_:							Any		=  generated_tasks.cpu().numpy()

												for task, generated_tokens in zip(lowerCamelCase       ,      lowerCamelCase  ):
																gen_token_dict[task].append(lowerCamelCase  )

				UpperCamelCase_:							Dict		=  [[] for _ in range(lowerCamelCase  )]
				for task, generated_tokens in gen_token_dict.items():
								for s in generated_tokens:
												UpperCamelCase_:							Any		=  tokenizer.decode(lowerCamelCase       ,      skip_special_tokens=lowerCamelCase       ,      clean_up_tokenization_spaces=lowerCamelCase  )
												code_gens[task].append(remove_last_block(lowerCamelCase  )  )
				return code_gens







def     A__   (  )					->		Union[str, Any]:
				# Setup configuration
				UpperCamelCase_:							Optional[Any]		=  HfArgumentParser(lowerCamelCase  )
				UpperCamelCase_:							str		=  parser.parse_args()

				transformers.logging.set_verbosity_error()
				# enables code execution in code_eval metric
				UpperCamelCase_:							List[str]		=  args.HF_ALLOW_CODE_EVAL
				# make sure tokenizer plays nice with multiprocessing
				UpperCamelCase_:							Union[str, Any]		=  """false"""

				if args.num_workers is None:
								UpperCamelCase_:							Dict		=  multiprocessing.cpu_count()

				# Use dataset load to feed to accelerate
				UpperCamelCase_:							List[Any]		=  Accelerator()
				set_seed(args.seed       ,      device_specific=lowerCamelCase  )

				# Load model and tokenizer
				UpperCamelCase_:							Any		=  AutoTokenizer.from_pretrained(args.model_ckpt  )
				UpperCamelCase_:							Union[str, Any]		=  tokenizer.eos_token
				UpperCamelCase_:							Optional[Any]		=  AutoModelForCausalLM.from_pretrained(args.model_ckpt  )

				# Generation settings
				UpperCamelCase_:							Union[str, Any]		=  {
				    """do_sample""": args.do_sample,
				    """temperature""": args.temperature,
				    """max_new_tokens""": args.max_new_tokens,
				    """top_p""": args.top_p,
				    """top_k""": args.top_k,
				    """stopping_criteria""": StoppingCriteriaList([EndOfFunctionCriteria(0       ,      lowerCamelCase       ,      lowerCamelCase  )]  ),
				}

				# Load evaluation dataset and metric
				UpperCamelCase_:							Any		=  load_dataset("""openai_humaneval"""  )
				UpperCamelCase_:							Union[str, Any]		=  load_metric("""code_eval"""  )

				UpperCamelCase_:							Union[str, Any]		=  args.num_tasks if args.num_tasks is not None else len(human_eval["""test"""]  )
				UpperCamelCase_:							List[Any]		=  args.n_samples // args.batch_size

				UpperCamelCase_:							Any		=  TokenizedDataset(lowerCamelCase       ,      human_eval["""test"""]       ,      n_copies=lowerCamelCase       ,      n_tasks=lowerCamelCase  )
				# do not confuse args.batch_size, which is actually the num_return_sequences
				UpperCamelCase_:							Optional[int]		=  DataLoader(lowerCamelCase       ,      batch_size=1  )

				# Run a quick test to see if code evaluation is enabled
				try:
								UpperCamelCase_:							List[str]		=  code_eval_metric.compute(references=[""""""]       ,      predictions=[[""""""]]  )
				except ValueError as exception:
								print(
								    """Code evaluation not enabled. Read the warning below carefully and then use `--HF_ALLOW_CODE_EVAL=\"1\"`"""
								    """ flag to enable code evaluation."""  )
								raise exception

				UpperCamelCase_,				UpperCamelCase_:							Dict		=  accelerator.prepare(lowerCamelCase       ,      lowerCamelCase  )

				UpperCamelCase_:							Union[str, Any]		=  complete_code(
				    lowerCamelCase       ,      lowerCamelCase       ,      lowerCamelCase       ,      lowerCamelCase       ,      n_tasks=lowerCamelCase       ,      batch_size=args.batch_size       ,      **lowerCamelCase       ,      )

				if accelerator.is_main_process:
								UpperCamelCase_:							List[Any]		=  []

								for task in tqdm(range(lowerCamelCase  )  ):
												UpperCamelCase_:							Optional[Any]		=  human_eval["""test"""][task]["""test"""]
												UpperCamelCase_:							Optional[int]		=  F'''check({human_eval["test"][task]["entry_point"]})'''
												references.append("""\n""" + test_func + """\n""" + entry_point  )

								# Evaluate completions with "code_eval" metric
								UpperCamelCase_,				UpperCamelCase_:							str		=  code_eval_metric.compute(
								    references=lowerCamelCase       ,      predictions=lowerCamelCase       ,      num_workers=args.num_workers  )
								print(F'''Results: {pass_at_k}'''  )

								# Save results to json file
								with open(args.output_file       ,      """w"""  ) as fp:
												json.dump(lowerCamelCase       ,      lowerCamelCase  )


# For some reason the folliwng seems to be necessary sometimes for code_eval to work nice with multiprocessing
# https://stackoverflow.com/questions/60804599/python-multiprocessing-keeps-spawning-the-whole-script
if __name__ == "__main__":
							main()






