



import uuid
from typing import Any, Dict, List, Optional, Union

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
				import tensorflow as tf

if is_torch_available():
				import torch


_lowercase     							=						logging.get_logger(__name__)
class 				__A					:





			def __init__(self							,    __magic_name__ = None							,    __magic_name__ = None							,    __magic_name__=None							,    __magic_name__=None				):
						if not conversation_id:
									lowerCamelCase__ :      Dict								=		uuid.uuida()
						if past_user_inputs is None:
									lowerCamelCase__ :      List[Any]								=		[]
						if generated_responses is None:
									lowerCamelCase__ :      str								=		[]

						lowerCamelCase__ :      uuid.UUID								=		conversation_id
						lowerCamelCase__ :      List[str]								=		past_user_inputs
						lowerCamelCase__ :      List[str]								=		generated_responses
						lowerCamelCase__ :      Optional[str]								=		text





			def __eq__(self							,    __magic_name__				):
						if not isinstance(__magic_name__							,    __magic_name__				):
									return False
						if self.uuid == other.uuid:
									return True
						return (
						    self.new_user_input == other.new_user_input
						    and self.past_user_inputs == other.past_user_inputs
						    and self.generated_responses == other.generated_responses
						)





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__ = False				):
						if self.new_user_input:
									if overwrite:
												logger.warning(
												    f"User input added while unprocessed input was existing: \"{self.new_user_input}\" was overwritten "
												    f"with: \"{text}\"."				)
												lowerCamelCase__ :      List[str]								=		text
									else:
												logger.warning(
												    f"User input added while unprocessed input was existing: \"{self.new_user_input}\" new input "
												    f"ignored: \"{text}\". Set `overwrite` to True to overwrite unprocessed user input"				)
						else:
									lowerCamelCase__ :      Optional[int]								=		text





			def        _snake_case    (self				):
						if self.new_user_input:
									self.past_user_inputs.append(self.new_user_input				)
						lowerCamelCase__ :      Dict								=		None





			def        _snake_case    (self							,    __magic_name__				):
						self.generated_responses.append(__magic_name__				)





			def        _snake_case    (self				):
						for user_input, generated_response in zip(self.past_user_inputs							,    self.generated_responses				):
									yield True, user_input
									yield False, generated_response
						if self.new_user_input:
									yield True, self.new_user_input





			def __repr__(self				):
						lowerCamelCase__ :      Any								=		f"Conversation id: {self.uuid} \n"
						for is_user, text in self.iter_texts():
									lowerCamelCase__ :      List[Any]								=		"""user""" if is_user else """bot"""
									output += f"{name} >> {text} \n"
						return output







@add_end_docstrings(
    A_					, r'''
        min_length_for_response (`int`, *optional*, defaults to 32):
            The minimum length (in number of tokens) for a response.
        minimum_tokens (`int`, *optional*, defaults to 10):
            The minimum length of tokens to leave for a response.
    '''					, )
