


from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCAmelCase_				:			str   =	logging.get_logger(__name__)

UpperCAmelCase_				:			List[str]   =	{
    'EleutherAI/gpt-neox-20b': 'https://huggingface.co/EleutherAI/gpt-neox-20b/resolve/main/config.json',
    # See all GPTNeoX models at https://huggingface.co/models?filter=gpt_neox
}
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      snake_case__    :					str					       =						'''gpt_neox'''
      def __init__(	self    :	List[str]					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=5_0_4_3_2					,							SCREAMING_SNAKE_CASE__    :	Any=6_1_4_4					,							SCREAMING_SNAKE_CASE__    :	Any=4_4					,							SCREAMING_SNAKE_CASE__    :	List[str]=6_4					,							SCREAMING_SNAKE_CASE__    :	int=2_4_5_7_6					,							SCREAMING_SNAKE_CASE__    :	List[Any]="gelu"					,							SCREAMING_SNAKE_CASE__    :	str=0.25					,							SCREAMING_SNAKE_CASE__    :	int=1_0_0_0_0					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=0.0					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=0.0					,							SCREAMING_SNAKE_CASE__    :	Any=0.1					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=2_0_4_8					,							SCREAMING_SNAKE_CASE__    :	List[str]=0.02					,							SCREAMING_SNAKE_CASE__    :	str=1E-5					,							SCREAMING_SNAKE_CASE__    :	Tuple=True					,							SCREAMING_SNAKE_CASE__    :	int=0					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=2					,							SCREAMING_SNAKE_CASE__    :	int=False					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]=True					,							SCREAMING_SNAKE_CASE__    :	List[str]=None					,							**SCREAMING_SNAKE_CASE__    :	Dict					,							)		->      Any:
             super().__init__(bos_token_id=SCREAMING_SNAKE_CASE__					,							eos_token_id=SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)
             a_ :		Optional[Any]							=  vocab_size
             a_ :		Tuple							=  max_position_embeddings
             a_ :		List[Any]							=  hidden_size
             a_ :		Tuple							=  num_hidden_layers
             a_ :		Optional[Any]							=  num_attention_heads
             a_ :		Union[str, Any]							=  intermediate_size
             a_ :		List[Any]							=  hidden_act
             a_ :		Dict							=  rotary_pct
             a_ :		Union[str, Any]							=  rotary_emb_base
             a_ :		List[Any]							=  attention_dropout
             a_ :		List[str]							=  hidden_dropout
             a_ :		Tuple							=  classifier_dropout
             a_ :		Tuple							=  initializer_range
             a_ :		Tuple							=  layer_norm_eps
             a_ :		Optional[int]							=  use_cache
             a_ :		Tuple							=  tie_word_embeddings
             a_ :		List[str]							=  use_parallel_residual
             a_ :		Union[str, Any]							=  rope_scaling
             self._rope_scaling_validation()

             if self.hidden_size % self.num_attention_heads != 0:
                    raise ValueError(
                        'The hidden size is not divisble by the number of attention heads! Make sure to update them!'					)

      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					)		->      int:
             if self.rope_scaling is None:
                    return

             if not isinstance(self.rope_scaling					,							SCREAMING_SNAKE_CASE__					) or len(self.rope_scaling					) != 2:
                    raise ValueError(
                        '`rope_scaling` must be a dictionary with with two fields, `name` and `factor`, '
                        F"""got {self.rope_scaling}"""					)
             a_ :		Tuple							=  self.rope_scaling.get('type'					,							SCREAMING_SNAKE_CASE__					)
             a_ :		Union[str, Any]							=  self.rope_scaling.get('factor'					,							SCREAMING_SNAKE_CASE__					)
             if rope_scaling_type is None or rope_scaling_type not in ["linear", "dynamic"]:
                    raise ValueError(
                        F"""`rope_scaling`'s name field must be one of ['linear', 'dynamic'], got {rope_scaling_type}"""					)
             if rope_scaling_factor is None or not isinstance(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					) or rope_scaling_factor <= 1.0:
                    raise ValueError(F"""`rope_scaling`'s factor field must be an float > 1, got {rope_scaling_factor}"""					)






