



"""simple docstring"""







import unittest

from transformers import (
    MODEL_FOR_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
    TextGenerationPipeline,
    logging,
    pipeline,
)
from transformers.testing_utils import (
    CaptureLogger,
    is_pipeline_test,
    require_accelerate,
    require_tf,
    require_torch,
    require_torch_gpu,
    require_torch_or_tf,
)

from .test_pipelines_common import ANY






@is_pipeline_test
@require_torch_or_tf
class      UpperCAmelCase__		(      unittest.TestCase     ):
		lowerCAmelCase_       :	int 							=     MODEL_FOR_CAUSAL_LM_MAPPING
		lowerCAmelCase_       :	List[str] 							=     TF_MODEL_FOR_CAUSAL_LM_MAPPING



		@require_torch
		def 					A_		( self  :    int				)			->						Optional[int]:






									'''simple docstring'''

									A         =					pipeline(task='text-generation'      ,							model='sshleifer/tiny-ctrl'      ,							framework='pt'				)
									# Using `do_sample=False` to force deterministic output
									A         =					text_generator('This is a test'      ,							do_sample=snake_case				)
									self.assertEqual(
									    snake_case      ,							[
									        {
									            'generated_text': (
									                'This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope.'
									                ' oscope. FiliFili@@'
									            )
									        }
									    ]      ,							)

									A         =					text_generator(['This is a test', 'This is a second test']				)
									self.assertEqual(
									    snake_case      ,							[
									        [
									            {
									                'generated_text': (
									                    'This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope.'
									                    ' oscope. FiliFili@@'
									                )
									            }
									        ],
									        [
									            {
									                'generated_text': (
									                    'This is a second test ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy'
									                    ' oscope. oscope. FiliFili@@'
									                )
									            }
									        ],
									    ]      ,							)

									A         =					text_generator('This is a test'      ,							do_sample=snake_case      ,							num_return_sequences=2      ,							return_tensors=snake_case				)
									self.assertEqual(
									    snake_case      ,							[
									        {'generated_token_ids': ANY(snake_case				)},
									        {'generated_token_ids': ANY(snake_case				)},
									    ]      ,							)
									A         =					text_generator.model.config.eos_token_id
									A         =					'<pad>'
									A         =					text_generator(
									    ['This is a test', 'This is a second test']      ,							do_sample=snake_case      ,							num_return_sequences=2      ,							batch_size=2      ,							return_tensors=snake_case      ,							)
									self.assertEqual(
									    snake_case      ,							[
									        [
									            {'generated_token_ids': ANY(snake_case				)},
									            {'generated_token_ids': ANY(snake_case				)},
									        ],
									        [
									            {'generated_token_ids': ANY(snake_case				)},
									            {'generated_token_ids': ANY(snake_case				)},
									        ],
									    ]      ,							)



		@require_tf
		def 					A_		( self  :    List[str]				)			->						Union[str, Any]:






									'''simple docstring'''

									A         =					pipeline(task='text-generation'      ,							model='sshleifer/tiny-ctrl'      ,							framework='tf'				)

									# Using `do_sample=False` to force deterministic output
									A         =					text_generator('This is a test'      ,							do_sample=snake_case				)
									self.assertEqual(
									    snake_case      ,							[
									        {
									            'generated_text': (
									                'This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵'
									                ' please,'
									            )
									        }
									    ]      ,							)

									A         =					text_generator(['This is a test', 'This is a second test']      ,							do_sample=snake_case				)
									self.assertEqual(
									    snake_case      ,							[
									        [
									            {
									                'generated_text': (
									                    'This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵'
									                    ' please,'
									                )
									            }
									        ],
									        [
									            {
									                'generated_text': (
									                    'This is a second test Chieftain Chieftain prefecture prefecture prefecture Cannes Cannes'
									                    ' Cannes 閲閲Cannes Cannes Cannes 攵 please,'
									                )
									            }
									        ],
									    ]      ,							)



		def 					A_		( self  :    Optional[Any]      ,							snake_case  :    Optional[Any]      ,							snake_case  :    str      ,							snake_case  :    List[Any]				)			->						Any:






									'''simple docstring'''

									A         =					TextGenerationPipeline(model=snake_case      ,							tokenizer=snake_case				)
									return text_generator, ["This is a test", "Another test"]



		def 					A_		( self  :    Dict				)			->						Any:






									'''simple docstring'''

									A         =					'Hello I believe in'
									A         =					pipeline('text-generation'      ,							model='hf-internal-testing/tiny-random-gpt2'				)
									A         =					text_generator(snake_case				)
									self.assertEqual(
									    snake_case      ,							[{'generated_text': 'Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe'}]      ,							)

									A         =					text_generator(snake_case      ,							stop_sequence=' fe'				)
									self.assertEqual(snake_case      ,							[{'generated_text': 'Hello I believe in fe'}]				)



		def 					A_		( self  :    Union[str, Any]      ,							snake_case  :    int      ,							snake_case  :    Tuple				)			->						Optional[Any]:






									'''simple docstring'''

									A         =					text_generator.model
									A         =					text_generator.tokenizer

									A         =					text_generator('This is a test'				)
									self.assertEqual(snake_case      ,							[{'generated_text': ANY(snake_case				)}]				)
									self.assertTrue(outputs[0]['generated_text'].startswith('This is a test'				)				)

									A         =					text_generator('This is a test'      ,							return_full_text=snake_case				)
									self.assertEqual(snake_case      ,							[{'generated_text': ANY(snake_case				)}]				)
									self.assertNotIn('This is a test'      ,							outputs[0]['generated_text']				)

									A         =					pipeline(task='text-generation'      ,							model=snake_case      ,							tokenizer=snake_case      ,							return_full_text=snake_case				)
									A         =					text_generator('This is a test'				)
									self.assertEqual(snake_case      ,							[{'generated_text': ANY(snake_case				)}]				)
									self.assertNotIn('This is a test'      ,							outputs[0]['generated_text']				)

									A         =					text_generator('This is a test'      ,							return_full_text=snake_case				)
									self.assertEqual(snake_case      ,							[{'generated_text': ANY(snake_case				)}]				)
									self.assertTrue(outputs[0]['generated_text'].startswith('This is a test'				)				)

									A         =					text_generator(['This is great !', 'Something else']      ,							num_return_sequences=2      ,							do_sample=snake_case				)
									self.assertEqual(
									    snake_case      ,							[
									        [{'generated_text': ANY(snake_case				)}, {'generated_text': ANY(snake_case				)}],
									        [{'generated_text': ANY(snake_case				)}, {'generated_text': ANY(snake_case				)}],
									    ]      ,							)

									if text_generator.tokenizer.pad_token is not None:
																A         =					text_generator(
																    ['This is great !', 'Something else']      ,							num_return_sequences=2      ,							batch_size=2      ,							do_sample=snake_case				)
																self.assertEqual(
																    snake_case      ,							[
																        [{'generated_text': ANY(snake_case				)}, {'generated_text': ANY(snake_case				)}],
																        [{'generated_text': ANY(snake_case				)}, {'generated_text': ANY(snake_case				)}],
																    ]      ,							)

									with self.assertRaises(snake_case				):
																A         =					text_generator('test'      ,							return_full_text=snake_case      ,							return_text=snake_case				)
									with self.assertRaises(snake_case				):
																A         =					text_generator('test'      ,							return_full_text=snake_case      ,							return_tensors=snake_case				)
									with self.assertRaises(snake_case				):
																A         =					text_generator('test'      ,							return_text=snake_case      ,							return_tensors=snake_case				)

									# Empty prompt is slighly special
									# it requires BOS token to exist.
									# Special case for Pegasus which will always append EOS so will
									# work even without BOS.
									if (
									    text_generator.tokenizer.bos_token_id is not None
									    or "Pegasus" in tokenizer.__class__.__name__
									    or "Git" in model.__class__.__name__
									):
																A         =					text_generator(''				)
																self.assertEqual(snake_case      ,							[{'generated_text': ANY(snake_case				)}]				)
									else:
																with self.assertRaises((ValueError, AssertionError)				):
																							A         =					text_generator(''				)

									if text_generator.framework == "tf":
																# TF generation does not support max_new_tokens, and it's impossible
																# to control long generation with only max_length without
																# fancy calculation, dismissing tests for now.
																return
									# We don't care about infinite range models.
									# They already work.
									# Skip this test for XGLM, since it uses sinusoidal positional embeddings which are resized on-the-fly.
									A         =					['RwkvForCausalLM', 'XGLMForCausalLM', 'GPTNeoXForCausalLM']
									if (
									    tokenizer.model_max_length < 10_000
									    and text_generator.model.__class__.__name__ not in EXTRA_MODELS_CAN_HANDLE_LONG_INPUTS
									):
																# Handling of large generations
																with self.assertRaises((RuntimeError, IndexError, ValueError, AssertionError)				):
																							text_generator('This is a test' * 500      ,							max_new_tokens=20				)

																A         =					text_generator('This is a test' * 500      ,							handle_long_generation='hole'      ,							max_new_tokens=20				)
																# Hole strategy cannot work
																with self.assertRaises(snake_case				):
																							text_generator(
																							    'This is a test' * 500      ,							handle_long_generation='hole'      ,							max_new_tokens=tokenizer.model_max_length + 10      ,							)



		@require_torch
		@require_accelerate
		@require_torch_gpu
		def 					A_		( self  :    Dict				)			->						Dict:






									'''simple docstring'''

									import torch

									# Classic `model_kwargs`
									A         =					pipeline(
									    model='hf-internal-testing/tiny-random-bloom'      ,							model_kwargs={'device_map': 'auto', 'torch_dtype': torch.bfloataa}      ,							)
									self.assertEqual(pipe.model.device      ,							torch.device(0				)				)
									self.assertEqual(pipe.model.lm_head.weight.dtype      ,							torch.bfloataa				)
									A         =					pipe('This is a test'				)
									self.assertEqual(
									    snake_case      ,							[
									        {
									            'generated_text': (
									                'This is a test test test test test test test test test test test test test test test test'
									                ' test'
									            )
									        }
									    ]      ,							)

									# Upgraded those two to real pipeline arguments (they just get sent for the model as they're unlikely to mean anything else.)
									A         =					pipeline(model='hf-internal-testing/tiny-random-bloom'      ,							device_map='auto'      ,							torch_dtype=torch.bfloataa				)
									self.assertEqual(pipe.model.device      ,							torch.device(0				)				)
									self.assertEqual(pipe.model.lm_head.weight.dtype      ,							torch.bfloataa				)
									A         =					pipe('This is a test'				)
									self.assertEqual(
									    snake_case      ,							[
									        {
									            'generated_text': (
									                'This is a test test test test test test test test test test test test test test test test'
									                ' test'
									            )
									        }
									    ]      ,							)

									# torch_dtype will be automatically set to float32 if not provided - check: https://github.com/huggingface/transformers/pull/20602
									A         =					pipeline(model='hf-internal-testing/tiny-random-bloom'      ,							device_map='auto'				)
									self.assertEqual(pipe.model.device      ,							torch.device(0				)				)
									self.assertEqual(pipe.model.lm_head.weight.dtype      ,							torch.floataa				)
									A         =					pipe('This is a test'				)
									self.assertEqual(
									    snake_case      ,							[
									        {
									            'generated_text': (
									                'This is a test test test test test test test test test test test test test test test test'
									                ' test'
									            )
									        }
									    ]      ,							)



		@require_torch
		@require_torch_gpu
		def 					A_		( self  :    Optional[Any]				)			->						Any:






									'''simple docstring'''

									import torch

									A         =					pipeline(model='hf-internal-testing/tiny-random-bloom'      ,							device=0      ,							torch_dtype=torch.floataa				)
									pipe('This is a test'				)



		@require_torch
		@require_accelerate
		@require_torch_gpu
		def 					A_		( self  :    Any				)			->						Tuple:






									'''simple docstring'''

									import torch

									A         =					pipeline(model='hf-internal-testing/tiny-random-bloom'      ,							device_map='auto'      ,							torch_dtype=torch.floataa				)
									pipe('This is a test'      ,							do_sample=snake_case      ,							top_p=0.5				)



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






									'''simple docstring'''

									A         =					'Hello world'
									A         =					pipeline('text-generation'      ,							model='hf-internal-testing/tiny-random-gpt2'				)
									if text_generator.model.framework == "tf":
																A         =					logging.get_logger('transformers.generation.tf_utils'				)
									else:
																A         =					logging.get_logger('transformers.generation.utils'				)
									A         =					'Both `max_new_tokens`'  # The beggining of the message to be checked in this test

									# Both are set by the user -> log warning
									with CaptureLogger(snake_case				) as cl:
																A         =					text_generator(snake_case      ,							max_length=10      ,							max_new_tokens=1				)
									self.assertIn(snake_case      ,							cl.out				)

									# The user only sets one -> no warning
									with CaptureLogger(snake_case				) as cl:
																A         =					text_generator(snake_case      ,							max_new_tokens=1				)
									self.assertNotIn(snake_case      ,							cl.out				)

									with CaptureLogger(snake_case				) as cl:
																A         =					text_generator(snake_case      ,							max_length=10				)
									self.assertNotIn(snake_case      ,							cl.out				)








