





import argparse
import os
from os.path import isfile

import torch

from transformers import GPTaConfig

def     UpperCAmelCase_						(						_UpperCAmelCase       :Any     ,	_UpperCAmelCase       :List[str]     ,	_UpperCAmelCase       :Any=0							)  ->							List[Any]:
     '''simple docstring'''


     if name is None:
          A_					      =					None
     else:
          A_					      =					'''.''' * max(0     ,	spaces - 2							) + '''# {:''' + str(50 - spaces							) + '''s}'''
          A_					      =					fmt.format(_UpperCAmelCase							)

     # Print and recurse (if needed).
     if isinstance(_UpperCAmelCase     ,	_UpperCAmelCase							):
          if msg is not None:
               print(_UpperCAmelCase							)
          for k in val.keys():
               recursive_print(_UpperCAmelCase     ,	val[k]     ,	spaces + 2							)
     elif isinstance(_UpperCAmelCase     ,	torch.Tensor							):
          print(_UpperCAmelCase     ,	''':'''     ,	val.size()							)
     else:
          print(_UpperCAmelCase     ,	''':'''     ,	_UpperCAmelCase							)

def     UpperCAmelCase_						(						_UpperCAmelCase       :str     ,	_UpperCAmelCase       :int     ,	_UpperCAmelCase       :List[str]     ,	_UpperCAmelCase       :List[Any]							)  ->							Optional[int]:
     '''simple docstring'''


     A_					      =					param.size()
     # other versions store [num_heads * num_splits * hidden_size, :]
     A_					      =					(num_heads, num_splits, hidden_size) + input_shape[1:]
     A_					      =					param.view(*_UpperCAmelCase							)
     A_					      =					param.transpose(0     ,	1							).contiguous()
     A_					      =					param.view(*_UpperCAmelCase							)
     return param

def     UpperCAmelCase_						(						_UpperCAmelCase       :str     ,	_UpperCAmelCase       :Optional[int]							)  ->							Union[str, Any]:
     '''simple docstring'''


     A_					      =					config.n_positions
     A_					      =					config.n_layer
     A_					      =					config.vocab_size
     A_					      =					config.n_head
     A_					      =					config.n_embd // config.n_head

     A_					      =					sd_megatron['''model.language_model.embedding.word_embeddings.weight'''][:vocab_size, :]
     A_					      =					{
         '''transformer.wte.weight''': word_embeddings,
         '''transformer.wpe.weight''': sd_megatron['''model.language_model.embedding.position_embeddings.weight'''],
         '''transformer.ln_f.weight''': sd_megatron['''model.language_model.encoder.final_layernorm.weight'''],
         '''transformer.ln_f.bias''': sd_megatron['''model.language_model.encoder.final_layernorm.bias'''],
     }

     A_					      =					'''model.language_model.encoder.layers.'''
     for i in range(_UpperCAmelCase							):
          A_					      =					torch.tril(torch.ones((n_positions, n_positions)     ,	dtype=torch.bool							)							)
          A_					      =					causal_mask.view(1     ,	1     ,	_UpperCAmelCase     ,	_UpperCAmelCase							)
          A_					      =					causal_mask
          A_					      =					torch.tensor(-1e4     ,	dtype=torch.bfloataa							)

          A_					      =					sd_megatron[f'{pf}{i}.input_layernorm.weight']
          A_					      =					sd_megatron[f'{pf}{i}.input_layernorm.bias']

          A_					      =					sd_megatron[f'{pf}{i}.self_attention.query_key_value.weight']
          A_					      =					fix_query_key_value_ordering(_UpperCAmelCase     ,	3     ,	_UpperCAmelCase     ,	_UpperCAmelCase							)
          A_					      =					vala.transpose(0     ,	1							).contiguous()

          A_					      =					sd_megatron[f'{pf}{i}.self_attention.query_key_value.bias']
          A_					      =					fix_query_key_value_ordering(_UpperCAmelCase     ,	3     ,	_UpperCAmelCase     ,	_UpperCAmelCase							)
          A_					      =					vala

          A_					      =					sd_megatron[f'{pf}{i}.self_attention.dense.weight'].transpose(
              0     ,	1							)
          A_					      =					sd_megatron[f'{pf}{i}.self_attention.dense.bias']
          A_					      =					sd_megatron[f'{pf}{i}.post_attention_layernorm.weight']
          A_					      =					sd_megatron[f'{pf}{i}.post_attention_layernorm.bias']
          A_					      =					sd_megatron[f'{pf}{i}.mlp.dense_h_to_4h.weight'].transpose(0     ,	1							)
          A_					      =					sd_megatron[f'{pf}{i}.mlp.dense_h_to_4h.bias']
          A_					      =					sd_megatron[f'{pf}{i}.mlp.dense_4h_to_h.weight'].transpose(
              0     ,	1							)
          A_					      =					sd_megatron[f'{pf}{i}.mlp.dense_4h_to_h.bias']

     # For LM head, transformers' wants the matrix to weight embeddings.
     A_					      =					word_embeddings

     return sd_hf