class 				__A					( A_      ):





			def __init__(self							,    *__magic_name__							,    **__magic_name__				):
						super().__init__(*__magic_name__							,    **__magic_name__				)
						if self.tokenizer.pad_token_id is None:
									lowerCamelCase__ :      Union[str, Any]								=		self.tokenizer.eos_token





			def        _snake_case    (self							,    __magic_name__=None							,    __magic_name__=None							,    __magic_name__=None							,    **__magic_name__				):
						lowerCamelCase__ :      List[str]								=		{}
						lowerCamelCase__ :      Any								=		{}
						lowerCamelCase__ :      Any								=		{}

						if min_length_for_response is not None:
									lowerCamelCase__ :      List[Any]								=		min_length_for_response
						if minimum_tokens is not None:
									lowerCamelCase__ :      Any								=		minimum_tokens

						if "max_length" in generate_kwargs:
									lowerCamelCase__ :      Tuple								=		generate_kwargs["""max_length"""]
									# self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
						if clean_up_tokenization_spaces is not None:
									lowerCamelCase__ :      Union[str, Any]								=		clean_up_tokenization_spaces

						if generate_kwargs:
									forward_params.update(__magic_name__				)
						return preprocess_params, forward_params, postprocess_params





			def __call__(self							,    __magic_name__							,    __magic_name__=0							,    **__magic_name__				):
						lowerCamelCase__ :      List[Any]								=		super().__call__(__magic_name__							,    num_workers=__magic_name__							,    **__magic_name__				)
						if isinstance(__magic_name__							,    __magic_name__				) and len(__magic_name__				) == 1:
									return outputs[0]
						return outputs





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__=32				):
						if not isinstance(__magic_name__							,    __magic_name__				):
									raise ValueError("""ConversationalPipeline, expects Conversation as inputs"""				)
						if conversation.new_user_input is None:
									raise ValueError(
									    f"Conversation with UUID {type(conversation.uuid				)} does not contain new user input to process. "
									    """Add user inputs with the conversation's `add_user_input` method"""				)
						if hasattr(self.tokenizer							,    """_build_conversation_input_ids"""				):
									lowerCamelCase__ :      Optional[Any]								=		self.tokenizer._build_conversation_input_ids(__magic_name__				)
						else:
									# If the tokenizer cannot handle conversations, we default to only the old version
									lowerCamelCase__ :      Any								=		self._legacy_parse_and_tokenize(__magic_name__				)

						if self.framework == "pt":
									lowerCamelCase__ :      Tuple								=		torch.LongTensor([input_ids]				)
						elif self.framework == "tf":
									lowerCamelCase__ :      Any								=		tf.constant([input_ids]				)
						return {"input_ids": input_ids, "conversation": conversation}





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__=10							,    **__magic_name__				):
						lowerCamelCase__ :      Optional[int]								=		generate_kwargs.get("""max_length"""							,    self.model.config.max_length				)

						lowerCamelCase__ :      Any								=		model_inputs["""input_ids"""].shape[1]
						if max_length - minimum_tokens < n:
									logger.warning(f"Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})"				)
									lowerCamelCase__ :      Union[str, Any]								=		max_length - minimum_tokens
									lowerCamelCase__ :      str								=		model_inputs["""input_ids"""][:, -trim:]
									if "attention_mask" in model_inputs:
												lowerCamelCase__ :      str								=		model_inputs["""attention_mask"""][:, -trim:]
						lowerCamelCase__ :      int								=		model_inputs.pop("""conversation"""				)
						lowerCamelCase__ :      Union[str, Any]								=		max_length
						lowerCamelCase__ :      Any								=		self.model.generate(**__magic_name__							,    **__magic_name__				)
						if self.model.config.is_encoder_decoder:
									lowerCamelCase__ :      List[str]								=		1
						else:
									lowerCamelCase__ :      Optional[Any]								=		n
						return {"output_ids": output_ids[:, start_position:], "conversation": conversation}





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__=True				):
						lowerCamelCase__ :      Any								=		model_outputs["""output_ids"""]
						lowerCamelCase__ :      Union[str, Any]								=		self.tokenizer.decode(
						    output_ids[0]							,    skip_special_tokens=__magic_name__							,    clean_up_tokenization_spaces=__magic_name__							,    )
						lowerCamelCase__ :      Any								=		model_outputs["""conversation"""]
						conversation.mark_processed()
						conversation.append_response(__magic_name__				)
						return conversation





			def        _snake_case    (self							,    __magic_name__				):
						lowerCamelCase__ :      Union[str, Any]								=		self.tokenizer.eos_token_id
						lowerCamelCase__ :      List[Any]								=		[]
						for is_user, text in conversation.iter_texts():
									if eos_token_id is not None:
												input_ids.extend(self.tokenizer.encode(__magic_name__							,    add_special_tokens=__magic_name__				) + [eos_token_id]				)
									else:
												input_ids.extend(self.tokenizer.encode(__magic_name__							,    add_special_tokens=__magic_name__				)				)

						if len(__magic_name__				) > self.tokenizer.model_max_length:
									lowerCamelCase__ :      Tuple								=		input_ids[-self.tokenizer.model_max_length :]
						return input_ids








import random
import unittest

import numpy as np

import transformers
from transformers import is_flax_available, is_torch_available
from transformers.testing_utils import is_pt_flax_cross_test, require_flax


if is_flax_available():
				import os

				import jax.numpy as jnp
				from jax import jit

				from transformers import AutoTokenizer, FlaxAutoModelForCausalLM
				from transformers.modeling_flax_pytorch_utils import load_flax_weights_in_pytorch_model

				_lowercase     							=						'''0.12'''  # assumed parallelism: 8


if is_torch_available():
				import torch






def      _A				(UpperCamelCase		:      Dict      ,			UpperCamelCase		:      int      ,			UpperCamelCase		:      Optional[Any]=None )  ->Optional[int]:



			'''simple docstring'''







			if rng is None:
						lowerCamelCase__ :      List[Any]								=		random.Random()

			lowerCamelCase__ :      Tuple								=		1
			for dim in shape:
						total_dims *= dim

			lowerCamelCase__ :      str								=		[]
			for _ in range(UpperCamelCase ):
						values.append(rng.randint(0      ,			vocab_size - 1 ) )

			lowerCamelCase__ :      str								=		np.array(UpperCamelCase      ,			dtype=jnp.intaa ).reshape(UpperCamelCase )

			return output






