






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


SCREAMING_SNAKE_CASE__				     =			logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__				     =			{
    '''tiiuae/falcon-40b''': '''https://huggingface.co/tiiuae/falcon-40b/resolve/main/config.json''',
    '''tiiuae/falcon-7b''': '''https://huggingface.co/tiiuae/falcon-7b/resolve/main/config.json''',
}






class 			_UpperCamelCase(       __lowerCamelCase		):
		__SCREAMING_SNAKE_CASE  :       List[str]							=							'''falcon'''
		__SCREAMING_SNAKE_CASE  :       Optional[Any]							=							['''past_key_values''']

		def __init__(					self		:    List[Any]    ,       SCREAMING_SNAKE_CASE__		:    Any=6_5_0_2_4    ,       SCREAMING_SNAKE_CASE__		:    Any=4_5_4_4    ,       SCREAMING_SNAKE_CASE__		:    Any=3_2    ,       SCREAMING_SNAKE_CASE__		:    List[str]=7_1    ,       SCREAMING_SNAKE_CASE__		:    int=1e-5    ,       SCREAMING_SNAKE_CASE__		:    Union[str, Any]=0.02    ,       SCREAMING_SNAKE_CASE__		:    str=True    ,       SCREAMING_SNAKE_CASE__		:    List[Any]=0.0    ,       SCREAMING_SNAKE_CASE__		:    str=0.0    ,       SCREAMING_SNAKE_CASE__		:    Optional[Any]=None    ,       SCREAMING_SNAKE_CASE__		:    Optional[Any]=False    ,       SCREAMING_SNAKE_CASE__		:    List[str]=False    ,       SCREAMING_SNAKE_CASE__		:    Optional[int]=True    ,       SCREAMING_SNAKE_CASE__		:    List[str]=True    ,       SCREAMING_SNAKE_CASE__		:    List[Any]=False    ,       SCREAMING_SNAKE_CASE__		:    str=1_1    ,       SCREAMING_SNAKE_CASE__		:    Union[str, Any]=1_1    ,       **SCREAMING_SNAKE_CASE__		:    Union[str, Any]    ,       ):



					'''simple docstring'''



					__a						:  Any									=     vocab_size
					# Backward compatibility with n_embed kwarg
					__a						:  int									=     kwargs.pop('n_embed'    ,       SCREAMING_SNAKE_CASE__    )
					__a						:  str									=     hidden_size if n_embed is None else n_embed
					__a						:  Any									=     num_hidden_layers
					__a						:  Dict									=     num_attention_heads
					__a						:  Union[str, Any]									=     layer_norm_epsilon
					__a						:  List[str]									=     initializer_range
					__a						:  Optional[Any]									=     use_cache
					__a						:  List[str]									=     hidden_dropout
					__a						:  Dict									=     attention_dropout

					__a						:  Any									=     bos_token_id
					__a						:  Optional[int]									=     eos_token_id
					__a						:  List[Any]									=     num_attention_heads if num_kv_heads is None else num_kv_heads
					__a						:  int									=     alibi
					__a						:  Tuple									=     new_decoder_architecture
					__a						:  str									=     multi_query  # Ignored when new_decoder_architecture is True
					__a						:  Any									=     parallel_attn
					__a						:  Dict									=     bias

					super().__init__(bos_token_id=SCREAMING_SNAKE_CASE__    ,       eos_token_id=SCREAMING_SNAKE_CASE__    ,       **SCREAMING_SNAKE_CASE__    )

		@property
		def    __lowerCAmelCase  (					self		:    Optional[Any]    ):



					'''simple docstring'''



					return self.hidden_size // self.num_attention_heads



		@property
		def    __lowerCAmelCase  (					self		:    Tuple    ):



					'''simple docstring'''



					return not self.alibi













from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
  from ..models.auto import AutoTokenizer
class 		SCREAMING_SNAKE_CASE__	:

  '''simple docstring'''







  def  lowerCamelCase       (      self		:			str       ,				lowerCamelCase		:			str							)						->	Optional[int]:


         """simple docstring"""
         raise NotImplementedError()





  def  lowerCamelCase       (      self		:			Any							)						->	Any:


         """simple docstring"""
         raise NotImplementedError()



