

import unittest

from transformers import (
    MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TextaTextGenerationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, require_tf, require_torch
from transformers.utils import is_torch_available

from .test_pipelines_common import ANY


if is_torch_available():
	import torch



@is_pipeline_test
class 					_a		(unittest.TestCase       ):


					'''simple docstring'''






					UpperCAmelCase__:     Union[str, Any]													= MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
					UpperCAmelCase__:     Dict													= TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING






					def 						__A    (				self					, A__					, A__					, A__    ):
						A__							:					int       =	TextaTextGenerationPipeline(model=_a					, tokenizer=_a    )
						return generator, ["Something to write", "Something else"]






					def 						__A    (				self					, A__					, A__    ):
						A__							:					Union[str, Any]       =	generator("""Something there"""    )
						self.assertEqual(_a					, [{"""generated_text""": ANY(_a    )}]    )
						# These are encoder decoder, they don't just append to incoming string
						self.assertFalse(outputs[0]["""generated_text"""].startswith("""Something there"""    )    )

						A__							:					Optional[Any]       =	generator(["""This is great !""", """Something else"""]					, num_return_sequences=2					, do_sample=_a    )
						self.assertEqual(
						    _a					, [
						        [{"""generated_text""": ANY(_a    )}, {"""generated_text""": ANY(_a    )}],
						        [{"""generated_text""": ANY(_a    )}, {"""generated_text""": ANY(_a    )}],
						    ]					, )

						A__							:					Tuple       =	generator(
						    ["""This is great !""", """Something else"""]					, num_return_sequences=2					, batch_size=2					, do_sample=_a    )
						self.assertEqual(
						    _a					, [
						        [{"""generated_text""": ANY(_a    )}, {"""generated_text""": ANY(_a    )}],
						        [{"""generated_text""": ANY(_a    )}, {"""generated_text""": ANY(_a    )}],
						    ]					, )

						with self.assertRaises(_a    ):
							generator(4    )






					@require_torch
					def 						__A    (				self    ):
						A__							:					Tuple       =	pipeline("""text2text-generation"""					, model="""patrickvonplaten/t5-tiny-random"""					, framework="""pt"""    )
						# do_sample=False necessary for reproducibility
						A__							:					List[Any]       =	generator("""Something there"""					, do_sample=_a    )
						self.assertEqual(_a					, [{"""generated_text""": """"""}]    )

						A__							:					str       =	3
						A__							:					Optional[int]       =	generator(
						    """Something there"""					, num_return_sequences=_a					, num_beams=_a					, )
						A__							:					List[str]       =	[
						    {"""generated_text""": """Beide Beide Beide Beide Beide Beide Beide Beide Beide"""},
						    {"""generated_text""": """Beide Beide Beide Beide Beide Beide Beide Beide"""},
						    {"""generated_text""": """"""},
						]
						self.assertEqual(_a					, _a    )

						A__							:					Optional[Any]       =	generator("""This is a test"""					, do_sample=_a					, num_return_sequences=2					, return_tensors=_a    )
						self.assertEqual(
						    _a					, [
						        {"""generated_token_ids""": ANY(torch.Tensor    )},
						        {"""generated_token_ids""": ANY(torch.Tensor    )},
						    ]					, )
						A__							:					int       =	generator.model.config.eos_token_id
						A__							:					List[Any]       =	"""<pad>"""
						A__							:					str       =	generator(
						    ["""This is a test""", """This is a second test"""]					, do_sample=_a					, num_return_sequences=2					, batch_size=2					, return_tensors=_a					, )
						self.assertEqual(
						    _a					, [
						        [
						            {"""generated_token_ids""": ANY(torch.Tensor    )},
						            {"""generated_token_ids""": ANY(torch.Tensor    )},
						        ],
						        [
						            {"""generated_token_ids""": ANY(torch.Tensor    )},
						            {"""generated_token_ids""": ANY(torch.Tensor    )},
						        ],
						    ]					, )






					@require_tf
					def 						__A    (				self    ):
						A__							:					Optional[Any]       =	pipeline("""text2text-generation"""					, model="""patrickvonplaten/t5-tiny-random"""					, framework="""tf"""    )
						# do_sample=False necessary for reproducibility
						A__							:					Optional[Any]       =	generator("""Something there"""					, do_sample=_a    )
						self.assertEqual(_a					, [{"""generated_text""": """"""}]    )






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 					_a		(__magic_name__	,     __magic_name__	,     __magic_name__       ):


					'''simple docstring'''






					UpperCAmelCase__:     str													= [r'''h\.\d+\.attn\.bias''', r'''h\.\d+\.attn\.masked_bias''']






					@register_to_config
					def __init__(				self					, A__					, A__					, A__ = None					, A__ = 5_0257					, A__ = 1024					, A__ = 768					, A__ = 12					, A__ = 12					, A__ = None					, A__ = "gelu_new"					, A__ = 0.1					, A__ = 0.1					, A__ = 0.1					, A__ = 1e-5					, A__ = 0.0_2					, A__ = True					, A__ = True					, A__ = False					, A__ = False					, ):
						super().__init__()

						A__							:					Union[str, Any]       =	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."""    )

						A__							:					str       =	prefix_inner_dim
						A__							:					Optional[Any]       =	prefix_hidden_dim

						A__							:					Tuple       =	(
						    nn.Linear(self.prefix_inner_dim					, self.prefix_hidden_dim    )
						    if self.prefix_hidden_dim is not None
						    else nn.Identity()
						)
						A__							:					int       =	(
						    nn.Linear(self.prefix_hidden_dim					, A__    ) if self.prefix_hidden_dim is not None else nn.Identity()
						)

						A__							:					Tuple       =	GPTaConfig(
						    vocab_size=A__					, n_positions=A__					, n_embd=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__					, )
						A__							:					int       =	GPTaLMHeadModel(A__    )






					def 						__A    (				self					, A__					, A__					, A__ = None					, A__ = None					, ):
						A__							:					List[str]       =	self.transformer.transformer.wte(A__    )
						A__							:					int       =	self.encode_prefix(A__    )
						A__							:					int       =	self.decode_prefix(A__    )
						A__							:					Optional[Any]       =	torch.cat((prefix_embeds, embedding_text)					, dim=1    )

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






					def 						__A    (				self					, A__					, A__    ):
						return torch.zeros(A__					, self.prefix_length					, dtype=torch.intaa					, device=A__    )






					def 						__A    (				self					, A__    ):
						return self.encode_prefix(A__    )






					@torch.no_grad()
					def 						__A    (				self					, A__					, A__					, A__    ):
						A__							:					List[Any]       =	torch.split(A__					, 1					, dim=0    )
						A__							:					Optional[int]       =	[]
						A__							:					str       =	[]
						for feature in features:
							A__							:					Dict       =	self.decode_prefix(feature.to(A__    )    )  # back to the clip feature
							# Only support beam search for now
							A__     ,     A__							:					Union[str, Any]       =	self.generate_beam(
							    input_embeds=A__					, device=A__					, eos_token_id=A__    )
							generated_tokens.append(output_tokens[0]    )
							generated_seq_lengths.append(seq_lengths[0]    )
						A__							:					int       =	torch.stack(A__    )
						A__							:					List[Any]       =	torch.stack(A__    )
						return generated_tokens, generated_seq_lengths






					@torch.no_grad()
					def 						__A    (				self					, A__=None					, A__=None					, A__=None					, A__ = 5					, A__ = 67					, A__ = 1.0					, A__ = None					, ):
						A__							:					Any       =	eos_token_id
						A__							:					Any       =	None
						A__							:					Optional[int]       =	None
						A__							:					Optional[Any]       =	torch.ones(A__					, device=A__					, dtype=torch.int    )
						A__							:					Any       =	torch.zeros(A__					, device=A__					, dtype=torch.bool    )

						if input_embeds is not None:
							A__							:					Dict       =	input_embeds
						else:
							A__							:					str       =	self.transformer.transformer.wte(A__    )

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

							if scores is None:
								A__     ,     A__							:					Optional[int]       =	logits.topk(A__					, -1    )
								A__							:					List[Any]       =	generated.expand(A__					, *generated.shape[1:]    )
								A__     ,     A__							:					List[Any]       =	next_tokens.permute(1					, 0    ), scores.squeeze(0    )
								if tokens is None:
									A__							:					Optional[Any]       =	next_tokens
								else:
									A__							:					List[Any]       =	tokens.expand(A__					, *tokens.shape[1:]    )
									A__							:					int       =	torch.cat((tokens, next_tokens)					, dim=1    )
							else:
								A__							:					Optional[int]       =	-float(np.inf    )
								A__							:					List[Any]       =	0
								A__							:					str       =	scores[:, None] + logits
								seq_lengths[~is_stopped] += 1
								A__							:					Dict       =	scores_sum / seq_lengths[:, None]
								A__     ,     A__							:					List[Any]       =	scores_sum_average.view(-1    ).topk(A__					, -1    )
								A__							:					Tuple       =	next_tokens // scores_sum.shape[1]
								A__							:					Optional[Any]       =	seq_lengths[next_tokens_source]
								A__							:					List[str]       =	next_tokens % scores_sum.shape[1]
								A__							:					Optional[int]       =	next_tokens.unsqueeze(1    )
								A__							:					int       =	tokens[next_tokens_source]
								A__							:					List[Any]       =	torch.cat((tokens, next_tokens)					, dim=1    )
								A__							:					str       =	generated[next_tokens_source]
								A__							:					Optional[Any]       =	scores_sum_average * seq_lengths
								A__							:					Union[str, Any]       =	is_stopped[next_tokens_source]

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

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



