





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__   : str					  = logging.get_logger(__name__)







class   _UpperCamelCase  :







					'''simple docstring'''

					def __init__(						self    :  str ,					__lowercase    :  List[str] = None ,					__lowercase    :  Optional[int] = None ,					__lowercase    :  int=None ,					__lowercase    :  int=None		):






						'''simple docstring'''







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

						UpperCAmelCase_								=     conversation_id
						UpperCAmelCase_								=     past_user_inputs
						UpperCAmelCase_								=     generated_responses
						UpperCAmelCase_								=     text





					def __eq__(						self    :  str ,					__lowercase    :  Optional[int]		):






						'''simple docstring'''







						if not isinstance(A_ ,					A_		):
							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        SCREAMING_SNAKE_CASE							(						self    :  int ,					__lowercase    :  List[Any] ,					__lowercase    :  Optional[Any] = False		):






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





					def        SCREAMING_SNAKE_CASE							(						self    :  Tuple		):






						'''simple docstring'''







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





					def        SCREAMING_SNAKE_CASE							(						self    :  int ,					__lowercase    :  List[str]		):






						'''simple docstring'''







						self.generated_responses.append(A_		)





					def        SCREAMING_SNAKE_CASE							(						self    :  Union[str, 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    :  Tuple		):






						'''simple docstring'''







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







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







					'''simple docstring'''

					def __init__(						self    :  Tuple ,					*__lowercase    :  Tuple ,					**__lowercase    :  Optional[Any]		):






						'''simple docstring'''







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





					def        SCREAMING_SNAKE_CASE							(						self    :  int ,					__lowercase    :  Tuple=None ,					__lowercase    :  Optional[Any]=None ,					__lowercase    :  Dict=None ,					**__lowercase    :  Optional[int]		):






						'''simple docstring'''







						UpperCAmelCase_								=     {}
						UpperCAmelCase_								=     {}
						UpperCAmelCase_								=     {}

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

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

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





					def __call__(						self    :  List[Any] ,					__lowercase    :  int ,					__lowercase    :  str=0 ,					**__lowercase    :  Optional[int]		):






						'''simple docstring'''







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





					def        SCREAMING_SNAKE_CASE							(						self    :  Optional[int] ,					__lowercase    :  str ,					__lowercase    :  int=32		):






						'''simple docstring'''







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

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





					def        SCREAMING_SNAKE_CASE							(						self    :  Union[str, Any] ,					__lowercase    :  List[Any] ,					__lowercase    :  Optional[Any]=10 ,					**__lowercase    :  str		):






						'''simple docstring'''







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

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





					def        SCREAMING_SNAKE_CASE							(						self    :  Any ,					__lowercase    :  List[str] ,					__lowercase    :  Optional[int]=True		):






						'''simple docstring'''







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





					def        SCREAMING_SNAKE_CASE							(						self    :  Optional[Any] ,					__lowercase    :  List[str]		):






						'''simple docstring'''







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

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








import json
import os
import tempfile

import datasets
from utils import generate_example_dataset, get_duration


UpperCamelCase__   : Tuple					  = 5_00_00
UpperCamelCase__   : Any					  = 50_00

UpperCamelCase__  ,			UpperCamelCase__   : Tuple					  = os.path.split(__file__)
UpperCamelCase__   : Union[str, Any]					  = os.path.join(RESULTS_BASEPATH, """results""", RESULTS_FILENAME.replace(""".py""", """.json"""))




@get_duration
def    A_(			A				,      A   ):
	for i in range(A   ):
		UpperCAmelCase_								=     dataset[i]




@get_duration
def    A_(			A				,      A				,      A   ):
	for i in range(0				,      len(A   )				,      A   ):
		UpperCAmelCase_								=     dataset[i : i + batch_size]




@get_duration
def    A_(			A				,      A				,      A   ):
	with dataset.formatted_as(type=A   ):
		for i in range(A   ):
			UpperCAmelCase_								=     dataset[i]




@get_duration
def    A_(			A				,      A				,      A				,      A   ):
	with dataset.formatted_as(type=A   ):
		for i in range(0				,      A				,      A   ):
			UpperCAmelCase_								=     dataset[i : i + batch_size]




def    A_(			):
	UpperCAmelCase_								=     {"""num examples""": SPEED_TEST_N_EXAMPLES}
	UpperCAmelCase_								=     [
	    (read, {"""length""": SMALL_TEST}),
	    (read, {"""length""": SPEED_TEST_N_EXAMPLES}),
	    (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10}),
	    (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 100}),
	    (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 1000}),
	    (read_formatted, {"""type""": """numpy""", """length""": SMALL_TEST}),
	    (read_formatted, {"""type""": """pandas""", """length""": SMALL_TEST}),
	    (read_formatted, {"""type""": """torch""", """length""": SMALL_TEST}),
	    (read_formatted, {"""type""": """tensorflow""", """length""": SMALL_TEST}),
	    (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10}),
	    (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 1000}),
	]

	UpperCAmelCase_								=     [
	    (read, {"""length""": SMALL_TEST}),
	    (read, {"""length""": SPEED_TEST_N_EXAMPLES}),
	    (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10}),
	    (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 100}),
	    (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 1000}),
	    (read_formatted, {"""type""": """numpy""", """length""": SMALL_TEST}),
	    (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10}),
	    (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 1000}),
	]
	with tempfile.TemporaryDirectory() as tmp_dir:
		print("""generating dataset"""   )
		UpperCAmelCase_								=     datasets.Features(
		    {"""list""": datasets.Sequence(datasets.Value("""float32"""   )   ), """numbers""": datasets.Value("""float32"""   )}   )
		UpperCAmelCase_								=     generate_example_dataset(
		    os.path.join(A				,      """dataset.arrow"""   )				,      A				,      num_examples=A				,      seq_shapes={"""list""": (100,)}				,      )
		print("""first set of iterations"""   )
		for func, kwargs in functions:
			print(func.__name__				,      str(A   )   )
			UpperCAmelCase_								=     func(A				,      **A   )

		print("""shuffling dataset"""   )
		UpperCAmelCase_								=     dataset.shuffle()
		print("""Second set of iterations (after shuffling"""   )
		for func, kwargs in functions_shuffled:
			print("""shuffled """				,      func.__name__				,      str(A   )   )
			UpperCAmelCase_								=     func(
			    A				,      **A   )

	with open(A				,      """wb"""   ) as f:
		f.write(json.dumps(A   ).encode("""utf-8"""   )   )


if __name__ == "__main__":  # useful to run the profiler
				benchmark_iterating()


