



"""simple docstring"""







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


lowercase__		:    str				   =     logging.get_logger(__name__)







class 				UpperCamelCase__       :




      """simple docstring"""




      def __init__(				self    :				Tuple   ,      SCREAMING_SNAKE_CASE_    :				str = None   ,      SCREAMING_SNAKE_CASE_    :				uuid.UUID = None   ,      SCREAMING_SNAKE_CASE_    :				List[str]=None   ,      SCREAMING_SNAKE_CASE_    :				Union[str, Any]=None	):
        if not conversation_id:
          lowerCAmelCase_       :    int								=			uuid.uuida()
        if past_user_inputs is None:
          lowerCAmelCase_       :    Any								=			[]
        if generated_responses is None:
          lowerCAmelCase_       :    Any								=			[]

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



      def __eq__(				self    :				Optional[int]   ,      SCREAMING_SNAKE_CASE_    :				int	):
        if not isinstance(UpperCAmelCase_   ,      UpperCAmelCase_	):
          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    :				Any   ,      SCREAMING_SNAKE_CASE_    :				str   ,      SCREAMING_SNAKE_CASE_    :				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}\"."	)
            lowerCAmelCase_       :    Optional[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:
          lowerCAmelCase_       :    str								=			text



      def 					SCREAMING_SNAKE_CASE__			(				self    :				Dict	):
        if self.new_user_input:
          self.past_user_inputs.append(self.new_user_input	)
        lowerCAmelCase_       :    Any								=			None



      def 					SCREAMING_SNAKE_CASE__			(				self    :				Optional[int]   ,      SCREAMING_SNAKE_CASE_    :				str	):
        self.generated_responses.append(UpperCAmelCase_	)



      def 					SCREAMING_SNAKE_CASE__			(				self    :				List[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    :				List[str]	):
        lowerCAmelCase_       :    List[str]								=			F"Conversation id: {self.uuid} \n"
        for is_user, text in self.iter_texts():
          lowerCAmelCase_       :    Optional[Any]								=			"user" if is_user else "bot"
          output += F"{name} >> {text} \n"
        return output







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




      """simple docstring"""




      def __init__(				self    :				Optional[Any]   ,      *SCREAMING_SNAKE_CASE_    :				Union[str, Any]   ,      **SCREAMING_SNAKE_CASE_    :				List[Any]	):
        super().__init__(*UpperCAmelCase_   ,      **UpperCAmelCase_	)
        if self.tokenizer.pad_token_id is None:
          lowerCAmelCase_       :    str								=			self.tokenizer.eos_token



      def 					SCREAMING_SNAKE_CASE__			(				self    :				List[str]   ,      SCREAMING_SNAKE_CASE_    :				Union[str, Any]=None   ,      SCREAMING_SNAKE_CASE_    :				Optional[int]=None   ,      SCREAMING_SNAKE_CASE_    :				List[Any]=None   ,      **SCREAMING_SNAKE_CASE_    :				List[str]	):
        lowerCAmelCase_       :    str								=			{}
        lowerCAmelCase_       :    str								=			{}
        lowerCAmelCase_       :    Optional[int]								=			{}

        if min_length_for_response is not None:
          lowerCAmelCase_       :    Union[str, Any]								=			min_length_for_response
        if minimum_tokens is not None:
          lowerCAmelCase_       :    int								=			minimum_tokens

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

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



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



      def 					SCREAMING_SNAKE_CASE__			(				self    :				Tuple   ,      SCREAMING_SNAKE_CASE_    :				Conversation   ,      SCREAMING_SNAKE_CASE_    :				Dict=3_2	):
        if not isinstance(UpperCAmelCase_   ,      UpperCAmelCase_	):
          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'	):
          lowerCAmelCase_       :    Any								=			self.tokenizer._build_conversation_input_ids(UpperCAmelCase_	)
        else:
          # If the tokenizer cannot handle conversations, we default to only the old version
          lowerCAmelCase_       :    Optional[Any]								=			self._legacy_parse_and_tokenize(UpperCAmelCase_	)

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



      def 					SCREAMING_SNAKE_CASE__			(				self    :				int   ,      SCREAMING_SNAKE_CASE_    :				int   ,      SCREAMING_SNAKE_CASE_    :				Any=1_0   ,      **SCREAMING_SNAKE_CASE_    :				int	):
        lowerCAmelCase_       :    Union[str, Any]								=			generate_kwargs.get('max_length'   ,      self.model.config.max_length	)

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



      def 					SCREAMING_SNAKE_CASE__			(				self    :				str   ,      SCREAMING_SNAKE_CASE_    :				Union[str, Any]   ,      SCREAMING_SNAKE_CASE_    :				Optional[Any]=True	):
        lowerCAmelCase_       :    Optional[Any]								=			model_outputs["output_ids"]
        lowerCAmelCase_       :    Union[str, Any]								=			self.tokenizer.decode(
            output_ids[0]   ,      skip_special_tokens=UpperCAmelCase_   ,      clean_up_tokenization_spaces=UpperCAmelCase_   ,      )
        lowerCAmelCase_       :    Any								=			model_outputs["conversation"]
        conversation.mark_processed()
        conversation.append_response(UpperCAmelCase_	)
        return conversation





      def 					SCREAMING_SNAKE_CASE__			(				self    :				Union[str, Any]   ,      SCREAMING_SNAKE_CASE_    :				Conversation	):
        lowerCAmelCase_       :    Optional[int]								=			self.tokenizer.eos_token_id
        lowerCAmelCase_       :    Optional[Any]								=			[]
        for is_user, text in conversation.iter_texts():
          if eos_token_id is not None:
            input_ids.extend(self.tokenizer.encode(UpperCAmelCase_   ,      add_special_tokens=UpperCAmelCase_	) + [eos_token_id]	)
          else:
            input_ids.extend(self.tokenizer.encode(UpperCAmelCase_   ,      add_special_tokens=UpperCAmelCase_	)	)

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










"""simple docstring"""







import os

from huggingface_hub.constants import HUGGINGFACE_HUB_CACHE, hf_cache_home


lowercase__		:    int				   =     HUGGINGFACE_HUB_CACHE


lowercase__		:    Tuple				   =     """config.json"""
lowercase__		:    Union[str, Any]				   =     """diffusion_pytorch_model.bin"""
lowercase__		:    List[Any]				   =     """diffusion_flax_model.msgpack"""
lowercase__		:    List[str]				   =     """model.onnx"""
lowercase__		:    List[Any]				   =     """diffusion_pytorch_model.safetensors"""
lowercase__		:    Dict				   =     """weights.pb"""
lowercase__		:    List[Any]				   =     """https://huggingface.co"""
lowercase__		:    List[Any]				   =     default_cache_path
lowercase__		:    Tuple				   =     """diffusers_modules"""
lowercase__		:    Tuple				   =     os.getenv("""HF_MODULES_CACHE""", os.path.join(hf_cache_home, """modules"""))
lowercase__		:    List[str]				   =     ["""fp16""", """non-ema"""]
lowercase__		:    Optional[Any]				   =     """.self_attn"""





