






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


__UpperCamelCase   :   Dict      =    logging.get_logger(__name__)


class 							__UpperCamelCase     :







	def __init__(	self   :				List[str]							,					_lowerCAmelCase   :				str = None							,					_lowerCAmelCase   :				uuid.UUID = None							,					_lowerCAmelCase   :				Optional[int]=None							,					_lowerCAmelCase   :				Optional[Any]=None      )		->			List[str]:
		"""simple docstring"""







		if not conversation_id:
			__lowercase   							=	uuid.uuida()
		if past_user_inputs is None:
			__lowercase   							=	[]
		if generated_responses is None:
			__lowercase   							=	[]

		__lowercase   							=	conversation_id
		__lowercase   							=	past_user_inputs
		__lowercase   							=	generated_responses
		__lowercase   							=	text







	def __eq__(	self   :				List[str]							,					_lowerCAmelCase   :				str      )		->			List[str]:
		"""simple docstring"""







		if not isinstance(_lowerCAmelCase							,					_lowerCAmelCase      ):
			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 			_a (	self   :				Dict							,					_lowerCAmelCase   :				str							,					_lowerCAmelCase   :				bool = False      )		->			Tuple:
		"""simple docstring"""







		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}".'      )
				__lowercase   							=	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:
			__lowercase   							=	text







	def 			_a (	self   :				Tuple      )		->			str:
		"""simple docstring"""







		if self.new_user_input:
			self.past_user_inputs.append(self.new_user_input      )
		__lowercase   							=	None







	def 			_a (	self   :				Dict							,					_lowerCAmelCase   :				str      )		->			List[Any]:
		"""simple docstring"""







		self.generated_responses.append(_lowerCAmelCase      )







	def 			_a (	self   :				Optional[int]      )		->			Any:
		"""simple docstring"""







		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   :				Dict      )		->			Any:
		"""simple docstring"""







		__lowercase   							=	F'Conversation id: {self.uuid} \n'
		for is_user, text in self.iter_texts():
			__lowercase   							=	"""user""" if is_user else """bot"""
			output += F'{name} >> {text} \n'
		return output


@add_end_docstrings(
    _lowerCAmelCase ,      R'\n        min_length_for_response (`int`, *optional*, defaults to 32):\n            The minimum length (in number of tokens) for a response.\n        minimum_tokens (`int`, *optional*, defaults to 10):\n            The minimum length of tokens to leave for a response.\n    ' ,      )
