






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


SCREAMING_SNAKE_CASE 							=						logging.get_logger(__name__)
class 							A_    :






 '''simple docstring'''




 def __init__(	self							,  _A = None							,  _A = None							,  _A=None							,  _A=None) ->		int:

     """simple docstring"""



     if not conversation_id:
         _UpperCAmelCase				:      Optional[int]			  =				uuid.uuida()
     if past_user_inputs is None:
         _UpperCAmelCase				:      str			  =				[]
     if generated_responses is None:
         _UpperCAmelCase				:      Optional[int]			  =				[]

     _UpperCAmelCase				:      int			  =				conversation_id
     _UpperCAmelCase				:      Any			  =				past_user_inputs
     _UpperCAmelCase				:      Union[str, Any]			  =				generated_responses
     _UpperCAmelCase				:      Union[str, Any]			  =				text


 def __eq__(	self							,  _A) ->		List[str]:

     """simple docstring"""



     if not isinstance(_SCREAMING_SNAKE_CASE							,  _SCREAMING_SNAKE_CASE):
         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							,  _A							,  _A = False) ->		str:

     """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				:      Union[str, 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:
         _UpperCAmelCase				:      Any			  =				text


 def  snake_case__				(	self) ->		Any:

     """simple docstring"""



     if self.new_user_input:
         self.past_user_inputs.append(self.new_user_input)
     _UpperCAmelCase				:      Dict			  =				None


 def  snake_case__				(	self							,  _A) ->		Any:

     """simple docstring"""



     self.generated_responses.append(_SCREAMING_SNAKE_CASE)


 def  snake_case__				(	self) ->		Dict:

     """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:

     """simple docstring"""



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



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






 '''simple docstring'''




 def __init__(	self							,  *_A							,  **_A) ->		Optional[Any]:

     """simple docstring"""



     super().__init__(*_SCREAMING_SNAKE_CASE							,  **_SCREAMING_SNAKE_CASE)
     if self.tokenizer.pad_token_id is None:
         _UpperCAmelCase				:      List[Any]			  =				self.tokenizer.eos_token


 def  snake_case__				(	self							,  _A=None							,  _A=None							,  _A=None							,  **_A) ->		Union[str, Any]:

     """simple docstring"""



     _UpperCAmelCase				:      str			  =				{}
     _UpperCAmelCase				:      Dict			  =				{}
     _UpperCAmelCase				:      Tuple			  =				{}

     if min_length_for_response is not None:
         _UpperCAmelCase				:      Optional[Any]			  =				min_length_for_response
     if minimum_tokens is not None:
         _UpperCAmelCase				:      Optional[int]			  =				minimum_tokens

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

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


 def __call__(	self							,  _A							,  _A=0							,  **_A) ->		Any:

     """simple docstring"""



     _UpperCAmelCase				:      List[str]			  =				super().__call__(_SCREAMING_SNAKE_CASE							,  num_workers=_SCREAMING_SNAKE_CASE							,  **_SCREAMING_SNAKE_CASE)
     if isinstance(_SCREAMING_SNAKE_CASE							,  _SCREAMING_SNAKE_CASE) and len(_SCREAMING_SNAKE_CASE) == 1:
         return outputs[0]
     return outputs


 def  snake_case__				(	self							,  _A							,  _A=32) ->		Dict[str, Any]:

     """simple docstring"""



     if not isinstance(_SCREAMING_SNAKE_CASE							,  _SCREAMING_SNAKE_CASE):
         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				:      str			  =				self.tokenizer._build_conversation_input_ids(_SCREAMING_SNAKE_CASE)
     else:
         # If the tokenizer cannot handle conversations, we default to only the old version
         _UpperCAmelCase				:      List[Any]			  =				self._legacy_parse_and_tokenize(_SCREAMING_SNAKE_CASE)

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


 def  snake_case__				(	self							,  _A							,  _A=10							,  **_A) ->		Union[str, Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Any			  =				generate_kwargs.get('''max_length'''							,  self.model.config.max_length)

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


 def  snake_case__				(	self							,  _A							,  _A=True) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Any			  =				model_outputs['''output_ids''']
     _UpperCAmelCase				:      List[Any]			  =				self.tokenizer.decode(
         output_ids[0]							,  skip_special_tokens=_SCREAMING_SNAKE_CASE							,  clean_up_tokenization_spaces=_SCREAMING_SNAKE_CASE							,  )
     _UpperCAmelCase				:      List[Any]			  =				model_outputs['''conversation''']
     conversation.mark_processed()
     conversation.append_response(_SCREAMING_SNAKE_CASE)
     return conversation


 def  snake_case__				(	self							,  _A) ->		Dict:

     """simple docstring"""



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

     if len(_SCREAMING_SNAKE_CASE) > self.tokenizer.model_max_length:
         _UpperCAmelCase				:      List[str]			  =				input_ids[-self.tokenizer.model_max_length :]
     return input_ids












import unittest

from transformers import load_tool

from .test_tools_common import ToolTesterMixin


SCREAMING_SNAKE_CASE 							=						'\nHugging Face was founded in 2016 by French entrepreneurs Clément Delangue, Julien Chaumond, and Thomas Wolf originally as a company that developed a chatbot app targeted at teenagers.[2] After open-sourcing the model behind the chatbot, the company pivoted to focus on being a platform for machine learning.\n\nIn March 2021, Hugging Face raised $40 million in a Series B funding round.[3]\n\nOn April 28, 2021, the company launched the BigScience Research Workshop in collaboration with several other research groups to release an open large language model.[4] In 2022, the workshop concluded with the announcement of BLOOM, a multilingual large language model with 176 billion parameters.[5]\n'
class 							A_    (						unittest.TestCase ,   __lowercase			):






 '''simple docstring'''




 def  snake_case__				(	self) ->		int:

     """simple docstring"""



     _UpperCAmelCase				:      int			  =				load_tool('''text-question-answering''')
     self.tool.setup()
     _UpperCAmelCase				:      Any			  =				load_tool('''text-question-answering'''							,  remote=_A)


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

     """simple docstring"""



     _UpperCAmelCase				:      Optional[Any]			  =				self.tool(_A							,  '''What did Hugging Face do in April 2021?''')
     self.assertEqual(_A							,  '''launched the BigScience Research Workshop''')


 def  snake_case__				(	self) ->		str:

     """simple docstring"""



     _UpperCAmelCase				:      str			  =				self.remote_tool(_A							,  '''What did Hugging Face do in April 2021?''')
     self.assertEqual(_A							,  '''launched the BigScience Research Workshop''')


 def  snake_case__				(	self) ->		Optional[int]:

     """simple docstring"""



     _UpperCAmelCase				:      Dict			  =				self.tool(text=_A							,  question='''What did Hugging Face do in April 2021?''')
     self.assertEqual(_A							,  '''launched the BigScience Research Workshop''')


 def  snake_case__				(	self) ->		Dict:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				self.remote_tool(text=_A							,  question='''What did Hugging Face do in April 2021?''')
     self.assertEqual(_A							,  '''launched the BigScience Research Workshop''')