def     UpperCAmelCase_						(						_UpperCAmelCase       :str     ,	_UpperCAmelCase       :Dict							)  ->							List[Any]:
     '''simple docstring'''


     A_					      =					64000
     A_					      =					config_megatron['''encoder_seq_length''']
     A_					      =					config_megatron['''hidden_size''']
     A_					      =					config_megatron['''num_layers''']
     A_					      =					config_megatron['''num_attention_heads''']
     A_					      =					config_megatron['''ffn_hidden_size''']
     A_					      =					'''gelu'''
     A_					      =					0.1
     A_					      =					0.1
     A_					      =					0.1
     A_					      =					config_megatron['''layernorm_epsilon''']  # 1e-5
     A_					      =					config_megatron['''init_method_std''']  # 0.02
     A_					      =					config_megatron['''apply_query_key_layer_scaling''']  # True
     A_					      =					True
     A_					      =					True

     # This identifies the 6.7B (7B) model which uses a different tokenizer
     if config_megatron["hidden_size"] == 4096:
          A_					      =					1  # <|endoftext|>
          A_					      =					1  # <|endoftext|>
          A_					      =					0  # <unk>
     else:
          A_					      =					2  # <s>
          A_					      =					3  # <|endoftext|>
          A_					      =					0  # <pad>

     return config_hf

def     UpperCAmelCase_						(						_UpperCAmelCase       :str							)  ->							List[str]:
     '''simple docstring'''


     print(_UpperCAmelCase							)

     A_					      =					args.checkpoint_path
     A_					      =					args.save_path
     if isfile(_UpperCAmelCase							):
          raise FileNotFoundError(f'ERROR! could not find file {checkpoint_path}'							)

     # Load the model.
     A_					      =					torch.load(_UpperCAmelCase     ,	map_location='''cpu'''							)

     # Load the config.
     A_					      =					checkpoint['''hyper_parameters''']['''cfg''']
     A_					      =					GPTaConfig()
     A_					      =					copy_config(config_hf=_UpperCAmelCase     ,	config_megatron=_UpperCAmelCase							)
     A_					      =					['''GPT2LMHeadModel''']

     A_					      =					checkpoint['''state_dict''']

     # Convert.
     print('''Converting'''							)
     A_					      =					convert_megatron_checkpoint(_UpperCAmelCase     ,	_UpperCAmelCase							)

     # Print the structure of converted state dict.
     if args.print_checkpoint_structure:
          recursive_print(_UpperCAmelCase     ,	_UpperCAmelCase							)

     A_					      =					'''GPTSw3Tokenizer'''

     # Store the config to file.
     print('''Saving config'''							)
     config_hf.save_pretrained(_UpperCAmelCase							)

     # Store the state_dict to file.
     A_					      =					os.path.join(_UpperCAmelCase     ,	'''pytorch_model.bin'''							)
     print(f'Saving checkpoint to "{output_checkpoint_file}"'							)
     torch.save(_UpperCAmelCase     ,	_UpperCAmelCase							)


if __name__ == "__main__":
      a__			:    Union[str, Any]			      =		argparse.ArgumentParser()
      parser.add_argument(
          '--checkpoint_path',
          type=str,
          required=True,
          help='e.g. megatron_gpt--val_loss=2.42-step=38000-consumed_samples=54720000',
      )
      parser.add_argument('--save_path', type=str, required=True, help='e.g. /home/user/gpt-sw3/hf')
      parser.add_argument('--print-checkpoint-structure', action='store_true')
      a__			:    Tuple			      =		parser.parse_args()
      main(_args)










from __future__ import annotations

def     UpperCAmelCase_						(						_UpperCAmelCase       :list[int | float]							)  ->							int | float:
     '''simple docstring'''


     if len(_UpperCAmelCase							) == 0:
          raise ValueError('''find_max() arg is an empty sequence'''							)
     A_					      =					nums[0]
     for x in nums:
          if x > max_num:
               A_					      =					x
     return max_num


if __name__ == "__main__":
      import doctest

      doctest.testmod(verbose=True)



