
"""simple docstring"""





import warnings

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


lowerCAmelCase_     =	logging.get_logger(__name__)

lowerCAmelCase_     =	{
    "RUCAIBox/mvp": "https://huggingface.co/RUCAIBox/mvp/resolve/main/config.json",
}




class  _snake_case		(    UpperCAmelCase_ ):







				"""simple docstring"""






				a			=						'mvp'
				a			=						['past_key_values']
				a			=						{'num_attention_heads': 'encoder_attention_heads', 'hidden_size': 'd_model'}





				def __init__(      self							:							str      ,    _A							:							Optional[Any]=5_0_2_6_7      ,    _A							:							Optional[int]=1_0_2_4      ,    _A							:							Dict=1_2      ,    _A							:							Tuple=4_0_9_6      ,    _A							:							Dict=1_6      ,    _A							:							Tuple=1_2      ,    _A							:							int=4_0_9_6      ,    _A							:							List[Any]=1_6      ,    _A							:							Optional[Any]=0.0      ,    _A							:							Optional[int]=0.0      ,    _A							:							str="gelu"      ,    _A							:							Union[str, Any]=1_0_2_4      ,    _A							:							int=0.1      ,    _A							:							Union[str, Any]=0.0      ,    _A							:							Optional[Any]=0.0      ,    _A							:							int=0.02      ,    _A							:							int=0.0      ,    _A							:							List[Any]=False      ,    _A							:							Optional[int]=True      ,    _A							:							Dict=1      ,    _A							:							Optional[Any]=0      ,    _A							:							List[Any]=2      ,    _A							:							Optional[int]=True      ,    _A							:							Dict=2      ,    _A							:							List[str]=2      ,    _A							:							Tuple=False      ,    _A							:							Optional[int]=1_0_0      ,    _A							:							Tuple=8_0_0      ,    **_A							:							int      ,    ):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    vocab_size
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    max_position_embeddings
											_SCREAMING_SNAKE_CASE		:						str			  =    d_model
											_SCREAMING_SNAKE_CASE		:						Any			  =    encoder_ffn_dim
											_SCREAMING_SNAKE_CASE		:						Dict			  =    encoder_layers
											_SCREAMING_SNAKE_CASE		:						int			  =    encoder_attention_heads
											_SCREAMING_SNAKE_CASE		:						Any			  =    decoder_ffn_dim
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    decoder_layers
											_SCREAMING_SNAKE_CASE		:						Union[str, Any]			  =    decoder_attention_heads
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    dropout
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    attention_dropout
											_SCREAMING_SNAKE_CASE		:						int			  =    activation_dropout
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    activation_function
											_SCREAMING_SNAKE_CASE		:						int			  =    init_std
											_SCREAMING_SNAKE_CASE		:						str			  =    encoder_layerdrop
											_SCREAMING_SNAKE_CASE		:						str			  =    decoder_layerdrop
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    classifier_dropout
											_SCREAMING_SNAKE_CASE		:						Any			  =    use_cache
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    encoder_layers
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    scale_embedding  # scale factor will be sqrt(d_model) if True
											_SCREAMING_SNAKE_CASE		:						int			  =    use_prompt
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    prompt_length
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    prompt_mid_dim

											super().__init__(
											    pad_token_id=_lowercase      ,    bos_token_id=_lowercase      ,    eos_token_id=_lowercase      ,    is_encoder_decoder=_lowercase      ,    decoder_start_token_id=_lowercase      ,    forced_eos_token_id=_lowercase      ,    **_lowercase      ,    )

											if self.forced_bos_token_id is None and kwargs.get("""force_bos_token_to_be_generated"""      ,    _lowercase):
																		_SCREAMING_SNAKE_CASE		:						Any			  =    self.bos_token_id
																		warnings.warn(
																		    f"""Please make sure the config includes `forced_bos_token_id={self.bos_token_id}` in future versions. """
																		    """The config can simply be saved and uploaded again to be fixed.""")



"""simple docstring"""





import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
			import torch

			from transformers import AutoModelForCausalLM




@require_torch
class  _snake_case		(    unittest.TestCase ):







				"""simple docstring"""






				def 		_lowerCAmelCase		(      self							:							List[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    -1

											_SCREAMING_SNAKE_CASE		:						List[str]			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Dict			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						Dict			  =    tokenizer.decode(greedy_ids[0])

											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Any			  =    TextStreamer(_A)
																		model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A      ,    streamer=_A)
											# The greedy text should be printed to stdout, except for the final "\n" in the streamer
											_SCREAMING_SNAKE_CASE		:						str			  =    cs.out[:-1]

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							Any):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Any			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    -1

											_SCREAMING_SNAKE_CASE		:						str			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    tokenizer.decode(greedy_ids[0])

											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    TextIteratorStreamer(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    {"""input_ids""": input_ids, """max_new_tokens""": 1_0, """do_sample""": False, """streamer""": streamer}
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    Thread(target=model.generate      ,    kwargs=_A)
											thread.start()
											_SCREAMING_SNAKE_CASE		:						Any			  =    """"""
											for new_text in streamer:
																		streamer_text += new_text

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							List[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Dict			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    -1

											_SCREAMING_SNAKE_CASE		:						Any			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						str			  =    greedy_ids[:, input_ids.shape[1] :]
											_SCREAMING_SNAKE_CASE		:						Dict			  =    tokenizer.decode(new_greedy_ids[0])

											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Any			  =    TextStreamer(_A      ,    skip_prompt=_A)
																		model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A      ,    streamer=_A)
											# The greedy text should be printed to stdout, except for the final "\n" in the streamer
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    cs.out[:-1]

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							Dict):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						int			  =    AutoTokenizer.from_pretrained("""distilgpt2""")
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoModelForCausalLM.from_pretrained("""distilgpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						int			  =    -1

											_SCREAMING_SNAKE_CASE		:						List[str]			  =    torch.ones((1, 5)      ,    device=_A).long() * model.config.bos_token_id
											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    TextStreamer(_A      ,    skip_special_tokens=_A)
																		model.generate(_A      ,    max_new_tokens=1      ,    do_sample=_A      ,    streamer=_A)

											# The prompt contains a special token, so the streamer should not print it. As such, the output text, when
											# re-tokenized, must only contain one token
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    cs.out[:-1]  # Remove the final "\n"
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    tokenizer(_A      ,    return_tensors="""pt""")
											self.assertEqual(streamer_text_tokenized.input_ids.shape      ,    (1, 1))





				def 		_lowerCAmelCase		(      self							:							str):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    -1

											_SCREAMING_SNAKE_CASE		:						Dict			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						int			  =    TextIteratorStreamer(_A      ,    timeout=0.001)
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    {"""input_ids""": input_ids, """max_new_tokens""": 1_0, """do_sample""": False, """streamer""": streamer}
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    Thread(target=model.generate      ,    kwargs=_A)
											thread.start()

											# The streamer will timeout after 0.001 seconds, so an exception will be raised
											with self.assertRaises(_A):
																		_SCREAMING_SNAKE_CASE		:						str			  =    """"""
																		for new_text in streamer:
																									streamer_text += new_text


