


from typing import Optional

import numpy as np
import torch
from torch import nn
from transformers import GPTaConfig, GPTaLMHeadModel
from transformers.modeling_utils import ModuleUtilsMixin

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin



class      __lowerCamelCase       (_a							,	_a							,	_a		):
       _lowercase											=			[r'''h\.\d+\.attn\.bias''', r'''h\.\d+\.attn\.masked_bias''']







       @register_to_config
       def __init__(       self:      int,A_:      int,A_:      int,A_:      Optional[int] = None,A_:      int = 5_0257,A_:      int = 1024,A_:      int = 768,A_:      int = 12,A_:      int = 12,A_:      Optional[int] = None,A_:      str = "gelu_new",A_:      float = 0.1,A_:      float = 0.1,A_:      float = 0.1,A_:      float = 1E-5,A_:      float = 0.0_2,A_:      bool = True,A_:      bool = True,A_:      bool = False,A_:      bool = False,):






            '''simple docstring'''




            super().__init__()

            __UpperCamelCase      =       prefix_length

            if prefix_inner_dim != n_embd and prefix_hidden_dim is None:
                 raise ValueError(
                     F'''`prefix_hidden_dim` cannot be `None` when `prefix_inner_dim`: {prefix_hidden_dim} and'''
                     F''' `n_embd`: {n_embd} are not equal.''' )

            __UpperCamelCase      =       prefix_inner_dim
            __UpperCamelCase      =       prefix_hidden_dim

            __UpperCamelCase      =       (
                nn.Linear(self.prefix_inner_dim,self.prefix_hidden_dim )
                if self.prefix_hidden_dim is not None
                else nn.Identity()
            )
            __UpperCamelCase      =       (
                nn.Linear(self.prefix_hidden_dim,_snake_case ) if self.prefix_hidden_dim is not None else nn.Identity()
            )

            __UpperCamelCase      =       GPTaConfig(
                vocab_size=_snake_case,n_positions=_snake_case,n_embd=_snake_case,n_layer=_snake_case,n_head=_snake_case,n_inner=_snake_case,activation_function=_snake_case,resid_pdrop=_snake_case,embd_pdrop=_snake_case,attn_pdrop=_snake_case,layer_norm_epsilon=_snake_case,initializer_range=_snake_case,scale_attn_weights=_snake_case,use_cache=_snake_case,scale_attn_by_inverse_layer_idx=_snake_case,reorder_and_upcast_attn=_snake_case,)
            __UpperCamelCase      =       GPTaLMHeadModel(_snake_case )







       def 						snake_case_     (       self:      List[Any],A_:      torch.Tensor,A_:      torch.Tensor,A_:      Optional[torch.Tensor] = None,A_:      Optional[torch.Tensor] = None,):






            '''simple docstring'''




            __UpperCamelCase      =       self.transformer.transformer.wte(_snake_case )
            __UpperCamelCase      =       self.encode_prefix(_snake_case )
            __UpperCamelCase      =       self.decode_prefix(_snake_case )
            __UpperCamelCase      =       torch.cat((prefix_embeds, embedding_text),dim=1 )

            if labels is not None:
                 __UpperCamelCase      =       self.get_dummy_token(input_ids.shape[0],input_ids.device )
                 __UpperCamelCase      =       torch.cat((dummy_token, input_ids),dim=1 )
            __UpperCamelCase      =       self.transformer(inputs_embeds=_snake_case,labels=_snake_case,attention_mask=_snake_case )
            if self.prefix_hidden_dim is not None:
                 return out, hidden
            else:
                 return out







       def 						snake_case_     (       self:      Any,A_:      int,A_:      torch.device ):






            '''simple docstring'''




            return torch.zeros(_snake_case,self.prefix_length,dtype=torch.intaa,device=_snake_case )







       def 						snake_case_     (       self:      Union[str, Any],A_:      List[str] ):






            '''simple docstring'''




            return self.encode_prefix(_snake_case )







       @torch.no_grad()
       def 						snake_case_     (       self:      Any,A_:      Dict,A_:      Optional[int],A_:      Any ):






            '''simple docstring'''




            __UpperCamelCase      =       torch.split(_snake_case,1,dim=0 )
            __UpperCamelCase      =       []
            __UpperCamelCase      =       []
            for feature in features:
                 __UpperCamelCase      =       self.decode_prefix(feature.to(_snake_case ) )  # back to the clip feature
                 # Only support beam search for now
                 __UpperCamelCase,  __UpperCamelCase      =       self.generate_beam(
                     input_embeds=_snake_case,device=_snake_case,eos_token_id=_snake_case )
                 generated_tokens.append(output_tokens[0] )
                 generated_seq_lengths.append(seq_lengths[0] )
            __UpperCamelCase      =       torch.stack(_snake_case )
            __UpperCamelCase      =       torch.stack(_snake_case )
            return generated_tokens, generated_seq_lengths







       @torch.no_grad()
       def 						snake_case_     (       self:      List[Any],A_:      Tuple=None,A_:      Optional[int]=None,A_:      List[Any]=None,A_:      int = 5,A_:      int = 67,A_:      float = 1.0,A_:      Optional[int] = None,):






            '''simple docstring'''




            __UpperCamelCase      =       eos_token_id
            __UpperCamelCase      =       None
            __UpperCamelCase      =       None
            __UpperCamelCase      =       torch.ones(_snake_case,device=_snake_case,dtype=torch.int )
            __UpperCamelCase      =       torch.zeros(_snake_case,device=_snake_case,dtype=torch.bool )

            if input_embeds is not None:
                 __UpperCamelCase      =       input_embeds
            else:
                 __UpperCamelCase      =       self.transformer.transformer.wte(_snake_case )

            for i in range(_snake_case ):
                 __UpperCamelCase      =       self.transformer(inputs_embeds=_snake_case )
                 __UpperCamelCase      =       outputs.logits
                 __UpperCamelCase      =       logits[:, -1, :] / (temperature if temperature > 0 else 1.0)
                 __UpperCamelCase      =       logits.softmax(-1 ).log()

                 if scores is None:
                      __UpperCamelCase,  __UpperCamelCase      =       logits.topk(_snake_case,-1 )
                      __UpperCamelCase      =       generated.expand(_snake_case,*generated.shape[1:] )
                      __UpperCamelCase,  __UpperCamelCase      =       next_tokens.permute(1,0 ), scores.squeeze(0 )
                      if tokens is None:
                           __UpperCamelCase      =       next_tokens
                      else:
                           __UpperCamelCase      =       tokens.expand(_snake_case,*tokens.shape[1:] )
                           __UpperCamelCase      =       torch.cat((tokens, next_tokens),dim=1 )
                 else:
                      __UpperCamelCase      =       -float(np.inf )
                      __UpperCamelCase      =       0
                      __UpperCamelCase      =       scores[:, None] + logits
                      seq_lengths[~is_stopped] += 1
                      __UpperCamelCase      =       scores_sum / seq_lengths[:, None]
                      __UpperCamelCase,  __UpperCamelCase      =       scores_sum_average.view(-1 ).topk(_snake_case,-1 )
                      __UpperCamelCase      =       next_tokens // scores_sum.shape[1]
                      __UpperCamelCase      =       seq_lengths[next_tokens_source]
                      __UpperCamelCase      =       next_tokens % scores_sum.shape[1]
                      __UpperCamelCase      =       next_tokens.unsqueeze(1 )
                      __UpperCamelCase      =       tokens[next_tokens_source]
                      __UpperCamelCase      =       torch.cat((tokens, next_tokens),dim=1 )
                      __UpperCamelCase      =       generated[next_tokens_source]
                      __UpperCamelCase      =       scores_sum_average * seq_lengths
                      __UpperCamelCase      =       is_stopped[next_tokens_source]

                 __UpperCamelCase      =       self.transformer.transformer.wte(next_tokens.squeeze() ).view(generated.shape[0],1,-1 )
                 __UpperCamelCase      =       torch.cat((generated, next_token_embed),dim=1 )
                 __UpperCamelCase      =       is_stopped + next_tokens.eq(_snake_case ).squeeze()
                 if is_stopped.all():
                      break

            __UpperCamelCase      =       scores / seq_lengths
            __UpperCamelCase      =       scores.argsort(descending=_snake_case )
            # tokens tensors are already padded to max_seq_length
            __UpperCamelCase      =       [tokens[i] for i in order]
            __UpperCamelCase      =       torch.stack(_snake_case,dim=0 )
            __UpperCamelCase      =       torch.tensor([seq_lengths[i] for i in order],dtype=seq_lengths.dtype )
            return output_texts, seq_lengths








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