"""simple docstring"""







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


A    =			logging.get_logger(__name__)

A    =			{
    'edbeeching/decision-transformer-gym-hopper-medium': (
        'https://huggingface.co/edbeeching/decision-transformer-gym-hopper-medium/resolve/main/config.json'
    ),
    # See all DecisionTransformer models at https://huggingface.co/models?filter=decision_transformer
}






class      UpperCAmelCase__		(      UpperCamelCase     ):
		lowerCAmelCase_       :	Tuple 							=     """decision_transformer"""
		lowerCAmelCase_       :	Tuple 							=     ["""past_key_values"""]
		lowerCAmelCase_       :	Dict 							=     {
		    """max_position_embeddings""": """n_positions""",
		    """num_attention_heads""": """n_head""",
		    """num_hidden_layers""": """n_layer""",
		}



		def __init__( self  :    Optional[Any]      ,							snake_case  :    int=17      ,							snake_case  :    int=4      ,							snake_case  :    Union[str, Any]=128      ,							snake_case  :    Optional[Any]=4_096      ,							snake_case  :    List[Any]=True      ,							snake_case  :    str=1      ,							snake_case  :    str=1_024      ,							snake_case  :    List[str]=3      ,							snake_case  :    str=1      ,							snake_case  :    Any=None      ,							snake_case  :    Optional[int]="relu"      ,							snake_case  :    List[Any]=0.1      ,							snake_case  :    str=0.1      ,							snake_case  :    Tuple=0.1      ,							snake_case  :    Union[str, Any]=1E-5      ,							snake_case  :    Optional[int]=0.02      ,							snake_case  :    List[Any]=True      ,							snake_case  :    Tuple=True      ,							snake_case  :    Optional[Any]=50_256      ,							snake_case  :    List[Any]=50_256      ,							snake_case  :    List[str]=False      ,							snake_case  :    Any=False      ,							**snake_case  :    List[str]      ,							)			->						List[Any]:






									'''simple docstring'''

									A         =					state_dim
									A         =					act_dim
									A         =					hidden_size
									A         =					max_ep_len
									A         =					action_tanh
									A         =					vocab_size
									A         =					n_positions
									A         =					n_layer
									A         =					n_head
									A         =					n_inner
									A         =					activation_function
									A         =					resid_pdrop
									A         =					embd_pdrop
									A         =					attn_pdrop
									A         =					layer_norm_epsilon
									A         =					initializer_range
									A         =					scale_attn_weights
									A         =					use_cache
									A         =					scale_attn_by_inverse_layer_idx
									A         =					reorder_and_upcast_attn

									A         =					bos_token_id
									A         =					eos_token_id

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



