





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_      							=     logging.get_logger(__name__)
class 			UpperCamelCase_		:




     def __init__(		self				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = None				,						lowerCAmelCase_=None				,						lowerCAmelCase_=None  )		->							List[Any]:






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

          _snake_case									=				conversation_id
          _snake_case									=				past_user_inputs
          _snake_case									=				generated_responses
          _snake_case									=				text




     def __eq__(		self				,						lowerCAmelCase_  )		->							Union[str, Any]:






          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 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_ = False  )		->							Union[str, Any]:






          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									=				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									=				text




     def 	lowerCAmelCase			(		self  )		->							List[Any]:






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




     def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							List[Any]:
          self.generated_responses.append(lowerCamelCase__  )




     def 	lowerCAmelCase			(		self  )		->							str:






          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  )		->							int:
          _snake_case									=				F'''Conversation id: {self.uuid} \n'''
          for is_user, text in self.iter_texts():
               _snake_case									=				'''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				,						*lowerCAmelCase_				,						**lowerCAmelCase_  )		->							int:
          super().__init__(*lowerCamelCase__				,						**lowerCamelCase__  )
          if self.tokenizer.pad_token_id is None:
               _snake_case									=				self.tokenizer.eos_token




     def 	lowerCAmelCase			(		self				,						lowerCAmelCase_=None				,						lowerCAmelCase_=None				,						lowerCAmelCase_=None				,						**lowerCAmelCase_  )		->							Optional[Any]:
          _snake_case									=				{}
          _snake_case									=				{}
          _snake_case									=				{}

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

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

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




     def __call__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_=0				,						**lowerCAmelCase_  )		->							str:
          _snake_case									=				super().__call__(lowerCamelCase__				,						num_workers=lowerCamelCase__				,						**lowerCamelCase__  )
          if isinstance(lowerCamelCase__				,						lowerCamelCase__  ) and len(lowerCamelCase__  ) == 1:
               return outputs[0]
          return outputs




     def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_=32  )		->							Dict[str, Any]:






          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'  ):
               _snake_case									=				self.tokenizer._build_conversation_input_ids(lowerCamelCase__  )
          else:
               # If the tokenizer cannot handle conversations, we default to only the old version
               _snake_case									=				self._legacy_parse_and_tokenize(lowerCamelCase__  )

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




     def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_=10				,						**lowerCAmelCase_  )		->							List[Any]:
          _snake_case									=				generate_kwargs.get('max_length'				,						self.model.config.max_length  )

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




     def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_=True  )		->							Optional[Any]:
          _snake_case									=				model_outputs['''output_ids''']
          _snake_case									=				self.tokenizer.decode(
              output_ids[0]				,						skip_special_tokens=lowerCamelCase__				,						clean_up_tokenization_spaces=lowerCamelCase__				,						)
          _snake_case									=				model_outputs['''conversation''']
          conversation.mark_processed()
          conversation.append_response(lowerCamelCase__  )
          return conversation






     def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							Dict:
          _snake_case									=				self.tokenizer.eos_token_id
          _snake_case									=				[]
          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:
               _snake_case									=				input_ids[-self.tokenizer.model_max_length :]
          return input_ids










from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


UpperCAmelCase_      							=     {
    """configuration_bigbird_pegasus""": [
        """BIGBIRD_PEGASUS_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """BigBirdPegasusConfig""",
        """BigBirdPegasusOnnxConfig""",
    ],
}

try:
					if not is_torch_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					UpperCAmelCase_      							=     [
					    """BIGBIRD_PEGASUS_PRETRAINED_MODEL_ARCHIVE_LIST""",
					    """BigBirdPegasusForCausalLM""",
					    """BigBirdPegasusForConditionalGeneration""",
					    """BigBirdPegasusForQuestionAnswering""",
					    """BigBirdPegasusForSequenceClassification""",
					    """BigBirdPegasusModel""",
					    """BigBirdPegasusPreTrainedModel""",
					]


if TYPE_CHECKING:
					from .configuration_bigbird_pegasus import (
					    BIGBIRD_PEGASUS_PRETRAINED_CONFIG_ARCHIVE_MAP,
					    BigBirdPegasusConfig,
					    BigBirdPegasusOnnxConfig,
					)

					try:
										if not is_torch_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_bigbird_pegasus import (
										    BIGBIRD_PEGASUS_PRETRAINED_MODEL_ARCHIVE_LIST,
										    BigBirdPegasusForCausalLM,
										    BigBirdPegasusForConditionalGeneration,
										    BigBirdPegasusForQuestionAnswering,
										    BigBirdPegasusForSequenceClassification,
										    BigBirdPegasusModel,
										    BigBirdPegasusPreTrainedModel,
										)


else:
					import sys

					UpperCAmelCase_      							=     _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)