class 		SCREAMING_SNAKE_CASE__	(  UpperCAmelCase     ):

  '''simple docstring'''







  def __init__(      self		:			str       ,				lowerCamelCase		:			"AutoTokenizer"       ,				lowerCamelCase		:			bool = False       ,				**lowerCamelCase		:			Any							)						->	Dict:


         """simple docstring"""
         _UpperCAmelCase							=    tokenizer
         _UpperCAmelCase							=    skip_prompt
         _UpperCAmelCase							=    decode_kwargs

         # variables used in the streaming process
         _UpperCAmelCase							=    []
         _UpperCAmelCase							=    0
         _UpperCAmelCase							=    True


  def  lowerCamelCase       (      self		:			Any       ,				lowerCamelCase		:			List[str]							)						->	List[Any]:


         """simple docstring"""
         if len(value.shape							) > 1 and value.shape[0] > 1:
                raise ValueError("""TextStreamer only supports batch size 1"""							)
         elif len(value.shape							) > 1:
                _UpperCAmelCase							=    value[0]

         if self.skip_prompt and self.next_tokens_are_prompt:
                _UpperCAmelCase							=    False
                return

         # Add the new token to the cache and decodes the entire thing.
         self.token_cache.extend(value.tolist()							)
         _UpperCAmelCase							=    self.tokenizer.decode(self.token_cache       ,				**self.decode_kwargs							)

         # After the symbol for a new line, we flush the cache.
         if text.endswith("""\n"""							):
                _UpperCAmelCase							=    text[self.print_len :]
                _UpperCAmelCase							=    []
                _UpperCAmelCase							=    0
         # If the last token is a CJK character, we print the characters.
         elif len(lowerCamelCase							) > 0 and self._is_chinese_char(ord(text[-1]							)							):
                _UpperCAmelCase							=    text[self.print_len :]
                self.print_len += len(lowerCamelCase							)
         # Otherwise, prints until the last space char (simple heuristic to avoid printing incomplete words,
         # which may change with the subsequent token -- there are probably smarter ways to do this!)
         else:
                _UpperCAmelCase							=    text[self.print_len : text.rfind(""" """							) + 1]
                self.print_len += len(lowerCamelCase							)

         self.on_finalized_text(lowerCamelCase							)


  def  lowerCamelCase       (      self		:			Dict							)						->	Optional[Any]:


         """simple docstring"""
         # Flush the cache, if it exists
         if len(self.token_cache							) > 0:
                _UpperCAmelCase							=    self.tokenizer.decode(self.token_cache       ,				**self.decode_kwargs							)
                _UpperCAmelCase							=    text[self.print_len :]
                _UpperCAmelCase							=    []
                _UpperCAmelCase							=    0
         else:
                _UpperCAmelCase							=    """"""

         _UpperCAmelCase							=    True
         self.on_finalized_text(lowerCamelCase       ,				stream_end=lowerCamelCase							)


  def  lowerCamelCase       (      self		:			Optional[int]       ,				lowerCamelCase		:			str       ,				lowerCamelCase		:			bool = False							)						->	Any:


         """simple docstring"""
         print(lowerCamelCase       ,				flush=lowerCamelCase       ,				end="""""" if not stream_end else None							)





  def  lowerCamelCase       (      self		:			int       ,				lowerCamelCase		:			Any							)						->	int:


         """simple docstring"""
         # This defines a "chinese character" as anything in the CJK Unicode block:
         #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
         #
         # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
         # despite its name. The modern Korean Hangul alphabet is a different block,
         # as is Japanese Hiragana and Katakana. Those alphabets are used to write
         # space-separated words, so they are not treated specially and handled
         # like the all of the other languages.
         if (
             (cp >= 0X4E00 and cp <= 0X9FFF)
             or (cp >= 0X3400 and cp <= 0X4DBF)  #
             or (cp >= 0X2_0000 and cp <= 0X2_A6DF)  #
             or (cp >= 0X2_A700 and cp <= 0X2_B73F)  #
             or (cp >= 0X2_B740 and cp <= 0X2_B81F)  #
             or (cp >= 0X2_B820 and cp <= 0X2_CEAF)  #
             or (cp >= 0XF900 and cp <= 0XFAFF)
             or (cp >= 0X2_F800 and cp <= 0X2_FA1F)  #
         ):  #
                return True

         return False



class 		SCREAMING_SNAKE_CASE__	(  UpperCAmelCase     ):

  '''simple docstring'''







  def __init__(      self		:			Optional[int]       ,				lowerCamelCase		:			"AutoTokenizer"       ,				lowerCamelCase		:			bool = False       ,				lowerCamelCase		:			Optional[float] = None       ,				**lowerCamelCase		:			Optional[Any]							)						->	Tuple:


         """simple docstring"""
         super().__init__(lowerCamelCase       ,				lowerCamelCase       ,				**lowerCamelCase							)
         _UpperCAmelCase							=    Queue()
         _UpperCAmelCase							=    None
         _UpperCAmelCase							=    timeout


  def  lowerCamelCase       (      self		:			Optional[Any]       ,				lowerCamelCase		:			str       ,				lowerCamelCase		:			bool = False							)						->	Optional[int]:


         """simple docstring"""
         self.text_queue.put(lowerCamelCase       ,				timeout=self.timeout							)
         if stream_end:
                self.text_queue.put(self.stop_signal       ,				timeout=self.timeout							)


  def __iter__(      self		:			Any							)						->	Union[str, Any]:


         """simple docstring"""
         return self





  def  lowerCamelCase       (      self		:			int							)						->	List[str]:


         """simple docstring"""
         _UpperCAmelCase							=    self.text_queue.get(timeout=self.timeout							)
         if value == self.stop_signal:
                raise StopIteration()
         else:
                return value