


'''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


A__						:   Tuple				=    logging.get_logger(__name__)





class   UpperCAmelCase_    :





    """simple docstring"""






    def __init__(				self     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=None      )							->	Dict:
         if not conversation_id:
              __lowerCamelCase  :      int														=						uuid.uuida()
         if past_user_inputs is None:
              __lowerCamelCase  :      List[str]														=						[]
         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     ,	SCREAMING_SNAKE_CASE_      )							->	int:
         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  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = False      )							->	Tuple:
         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[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:
              __lowerCamelCase  :      List[str]														=						text







    def  lowercase_							(				self      )							->	str:
         if self.new_user_input:
              self.past_user_inputs.append(self.new_user_input      )
         __lowerCamelCase  :      Any														=						None







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_      )							->	Union[str, Any]:
         self.generated_responses.append(SCREAMING_SNAKE_CASE_      )







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







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





    """simple docstring"""






    def __init__(				self     ,	*SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )							->	Optional[int]:
         super().__init__(*SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )
         if self.tokenizer.pad_token_id is None:
              __lowerCamelCase  :      Tuple														=						self.tokenizer.eos_token







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=None     ,	**SCREAMING_SNAKE_CASE_      )							->	str:
         __lowerCamelCase  :      Union[str, Any]														=						{}
         __lowerCamelCase  :      int														=						{}
         __lowerCamelCase  :      Dict														=						{}

         if min_length_for_response is not None:
              __lowerCamelCase  :      Union[str, Any]														=						min_length_for_response
         if minimum_tokens is not None:
              __lowerCamelCase  :      Any														=						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  :      str														=						clean_up_tokenization_spaces

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







    def __call__(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=0     ,	**SCREAMING_SNAKE_CASE_      )							->	Any:
         __lowerCamelCase  :      int														=						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  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=32      )							->	Dict[str, Any]:
         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'      ):
              __lowerCamelCase  :      Optional[Any]														=						self.tokenizer._build_conversation_input_ids(SCREAMING_SNAKE_CASE_      )
         else:
              # If the tokenizer cannot handle conversations, we default to only the old version
              __lowerCamelCase  :      Union[str, Any]														=						self._legacy_parse_and_tokenize(SCREAMING_SNAKE_CASE_      )

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







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=10     ,	**SCREAMING_SNAKE_CASE_      )							->	List[str]:
         __lowerCamelCase  :      Dict														=						generate_kwargs.get('max_length'     ,	self.model.config.max_length      )

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







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=True      )							->	Dict:
         __lowerCamelCase  :      Any														=						model_outputs['output_ids']
         __lowerCamelCase  :      Optional[int]														=						self.tokenizer.decode(
             output_ids[0]     ,	skip_special_tokens=SCREAMING_SNAKE_CASE_     ,	clean_up_tokenization_spaces=SCREAMING_SNAKE_CASE_     ,	)
         __lowerCamelCase  :      int														=						model_outputs['conversation']
         conversation.mark_processed()
         conversation.append_response(SCREAMING_SNAKE_CASE_      )
         return conversation







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_      )							->	Dict:
         __lowerCamelCase  :      Tuple														=						self.tokenizer.eos_token_id
         __lowerCamelCase  :      List[str]														=						[]
         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:
              __lowerCamelCase  :      Any														=						input_ids[-self.tokenizer.model_max_length :]
         return input_ids










'''simple docstring'''



import json
from typing import List, Optional, Tuple

from tokenizers import normalizers
from tokenizers.pre_tokenizers import BertPreTokenizer, PreTokenizer

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roformer import RoFormerTokenizer
from .tokenization_utils import JiebaPreTokenizer


A__						:   str				=    logging.get_logger(__name__)

A__						:   str				=    {"""vocab_file""": """vocab.txt""", """tokenizer_file""": """tokenizer.json"""}

A__						:   Tuple				=    {
    """vocab_file""": {
        """junnyu/roformer_chinese_small""": """https://huggingface.co/junnyu/roformer_chinese_small/resolve/main/vocab.txt""",
        """junnyu/roformer_chinese_base""": """https://huggingface.co/junnyu/roformer_chinese_base/resolve/main/vocab.txt""",
        """junnyu/roformer_chinese_char_small""": (
            """https://huggingface.co/junnyu/roformer_chinese_char_small/resolve/main/vocab.txt"""
        ),
        """junnyu/roformer_chinese_char_base""": (
            """https://huggingface.co/junnyu/roformer_chinese_char_base/resolve/main/vocab.txt"""
        ),
        """junnyu/roformer_small_discriminator""": (
            """https://huggingface.co/junnyu/roformer_small_discriminator/resolve/main/vocab.txt"""
        ),
        """junnyu/roformer_small_generator""": (
            """https://huggingface.co/junnyu/roformer_small_generator/resolve/main/vocab.txt"""
        ),
    }
}

A__						:   str				=    {
    """junnyu/roformer_chinese_small""": 1536,
    """junnyu/roformer_chinese_base""": 1536,
    """junnyu/roformer_chinese_char_small""": 512,
    """junnyu/roformer_chinese_char_base""": 512,
    """junnyu/roformer_small_discriminator""": 128,
    """junnyu/roformer_small_generator""": 128,
}


A__						:   Tuple				=    {
    """junnyu/roformer_chinese_small""": {"""do_lower_case""": True},
    """junnyu/roformer_chinese_base""": {"""do_lower_case""": True},
    """junnyu/roformer_chinese_char_small""": {"""do_lower_case""": True},
    """junnyu/roformer_chinese_char_base""": {"""do_lower_case""": True},
    """junnyu/roformer_small_discriminator""": {"""do_lower_case""": True},
    """junnyu/roformer_small_generator""": {"""do_lower_case""": True},
}





class   UpperCAmelCase_    (_UpperCAmelCase    ):





    """simple docstring"""






    lowerCamelCase    :   Union[str, Any]					=     VOCAB_FILES_NAMES
    lowerCamelCase    :   int					=     PRETRAINED_VOCAB_FILES_MAP
    lowerCamelCase    :   Any					=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    lowerCamelCase    :   Optional[int]					=     PRETRAINED_INIT_CONFIGURATION
    lowerCamelCase    :   Dict					=     RoFormerTokenizer







    def __init__(				self     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=True     ,	SCREAMING_SNAKE_CASE_="[UNK]"     ,	SCREAMING_SNAKE_CASE_="[SEP]"     ,	SCREAMING_SNAKE_CASE_="[PAD]"     ,	SCREAMING_SNAKE_CASE_="[CLS]"     ,	SCREAMING_SNAKE_CASE_="[MASK]"     ,	SCREAMING_SNAKE_CASE_=True     ,	SCREAMING_SNAKE_CASE_=None     ,	**SCREAMING_SNAKE_CASE_     ,	)							->	Optional[Any]:
         super().__init__(
             SCREAMING_SNAKE_CASE_     ,	tokenizer_file=SCREAMING_SNAKE_CASE_     ,	do_lower_case=SCREAMING_SNAKE_CASE_     ,	unk_token=SCREAMING_SNAKE_CASE_     ,	sep_token=SCREAMING_SNAKE_CASE_     ,	pad_token=SCREAMING_SNAKE_CASE_     ,	cls_token=SCREAMING_SNAKE_CASE_     ,	mask_token=SCREAMING_SNAKE_CASE_     ,	tokenize_chinese_chars=SCREAMING_SNAKE_CASE_     ,	strip_accents=SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_     ,	)

         __lowerCamelCase  :      Optional[int]														=						json.loads(self.backend_tokenizer.normalizer.__getstate__()      )
         if (
             pre_tok_state.get('lowercase'     ,	SCREAMING_SNAKE_CASE_      ) != do_lower_case
             or pre_tok_state.get('strip_accents'     ,	SCREAMING_SNAKE_CASE_      ) != strip_accents
         ):
              __lowerCamelCase  :      Optional[int]														=						getattr(SCREAMING_SNAKE_CASE_     ,	pre_tok_state.pop('type'      )      )
              __lowerCamelCase  :      Union[str, Any]														=						do_lower_case
              __lowerCamelCase  :      str														=						strip_accents
              __lowerCamelCase  :      Optional[Any]														=						pre_tok_class(**SCREAMING_SNAKE_CASE_      )

         __lowerCamelCase  :      Tuple														=						do_lower_case







    def __getstate__(				self      )							->	List[str]:
         __lowerCamelCase  :      Union[str, Any]														=						self.__dict__.copy()
         __lowerCamelCase  :      Dict														=						BertPreTokenizer()
         return state







    def __setstate__(				self     ,	SCREAMING_SNAKE_CASE_      )							->	Optional[Any]:
         __lowerCamelCase  :      Optional[int]														=						d
         __lowerCamelCase  :      List[Any]														=						self.__dict__['_tokenizer'].get_vocab()
         __lowerCamelCase  :      Union[str, Any]														=						PreTokenizer.custom(JiebaPreTokenizer(SCREAMING_SNAKE_CASE_      )      )







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=None      )							->	str:
         __lowerCamelCase  :      Union[str, Any]														=						[self.cls_token_id] + token_ids_a + [self.sep_token_id]

         if token_ids_a:
              output += token_ids_a + [self.sep_token_id]

         return output







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = None      )							->	List[int]:
         __lowerCamelCase  :      List[str]														=						[self.sep_token_id]
         __lowerCamelCase  :      Dict														=						[self.cls_token_id]
         if token_ids_a is None:
              return len(cls + token_ids_a + sep      ) * [0]
         return len(cls + token_ids_a + sep      ) * [0] + len(token_ids_a + sep      ) * [1]







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = None      )							->	Tuple[str]:
         __lowerCamelCase  :      Optional[Any]														=						self._tokenizer.model.save(SCREAMING_SNAKE_CASE_     ,	name=SCREAMING_SNAKE_CASE_      )
         return tuple(SCREAMING_SNAKE_CASE_      )







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=False     ,	**SCREAMING_SNAKE_CASE_     ,	)							->	Any:
         __lowerCamelCase  :      Tuple														=						BertPreTokenizer()
         return super().save_pretrained(SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )






