



"""simple docstring"""

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


a_			   =					logging.get_logger(__name__)







class    __lowercase     :







						"""simple docstring"""







						def __init__(self   ,		lowercase__ = None   ,		lowercase__ = None   ,		lowercase__=None   ,		lowercase__=None       ):
											if not conversation_id:
																snake_case_					:		int 					=					uuid.uuida()
											if past_user_inputs is None:
																snake_case_					:		int 					=					[]
											if generated_responses is None:
																snake_case_					:		List[Any] 					=					[]

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







						def __eq__(self   ,		lowercase__       ):
											if not isinstance(lowercase__   ,		lowercase__       ):
																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 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = 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}".'       )
																					snake_case_					:		Any 					=					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:
																snake_case_					:		List[Any] 					=					text







						def 						__UpperCamelCase							(self       ):
											if self.new_user_input:
																self.past_user_inputs.append(self.new_user_input       )
											snake_case_					:		Dict 					=					None







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											self.generated_responses.append(lowercase__       )







						def 						__UpperCamelCase							(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       ):
											snake_case_					:		List[str] 					=					f'Conversation id: {self.uuid} \n'
											for is_user, text in self.iter_texts():
																snake_case_					:		Optional[Any] 					=					"""user""" if is_user else """bot"""
																output += f'{name} >> {text} \n'
											return output







@add_end_docstrings(
    _UpperCAmelCase					,    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    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







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







						def 						__UpperCamelCase							(self   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		**lowercase__       ):
											snake_case_					:		Tuple 					=					{}
											snake_case_					:		Optional[Any] 					=					{}
											snake_case_					:		Optional[Any] 					=					{}

											if min_length_for_response is not None:
																snake_case_					:		Dict 					=					min_length_for_response
											if minimum_tokens is not None:
																snake_case_					:		Union[str, Any] 					=					minimum_tokens

											if "max_length" in generate_kwargs:
																snake_case_					:		List[str] 					=					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:
																snake_case_					:		Union[str, Any] 					=					clean_up_tokenization_spaces

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







						def __call__(self   ,		lowercase__   ,		lowercase__=0   ,		**lowercase__       ):
											snake_case_					:		List[str] 					=					super().__call__(lowercase__   ,		num_workers=lowercase__   ,		**lowercase__       )
											if isinstance(lowercase__   ,		lowercase__       ) and len(lowercase__       ) == 1:
																return outputs[0]
											return outputs







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=32       ):
											if not isinstance(lowercase__   ,		lowercase__       ):
																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"""       ):
																snake_case_					:		str 					=					self.tokenizer._build_conversation_input_ids(lowercase__       )
											else:
																# If the tokenizer cannot handle conversations, we default to only the old version
																snake_case_					:		Any 					=					self._legacy_parse_and_tokenize(lowercase__       )

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







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=10   ,		**lowercase__       ):
											snake_case_					:		Optional[int] 					=					generate_kwargs.get("""max_length"""   ,		self.model.config.max_length       )

											snake_case_					:		str 					=					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})'       )
																snake_case_					:		int 					=					max_length - minimum_tokens
																snake_case_					:		List[Any] 					=					model_inputs["""input_ids"""][:, -trim:]
																if "attention_mask" in model_inputs:
																					snake_case_					:		Union[str, Any] 					=					model_inputs["""attention_mask"""][:, -trim:]
											snake_case_					:		int 					=					model_inputs.pop("""conversation"""       )
											snake_case_					:		Dict 					=					max_length
											snake_case_					:		Tuple 					=					self.model.generate(**lowercase__   ,		**lowercase__       )
											if self.model.config.is_encoder_decoder:
																snake_case_					:		str 					=					1
											else:
																snake_case_					:		List[Any] 					=					n
											return {"output_ids": output_ids[:, start_position:], "conversation": conversation}







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=True       ):
											snake_case_					:		int 					=					model_outputs["""output_ids"""]
											snake_case_					:		List[Any] 					=					self.tokenizer.decode(
											    output_ids[0]   ,		skip_special_tokens=lowercase__   ,		clean_up_tokenization_spaces=lowercase__   ,		)
											snake_case_					:		Optional[Any] 					=					model_outputs["""conversation"""]
											conversation.mark_processed()
											conversation.append_response(lowercase__       )
											return conversation







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		Dict 					=					self.tokenizer.eos_token_id
											snake_case_					:		Optional[int] 					=					[]
											for is_user, text in conversation.iter_texts():
																if eos_token_id is not None:
																					input_ids.extend(self.tokenizer.encode(lowercase__   ,		add_special_tokens=lowercase__       ) + [eos_token_id]       )
																else:
																					input_ids.extend(self.tokenizer.encode(lowercase__   ,		add_special_tokens=lowercase__       )       )

											if len(lowercase__       ) > self.tokenizer.model_max_length:
																snake_case_					:		int 					=					input_ids[-self.tokenizer.model_max_length :]
											return input_ids









"""simple docstring"""

def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  int				,    SCREAMING_SNAKE_CASE__ :  int			):



					"""simple docstring"""





					return numa ^ numa < 0


if __name__ == "__main__":
			import doctest

			doctest.testmod()