import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					)		->      List[Any]:
             # this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
             # `transformers` is loaded, and it's too late for inside pytest - so we are changing it
             # while running an external program

             # python one-liner segments

             # this must be loaded before socket.socket is monkey-patched
             a_ :		str							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

             a_ :		Any							=  '\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

             a_ :		Dict							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

             # Force fetching the files so that we can use the cache
             a_ :		Dict							=  'hf-internal-testing/tiny-random-bert'
             BertConfig.from_pretrained(SCREAMING_SNAKE_CASE__					)
             BertModel.from_pretrained(SCREAMING_SNAKE_CASE__					)
             BertTokenizer.from_pretrained(SCREAMING_SNAKE_CASE__					)
             pipeline(task='fill-mask'					,							model=SCREAMING_SNAKE_CASE__					)

             # baseline - just load from_pretrained with normal network
             a_ :		Union[str, Any]							=  [sys.executable, '-c', '\n'.join([load, run, mock]					)]

             # should succeed
             a_ :		List[str]							=  self.get_env()
             # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
             a_ :		Any							=  '1'
             a_ :		Optional[Any]							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      List[Any]:
             # python one-liner segments
             # this must be loaded before socket.socket is monkey-patched
             a_ :		Tuple							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

             a_ :		str							=  '\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

             a_ :		Any							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

             # Force fetching the files so that we can use the cache
             a_ :		Tuple							=  'hf-internal-testing/tiny-random-bert'
             BertConfig.from_pretrained(SCREAMING_SNAKE_CASE__					)
             BertModel.from_pretrained(SCREAMING_SNAKE_CASE__					)
             BertTokenizer.from_pretrained(SCREAMING_SNAKE_CASE__					)
             pipeline(task='fill-mask'					,							model=SCREAMING_SNAKE_CASE__					)

             # baseline - just load from_pretrained with normal network
             a_ :		Optional[Any]							=  [sys.executable, '-c', '\n'.join([load, run, mock]					)]

             # should succeed
             a_ :		Dict							=  self.get_env()
             a_ :		Dict							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Union[str, Any]:
             # this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
             # `transformers` is loaded, and it's too late for inside pytest - so we are changing it
             # while running an external program

             # python one-liner segments

             # this must be loaded before socket.socket is monkey-patched
             a_ :		Any							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

             a_ :		Dict							=  '\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

             a_ :		Any							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

             # baseline - just load from_pretrained with normal network
             a_ :		Any							=  [sys.executable, '-c', '\n'.join([load, run]					)]

             # should succeed
             a_ :		Tuple							=  self.get_env()
             a_ :		int							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)

             # next emulate no network
             a_ :		str							=  [sys.executable, '-c', '\n'.join([load, mock, run]					)]

             # Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
             # env["TRANSFORMERS_OFFLINE"] = "0"
             # result = subprocess.run(cmd, env=env, check=False, capture_output=True)
             # self.assertEqual(result.returncode, 1, result.stderr)

             # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
             a_ :		Dict							=  '1'
             a_ :		int							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[int]					)		->      Union[str, Any]:
             a_ :		Union[str, Any]							=  '\nfrom transformers import pipeline\n        '
             a_ :		Optional[Any]							=  '\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

             a_ :		Any							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
             a_ :		Union[str, Any]							=  self.get_env()
             a_ :		Optional[Any]							=  '1'
             a_ :		int							=  [sys.executable, '-c', '\n'.join([load, mock, run]					)]
             a_ :		Union[str, Any]							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							1					,							result.stderr					)
             self.assertIn(
                 'You cannot infer task automatically within `pipeline` when using offline mode'					,							result.stderr.decode().replace('\n'					,							''					)					,							)

      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Dict:
             a_ :		Optional[int]							=  '\nfrom transformers import AutoModel\n        '
             a_ :		Dict							=  '\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

             # baseline - just load from_pretrained with normal network
             a_ :		Tuple							=  [sys.executable, '-c', '\n'.join([load, run]					)]

             # should succeed
             a_ :		Optional[Any]							=  self.get_env()
             a_ :		Dict							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)

             # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
             a_ :		Optional[int]							=  '1'
             a_ :		Optional[Any]							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)