class 							__UpperCamelCase     (  _lowerCAmelCase ):







	def __init__(	self   :				List[Any]							,					*_lowerCAmelCase   :				Tuple							,					**_lowerCAmelCase   :				Optional[Any]      )		->			str:
		"""simple docstring"""







		super().__init__(*_lowerCAmelCase							,					**_lowerCAmelCase      )
		if self.tokenizer.pad_token_id is None:
			__lowercase   							=	self.tokenizer.eos_token







	def 			_a (	self   :				Tuple							,					_lowerCAmelCase   :				Dict=None							,					_lowerCAmelCase   :				Tuple=None							,					_lowerCAmelCase   :				List[Any]=None							,					**_lowerCAmelCase   :				Dict      )		->			str:
		"""simple docstring"""







		__lowercase   							=	{}
		__lowercase   							=	{}
		__lowercase   							=	{}

		if min_length_for_response is not None:
			__lowercase   							=	min_length_for_response
		if minimum_tokens is not None:
			__lowercase   							=	minimum_tokens

		if "max_length" in generate_kwargs:
			__lowercase   							=	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:
			__lowercase   							=	clean_up_tokenization_spaces

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







	def __call__(	self   :				Tuple							,					_lowerCAmelCase   :				Union[Conversation, List[Conversation]]							,					_lowerCAmelCase   :				str=0							,					**_lowerCAmelCase   :				Dict      )		->			Optional[int]:
		"""simple docstring"""







		__lowercase   							=	super().__call__(_lowerCAmelCase							,					num_workers=_lowerCAmelCase							,					**_lowerCAmelCase      )
		if isinstance(_lowerCAmelCase							,					_lowerCAmelCase      ) and len(_lowerCAmelCase      ) == 1:
			return outputs[0]
		return outputs







	def 			_a (	self   :				List[str]							,					_lowerCAmelCase   :				Conversation							,					_lowerCAmelCase   :				Optional[int]=32      )		->			Dict[str, Any]:
		"""simple docstring"""







		if not isinstance(_lowerCAmelCase							,					_lowerCAmelCase      ):
			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"""      ):
			__lowercase   							=	self.tokenizer._build_conversation_input_ids(_lowerCAmelCase      )
		else:
			# If the tokenizer cannot handle conversations, we default to only the old version
			__lowercase   							=	self._legacy_parse_and_tokenize(_lowerCAmelCase      )

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







	def 			_a (	self   :				str							,					_lowerCAmelCase   :				Tuple							,					_lowerCAmelCase   :				Union[str, Any]=10							,					**_lowerCAmelCase   :				int      )		->			Dict:
		"""simple docstring"""







		__lowercase   							=	generate_kwargs.get("""max_length"""							,					self.model.config.max_length      )

		__lowercase   							=	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})'      )
			__lowercase   							=	max_length - minimum_tokens
			__lowercase   							=	model_inputs["""input_ids"""][:, -trim:]
			if "attention_mask" in model_inputs:
				__lowercase   							=	model_inputs["""attention_mask"""][:, -trim:]
		__lowercase   							=	model_inputs.pop("""conversation"""      )
		__lowercase   							=	max_length
		__lowercase   							=	self.model.generate(**_lowerCAmelCase							,					**_lowerCAmelCase      )
		if self.model.config.is_encoder_decoder:
			__lowercase   							=	1
		else:
			__lowercase   							=	n
		return {"output_ids": output_ids[:, start_position:], "conversation": conversation}







	def 			_a (	self   :				Optional[Any]							,					_lowerCAmelCase   :				Any							,					_lowerCAmelCase   :				Optional[int]=True      )		->			List[Any]:
		"""simple docstring"""







		__lowercase   							=	model_outputs["""output_ids"""]
		__lowercase   							=	self.tokenizer.decode(
		    output_ids[0]							,					skip_special_tokens=_lowerCAmelCase							,					clean_up_tokenization_spaces=_lowerCAmelCase							,					)
		__lowercase   							=	model_outputs["""conversation"""]
		conversation.mark_processed()
		conversation.append_response(_lowerCAmelCase      )
		return conversation







	def 			_a (	self   :				Optional[int]							,					_lowerCAmelCase   :				Conversation      )		->			Dict:
		"""simple docstring"""







		__lowercase   							=	self.tokenizer.eos_token_id
		__lowercase   							=	[]
		for is_user, text in conversation.iter_texts():
			if eos_token_id is not None:
				input_ids.extend(self.tokenizer.encode(_lowerCAmelCase							,					add_special_tokens=_lowerCAmelCase      ) + [eos_token_id]      )
			else:
				input_ids.extend(self.tokenizer.encode(_lowerCAmelCase							,					add_special_tokens=_lowerCAmelCase      )      )

		if len(_lowerCAmelCase      ) > self.tokenizer.model_max_length:
			__lowercase   							=	input_ids[-self.tokenizer.model_max_length :]
		return input_ids






"""simple docstring"""







from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


a_							=    logging.get_logger(__name__)


a_							=    TypeVar("""DatasetType""", Dataset, IterableDataset)

def     UpperCAmelCase_					(				__a   :			List[DatasetType]      ,							__a   :			Optional[List[float]] = None      ,							__a   :			Optional[int] = None      ,							__a   :			Optional[DatasetInfo] = None      ,							__a   :			Optional[NamedSplit] = None      ,							__a   :			Literal["first_exhausted", "all_exhausted"] = "first_exhausted"      ,							):




			'''simple docstring'''



			from .arrow_dataset import Dataset
			from .iterable_dataset import IterableDataset

			if not datasets:
						raise ValueError('Unable to interleave an empty list of datasets.'						)
			for i, dataset in enumerate(__a						):
						if not isinstance(__a      ,							(Dataset, IterableDataset)						):
									if isinstance(__a      ,							(DatasetDict, IterableDatasetDict)						):
												if not dataset:
															raise ValueError(
															    f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} "
															    'is an empty dataset dictionary.'						)
												raise ValueError(
												    f"Dataset at position {i} has at least one split: {list(__a						)}\n"
												    f"Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__a						)						)}']"						)
									raise ValueError(
									    f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__a						).__name__}."						)
						if i == 0:
									_lowerCamelCase       ,							_lowerCamelCase					:					Tuple	       =    (
									    (Dataset, IterableDataset) if isinstance(__a      ,							__a						) else (IterableDataset, Dataset)
									)
						elif not isinstance(__a      ,							__a						):
									raise ValueError(
									    f"Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."						)
			if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
						raise ValueError(f"{stopping_strategy} is not supported. Please enter a valid stopping_strategy."						)
			if dataset_type is Dataset:
						return _interleave_map_style_datasets(
						    __a      ,							__a      ,							__a      ,							info=__a      ,							split=__a      ,							stopping_strategy=__a						)
			else:
						return _interleave_iterable_datasets(
						    __a      ,							__a      ,							__a      ,							info=__a      ,							split=__a      ,							stopping_strategy=__a						)



def     UpperCAmelCase_					(				__a   :			List[DatasetType]      ,							__a   :			Optional[DatasetInfo] = None      ,							__a   :			Optional[NamedSplit] = None      ,							__a   :			int = 0      ,							):




			'''simple docstring'''



			if not dsets:
						raise ValueError('Unable to concatenate an empty list of datasets.'						)
			for i, dataset in enumerate(__a						):
						if not isinstance(__a      ,							(Dataset, IterableDataset)						):
									if isinstance(__a      ,							(DatasetDict, IterableDatasetDict)						):
												if not dataset:
															raise ValueError(
															    f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} "
															    'is an empty dataset dictionary.'						)
												raise ValueError(
												    f"Dataset at position {i} has at least one split: {list(__a						)}\n"
												    f"Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__a						)						)}']"						)
									raise ValueError(
									    f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__a						).__name__}."						)
						if i == 0:
									_lowerCamelCase       ,							_lowerCamelCase					:					Optional[Any]	       =    (
									    (Dataset, IterableDataset) if isinstance(__a      ,							__a						) else (IterableDataset, Dataset)
									)
						elif not isinstance(__a      ,							__a						):
									raise ValueError(
									    f"Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."						)
			if dataset_type is Dataset:
						return _concatenate_map_style_datasets(__a      ,							info=__a      ,							split=__a      ,							axis=__a						)
			else:
						return _concatenate_iterable_datasets(__a      ,							info=__a      ,							split=__a      ,							axis=__a						)