def      _A				(UpperCamelCase		:      str      ,			UpperCamelCase		:      Dict=None )  ->List[str]:



			'''simple docstring'''







			lowerCamelCase__ :      Any								=		ids_tensor(UpperCamelCase      ,			vocab_size=2      ,			rng=UpperCamelCase )
			# make sure that at least one token is attended to for each batch
			lowerCamelCase__ :      int								=		1
			return attn_mask
@require_flax
class 				__A					:
			UpperCamelCase					:List[Any]     =			None
			UpperCamelCase					:Dict     =			()





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__ :      Dict								=		self.model_tester.prepare_config_and_inputs_for_common()

						# cut to half length & take max batch_size 3
						lowerCamelCase__ :      List[Any]								=		2
						lowerCamelCase__ :      Any								=		inputs["""input_ids"""].shape[-1] // 2
						lowerCamelCase__ :      Dict								=		inputs["""input_ids"""][:max_batch_size, :sequence_length]

						lowerCamelCase__ :      List[Any]								=		jnp.ones_like(__magic_name__				)
						lowerCamelCase__ :      Dict								=		attention_mask[:max_batch_size, :sequence_length]

						# generate max 5 tokens
						lowerCamelCase__ :      Tuple								=		input_ids.shape[-1] + 5
						if config.eos_token_id is not None and config.pad_token_id is None:
									# hack to allow generate for models such as GPT2 as is done in `generate()`
									lowerCamelCase__ :      List[Any]								=		config.eos_token_id
						return config, input_ids, attention_mask, max_length





			@is_pt_flax_cross_test
			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      List[Any]								=		self._get_input_ids_and_config()
						lowerCamelCase__ :      Tuple								=		False
						lowerCamelCase__ :      Tuple								=		max_length
						lowerCamelCase__ :      List[Any]								=		0

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      List[Any]								=		model_class(__magic_name__				)

									lowerCamelCase__ :      Tuple								=		model_class.__name__[4:]  # Skip the "Flax" at the beginning
									lowerCamelCase__ :      Union[str, Any]								=		getattr(__magic_name__							,    __magic_name__				)
									lowerCamelCase__ :      Dict								=		pt_model_class(__magic_name__				).eval()
									lowerCamelCase__ :      Union[str, Any]								=		load_flax_weights_in_pytorch_model(__magic_name__							,    flax_model.params				)

									lowerCamelCase__ :      str								=		flax_model.generate(__magic_name__				).sequences
									lowerCamelCase__ :      str								=		pt_model.generate(torch.tensor(__magic_name__							,    dtype=torch.long				)				)

									if flax_generation_outputs.shape[-1] > pt_generation_outputs.shape[-1]:
												lowerCamelCase__ :      Optional[Any]								=		flax_generation_outputs[:, : pt_generation_outputs.shape[-1]]

									self.assertListEqual(pt_generation_outputs.numpy().tolist()							,    flax_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      Tuple								=		self._get_input_ids_and_config()
						lowerCamelCase__ :      Dict								=		False
						lowerCamelCase__ :      str								=		max_length

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      Tuple								=		model_class(__magic_name__				)

									lowerCamelCase__ :      int								=		model.generate(__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      int								=		jit(model.generate				)
									lowerCamelCase__ :      List[str]								=		jit_generate(__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      Union[str, Any]								=		self._get_input_ids_and_config()
						lowerCamelCase__ :      List[str]								=		True
						lowerCamelCase__ :      Dict								=		max_length

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      str								=		model_class(__magic_name__				)

									lowerCamelCase__ :      Dict								=		model.generate(__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      Tuple								=		jit(model.generate				)
									lowerCamelCase__ :      List[str]								=		jit_generate(__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      Dict								=		self._get_input_ids_and_config()
						lowerCamelCase__ :      Union[str, Any]								=		False
						lowerCamelCase__ :      Dict								=		max_length
						lowerCamelCase__ :      str								=		2

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      Optional[int]								=		model_class(__magic_name__				)

									lowerCamelCase__ :      Union[str, Any]								=		model.generate(__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      Dict								=		jit(model.generate				)
									lowerCamelCase__ :      Optional[int]								=		jit_generate(__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      int								=		self._get_input_ids_and_config()
						lowerCamelCase__ :      Dict								=		False
						lowerCamelCase__ :      List[Any]								=		max_length
						lowerCamelCase__ :      List[Any]								=		2
						lowerCamelCase__ :      List[str]								=		2

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      Optional[Any]								=		model_class(__magic_name__				)

									lowerCamelCase__ :      Union[str, Any]								=		model.generate(__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[0]							,    input_ids.shape[0] * config.num_return_sequences				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      Tuple								=		self._get_input_ids_and_config()
						lowerCamelCase__ :      Optional[int]								=		True
						lowerCamelCase__ :      str								=		max_length
						lowerCamelCase__ :      Optional[int]								=		0.8
						lowerCamelCase__ :      List[Any]								=		10
						lowerCamelCase__ :      int								=		0.3
						lowerCamelCase__ :      Union[str, Any]								=		1
						lowerCamelCase__ :      int								=		8
						lowerCamelCase__ :      Tuple								=		9

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      Optional[Any]								=		model_class(__magic_name__				)

									lowerCamelCase__ :      Any								=		model.generate(__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      Optional[int]								=		jit(model.generate				)
									lowerCamelCase__ :      List[Any]								=		jit_generate(__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      List[str]								=		self._get_input_ids_and_config()
						lowerCamelCase__ :      str								=		max_length
						lowerCamelCase__ :      List[str]								=		1
						lowerCamelCase__ :      Optional[int]								=		8
						lowerCamelCase__ :      Union[str, Any]								=		9

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      Dict								=		model_class(__magic_name__				)

									lowerCamelCase__ :      Tuple								=		model.generate(__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      Optional[int]								=		jit(model.generate				)
									lowerCamelCase__ :      Dict								=		jit_generate(__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      int								=		self._get_input_ids_and_config()
						lowerCamelCase__ :      Tuple								=		max_length
						lowerCamelCase__ :      Optional[int]								=		2
						lowerCamelCase__ :      Union[str, Any]								=		1
						lowerCamelCase__ :      Tuple								=		8
						lowerCamelCase__ :      Tuple								=		9

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      Optional[int]								=		model_class(__magic_name__				)

									lowerCamelCase__ :      List[Any]								=		model.generate(__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      List[Any]								=		jit(model.generate				)
									lowerCamelCase__ :      Union[str, Any]								=		jit_generate(__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      Tuple								=		self._get_input_ids_and_config()

						# pad attention mask on the left
						lowerCamelCase__ :      Any								=		attention_mask.at[(0, 0)].set(0				)

						lowerCamelCase__ :      List[str]								=		False
						lowerCamelCase__ :      str								=		max_length

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      Dict								=		model_class(__magic_name__				)

									lowerCamelCase__ :      List[str]								=		model.generate(__magic_name__							,    attention_mask=__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      Tuple								=		jit(model.generate				)
									lowerCamelCase__ :      Any								=		jit_generate(__magic_name__							,    attention_mask=__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      Optional[Any]								=		self._get_input_ids_and_config()

						# pad attention mask on the left
						lowerCamelCase__ :      Optional[Any]								=		attention_mask.at[(0, 0)].set(0				)

						lowerCamelCase__ :      int								=		True
						lowerCamelCase__ :      str								=		max_length

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      int								=		model_class(__magic_name__				)

									lowerCamelCase__ :      str								=		model.generate(__magic_name__							,    attention_mask=__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      List[Any]								=		jit(model.generate				)
									lowerCamelCase__ :      List[Any]								=		jit_generate(__magic_name__							,    attention_mask=__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)





			def        _snake_case    (self				):
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      Optional[Any]								=		self._get_input_ids_and_config()

						# pad attention mask on the left
						lowerCamelCase__ :      Union[str, Any]								=		attention_mask.at[(0, 0)].set(0				)

						lowerCamelCase__ :      Tuple								=		2
						lowerCamelCase__ :      List[Any]								=		max_length

						for model_class in self.all_generative_model_classes:
									lowerCamelCase__ :      Optional[Any]								=		model_class(__magic_name__				)

									lowerCamelCase__ :      Union[str, Any]								=		model.generate(__magic_name__							,    attention_mask=__magic_name__				).sequences
									self.assertEqual(generation_outputs.shape[-1]							,    __magic_name__				)

									lowerCamelCase__ :      int								=		jit(model.generate				)
									lowerCamelCase__ :      Any								=		jit_generate(__magic_name__							,    attention_mask=__magic_name__				).sequences

									self.assertListEqual(generation_outputs.tolist()							,    jit_generation_outputs.tolist()				)







@require_flax
class 				__A					( unittest.TestCase      ):





			def        _snake_case    (self				):
						lowerCamelCase__ :      Dict								=		AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-bert"""				)
						lowerCamelCase__ :      Tuple								=		FlaxAutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-bert-flax-only"""				)

						lowerCamelCase__ :      Dict								=		"""Hello world"""
						lowerCamelCase__ :      Union[str, Any]								=		tokenizer(__magic_name__							,    return_tensors="""np"""				).input_ids

						# typos are quickly detected (the correct argument is `do_sample`)
						with self.assertRaisesRegex(__magic_name__							,    """do_samples"""				):
									model.generate(__magic_name__							,    do_samples=__magic_name__				)

						# arbitrary arguments that will not be used anywhere are also not accepted
						with self.assertRaisesRegex(__magic_name__							,    """foo"""				):
									lowerCamelCase__ :      Tuple								=		{"""foo""": """bar"""}
									model.generate(__magic_name__							,    **__magic_name__				)