snake_case_ : int    		=       logging.get_logger(__name__)

snake_case_ : str    		=       {}







class __snake_case			(			a		):
   UpperCAmelCase__  :    str          =  '''llama'''
   UpperCAmelCase__  :    Dict          =  ['''past_key_values''']







   def __init__(      self		:						str					,						_snake_case		:						List[str]=32000					,						_snake_case		:						int=4096					,						_snake_case		:						List[str]=11008					,						_snake_case		:						Optional[int]=32					,						_snake_case		:						List[Any]=32					,						_snake_case		:						Tuple=None					,						_snake_case		:						int="silu"					,						_snake_case		:						List[Any]=2048					,						_snake_case		:						List[str]=0.0_2					,						_snake_case		:						Any=1e-6					,						_snake_case		:						List[str]=True					,						_snake_case		:						Optional[Any]=0					,						_snake_case		:						Dict=1					,						_snake_case		:						List[Any]=2					,						_snake_case		:						str=1					,						_snake_case		:						Union[str, Any]=False					,						_snake_case		:						str=None					,						**_snake_case		:						List[Any]					,						):

        """simple docstring"""


        UpperCAmelCase_          =							vocab_size
        UpperCAmelCase_          =							max_position_embeddings
        UpperCAmelCase_          =							hidden_size
        UpperCAmelCase_          =							intermediate_size
        UpperCAmelCase_          =							num_hidden_layers
        UpperCAmelCase_          =							num_attention_heads

        # for backward compatibility
        if num_key_value_heads is None:
             UpperCAmelCase_          =							num_attention_heads

        UpperCAmelCase_          =							num_key_value_heads
        UpperCAmelCase_          =							hidden_act
        UpperCAmelCase_          =							initializer_range
        UpperCAmelCase_          =							rms_norm_eps
        UpperCAmelCase_          =							pretraining_tp
        UpperCAmelCase_          =							use_cache
        UpperCAmelCase_          =							rope_scaling
        self._rope_scaling_validation()

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







   def lowerCamelCase						(      self		:						Optional[int]):

        """simple docstring"""


        if self.rope_scaling is None:
             return

        if not isinstance(self.rope_scaling					,						_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}""")
        UpperCAmelCase_          =							self.rope_scaling.get('''type'''					,						_snake_case)
        UpperCAmelCase_          =							self.rope_scaling.get('''factor'''					,						_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(_snake_case					,						_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}""")






