


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

class 							lowercase  :







     def __init__(						self :						str   ,   _lowercase :						str = None   ,   _lowercase :						uuid.UUID = None   ,   _lowercase :						Any=None   ,   _lowercase :						Optional[int]=None      ):
           if not conversation_id:
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		uuid.uuida()
           if past_user_inputs is None:
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		[]
           if generated_responses is None:
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		[]

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







     def __eq__(						self :						Optional[int]   ,   _lowercase :						Dict      ):
           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 							lowercase__       (						self :						Dict   ,   _lowercase :						str   ,   _lowercase :						bool = 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}\"."""      )
                       SCREAMING_SNAKE_CASE__       :  int						     =		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:
                 SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		text







     def 							lowercase__       (						self :						int      ):
           if self.new_user_input:
                 self.past_user_inputs.append(self.new_user_input      )
           SCREAMING_SNAKE_CASE__       :  int						     =		None







     def 							lowercase__       (						self :						List[Any]   ,   _lowercase :						str      ):
           self.generated_responses.append(_lowercase      )







     def 							lowercase__       (						self :						Union[str, Any]      ):
           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 :						str      ):
           SCREAMING_SNAKE_CASE__       :  Dict						     =		f"""Conversation id: {self.uuid} \n"""
           for is_user, text in self.iter_texts():
                 SCREAMING_SNAKE_CASE__       :  List[str]						     =		'''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			):







     def __init__(						self :						Dict   ,   *_lowercase :						List[str]   ,   **_lowercase :						List[Any]      ):
           super().__init__(*_lowercase   ,   **_lowercase      )
           if self.tokenizer.pad_token_id is None:
                 SCREAMING_SNAKE_CASE__       :  Tuple						     =		self.tokenizer.eos_token







     def 							lowercase__       (						self :						str   ,   _lowercase :						List[Any]=None   ,   _lowercase :						Tuple=None   ,   _lowercase :						Tuple=None   ,   **_lowercase :						List[Any]      ):
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		{}
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		{}
           SCREAMING_SNAKE_CASE__       :  int						     =		{}

           if min_length_for_response is not None:
                 SCREAMING_SNAKE_CASE__       :  str						     =		min_length_for_response
           if minimum_tokens is not None:
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		minimum_tokens

           if "max_length" in generate_kwargs:
                 SCREAMING_SNAKE_CASE__       :  Optional[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:
                 SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		clean_up_tokenization_spaces

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







     def __call__(						self :						List[str]   ,   _lowercase :						Union[Conversation, List[Conversation]]   ,   _lowercase :						Optional[Any]=0   ,   **_lowercase :						Optional[int]      ):
           SCREAMING_SNAKE_CASE__       :  Any						     =		super().__call__(_lowercase   ,   num_workers=_lowercase   ,   **_lowercase      )
           if isinstance(_lowercase   ,   _lowercase      ) and len(_lowercase      ) == 1:
                 return outputs[0]
           return outputs







     def 							lowercase__       (						self :						int   ,   _lowercase :						Conversation   ,   _lowercase :						str=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'''      ):
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		self.tokenizer._build_conversation_input_ids(_lowercase      )
           else:
                 # If the tokenizer cannot handle conversations, we default to only the old version
                 SCREAMING_SNAKE_CASE__       :  Any						     =		self._legacy_parse_and_tokenize(_lowercase      )

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







     def 							lowercase__       (						self :						str   ,   _lowercase :						Optional[Any]   ,   _lowercase :						Tuple=10   ,   **_lowercase :						Optional[Any]      ):
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		generate_kwargs.get('''max_length'''   ,   self.model.config.max_length      )

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







     def 							lowercase__       (						self :						Optional[Any]   ,   _lowercase :						str   ,   _lowercase :						int=True      ):
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		model_outputs['''output_ids''']
           SCREAMING_SNAKE_CASE__       :  str						     =		self.tokenizer.decode(
               output_ids[0]   ,   skip_special_tokens=_lowercase   ,   clean_up_tokenization_spaces=_lowercase   ,   )
           SCREAMING_SNAKE_CASE__       :  str						     =		model_outputs['''conversation''']
           conversation.mark_processed()
           conversation.append_response(_lowercase      )
           return conversation







     def 							lowercase__       (						self :						Union[str, Any]   ,   _lowercase :						Conversation      ):
           SCREAMING_SNAKE_CASE__       :  str						     =		self.tokenizer.eos_token_id
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		[]
           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:
                 SCREAMING_SNAKE_CASE__       :  Tuple						     =		input_ids[-self.tokenizer.model_max_length :]
           return input_ids







"""simple docstring"""






from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Features, Sequence, Value
from .base import TaskTemplate





@dataclass(frozen=lowerCAmelCase      )
class 					_A		(			lowerCAmelCase      ):
      # `task` is not a ClassVar since we want it to be part of the `asdict` output for JSON serialization
      snake_case__     :  str				    =      field(default='question-answering-extractive'    ,       metadata={'include_in_asdict_even_if_is_default': True}      )
      snake_case__     :  ClassVar[Features]				    =      Features({'question': Value('string'      ), 'context': Value('string'      )}      )
      snake_case__     :  ClassVar[Features]				    =      Features(
          {
              'answers': Sequence(
                  {
                      'text': Value('string'      ),
                      'answer_start': Value('int32'      ),
                  }      )
          }      )
      snake_case__     :  str				    =      "question"
      snake_case__     :  str				    =      "context"
      snake_case__     :  str				    =      "answers"

      @property
      def 							A__			(		self		):







            """simple docstring"""



            return {self.question_column: "question", self.context_column: "context", self.answers_column: "answers"}



