



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


lowerCAmelCase:      int		 =		logging.get_logger(__name__)

class 							a__:







				def __init__(   self				:						Union[str, Any]		,    __snake_case				:						str = None		,    __snake_case				:						uuid.UUID = None		,    __snake_case				:						Dict=None		,    __snake_case				:						Any=None			):
						if not conversation_id:
								a		:      Any       =  uuid.uuida()
						if past_user_inputs is None:
								a		:      str       =  []
						if generated_responses is None:
								a		:      List[str]       =  []

						a		:      int       =  conversation_id
						a		:      Any       =  past_user_inputs
						a		:      Dict       =  generated_responses
						a		:      Optional[int]       =  text







				def __eq__(   self				:						Optional[Any]		,    __snake_case				:						Union[str, Any]			):
						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     lowercase_ (   self				:						int		,    __snake_case				:						str		,    __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}\"."""			)
										a		:      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:
								a		:      Dict       =  text







				def     lowercase_ (   self				:						List[str]			):
						if self.new_user_input:
								self.past_user_inputs.append(self.new_user_input			)
						a		:      int       =  None







				def     lowercase_ (   self				:						List[Any]		,    __snake_case				:						str			):
						self.generated_responses.append(UpperCamelCase_			)







				def     lowercase_ (   self				:						Tuple			):
						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				:						Union[str, Any]			):
						a		:      int       =  F"""Conversation id: {self.uuid} \n"""
						for is_user, text in self.iter_texts():
								a		:      Optional[Any]       =  '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 							a__(      __lowerCamelCase			):







				def __init__(   self				:						List[str]		,    *__snake_case				:						List[Any]		,    **__snake_case				:						str			):
						super().__init__(*UpperCamelCase_		,    **UpperCamelCase_			)
						if self.tokenizer.pad_token_id is None:
								a		:      Optional[Any]       =  self.tokenizer.eos_token







				def     lowercase_ (   self				:						str		,    __snake_case				:						int=None		,    __snake_case				:						Any=None		,    __snake_case				:						Union[str, Any]=None		,    **__snake_case				:						int			):
						a		:      Union[str, Any]       =  {}
						a		:      str       =  {}
						a		:      Tuple       =  {}

						if min_length_for_response is not None:
								a		:      Dict       =  min_length_for_response
						if minimum_tokens is not None:
								a		:      Any       =  minimum_tokens

						if "max_length" in generate_kwargs:
								a		:      List[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:
								a		:      int       =  clean_up_tokenization_spaces

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







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







				def     lowercase_ (   self				:						List[Any]		,    __snake_case				:						Conversation		,    __snake_case				:						Optional[Any]=32			):
						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'			):
								a		:      Any       =  self.tokenizer._build_conversation_input_ids(UpperCamelCase_			)
						else:
								# If the tokenizer cannot handle conversations, we default to only the old version
								a		:      List[Any]       =  self._legacy_parse_and_tokenize(UpperCamelCase_			)

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







				def     lowercase_ (   self				:						Union[str, Any]		,    __snake_case				:						Union[str, Any]		,    __snake_case				:						str=10		,    **__snake_case				:						List[str]			):
						a		:      Union[str, Any]       =  generate_kwargs.get('max_length'		,    self.model.config.max_length			)

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







				def     lowercase_ (   self				:						Dict		,    __snake_case				:						Optional[Any]		,    __snake_case				:						int=True			):
						a		:      List[Any]       =  model_outputs['output_ids']
						a		:      Tuple       =  self.tokenizer.decode(
						    output_ids[0]		,    skip_special_tokens=UpperCamelCase_		,    clean_up_tokenization_spaces=UpperCamelCase_		,    )
						a		:      Optional[int]       =  model_outputs['conversation']
						conversation.mark_processed()
						conversation.append_response(UpperCamelCase_			)
						return conversation







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




'''simple docstring'''






import os
import re
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase:      List[Any]		 =		logging.get_logger(__name__)

lowerCAmelCase:      Dict		 =		{
    'vocab_file': 'vocab.txt',
    'merges_file': 'bpe.codes',
}

lowerCAmelCase:      List[Any]		 =		{
    'vocab_file': {
        'vinai/phobert-base': 'https://huggingface.co/vinai/phobert-base/resolve/main/vocab.txt',
        'vinai/phobert-large': 'https://huggingface.co/vinai/phobert-large/resolve/main/vocab.txt',
    },
    'merges_file': {
        'vinai/phobert-base': 'https://huggingface.co/vinai/phobert-base/resolve/main/bpe.codes',
        'vinai/phobert-large': 'https://huggingface.co/vinai/phobert-large/resolve/main/bpe.codes',
    },
}

lowerCAmelCase:      Optional[Any]		 =		{
    'vinai/phobert-base': 2_5_6,
    'vinai/phobert-large': 2_5_6,
}
def    lowerCamelCase__ (		_A       ):
  a		:      List[Any]       =  set()
  a		:      Union[str, Any]       =  word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)       )
    a		:      Tuple       =  char

  a		:      Any       =  set(_A       )
  return pairs

class 							a__(      lowerCamelCase__			):
    lowercase__				=					VOCAB_FILES_NAMES
    lowercase__				=					PRETRAINED_VOCAB_FILES_MAP
    lowercase__				=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES







    def __init__(   self				:						Optional[Any]		,    __snake_case				:						Union[str, Any]		,    __snake_case				:						int		,    __snake_case				:						Any="<s>"		,    __snake_case				:						List[Any]="</s>"		,    __snake_case				:						Union[str, Any]="</s>"		,    __snake_case				:						List[Any]="<s>"		,    __snake_case				:						Optional[Any]="<unk>"		,    __snake_case				:						Union[str, Any]="<pad>"		,    __snake_case				:						List[Any]="<mask>"		,    **__snake_case				:						int		,    ):
      super().__init__(
          bos_token=__snake_case		,    eos_token=__snake_case		,    unk_token=__snake_case		,    sep_token=__snake_case		,    cls_token=__snake_case		,    pad_token=__snake_case		,    mask_token=__snake_case		,    **__snake_case		,    )

      a		:      Optional[int]       =  vocab_file
      a		:      Optional[Any]       =  merges_file

      a		:      int       =  {}
      a		:      List[str]       =  0
      a		:      Union[str, Any]       =  1
      a		:      Optional[Any]       =  2
      a		:      List[Any]       =  3

      self.add_from_file(__snake_case			)

      a		:      Dict       =  {v: k for k, v in self.encoder.items()}

      with open(__snake_case		,    encoding='utf-8'			) as merges_handle:
        a		:      Optional[int]       =  merges_handle.read().split('\n'			)[:-1]
      a		:      Any       =  [tuple(merge.split()[:-1]			) for merge in merges]
      a		:      List[Any]       =  dict(zip(__snake_case		,    range(len(__snake_case			)			)			)			)
      a		:      List[str]       =  {}







    def     lowercase_ (   self				:						Tuple		,    __snake_case				:						List[int]		,    __snake_case				:						Optional[List[int]] = None			):

      if token_ids_a is None:
        return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
      a		:      Union[str, Any]       =  [self.cls_token_id]
      a		:      Any       =  [self.sep_token_id]
      return cls + token_ids_a + sep + sep + token_ids_a + sep







    def     lowercase_ (   self				:						Optional[int]		,    __snake_case				:						List[int]		,    __snake_case				:						Optional[List[int]] = None		,    __snake_case				:						bool = False			):

      if already_has_special_tokens:
        return super().get_special_tokens_mask(
            token_ids_a=__snake_case		,    token_ids_a=__snake_case		,    already_has_special_tokens=__snake_case			)

      if token_ids_a is None:
        return [1] + ([0] * len(__snake_case			)) + [1]
      return [1] + ([0] * len(__snake_case			)) + [1, 1] + ([0] * len(__snake_case			)) + [1]







    def     lowercase_ (   self				:						int		,    __snake_case				:						List[int]		,    __snake_case				:						Optional[List[int]] = None			):
      a		:      Dict       =  [self.sep_token_id]
      a		:      Optional[int]       =  [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 + sep + token_ids_a + sep			) * [0]







    @property
    def     lowercase_ (   self				:						Optional[int]			):
      return len(self.encoder			)







    def     lowercase_ (   self				:						Dict			):
      return dict(self.encoder		,    **self.added_tokens_encoder			)







    def     lowercase_ (   self				:						Optional[int]		,    __snake_case				:						Optional[Any]			):
      if token in self.cache:
        return self.cache[token]
      a		:      int       =  tuple(__snake_case			)
      a		:      int       =  tuple(list(word[:-1]			) + [word[-1] + '</w>']			)
      a		:      Dict       =  get_pairs(__snake_case			)

      if not pairs:
        return token

      while True:
        a		:      str       =  min(__snake_case		,    key=lambda __snake_case			: self.bpe_ranks.get(__snake_case		,    float('inf'			)			)			)
        if bigram not in self.bpe_ranks:
          break
        a     ,			a		:      Tuple       =  bigram
        a		:      int       =  []
        a		:      Optional[int]       =  0
        while i < len(__snake_case			):
          try:
            a		:      List[str]       =  word.index(__snake_case		,    __snake_case			)
          except ValueError:
            new_word.extend(word[i:]			)
            break
          else:
            new_word.extend(word[i:j]			)
            a		:      Dict       =  j

          if word[i] == first and i < len(__snake_case			) - 1 and word[i + 1] == second:
            new_word.append(first + second			)
            i += 2
          else:
            new_word.append(word[i]			)
            i += 1
        a		:      Optional[Any]       =  tuple(__snake_case			)
        a		:      Tuple       =  new_word
        if len(__snake_case			) == 1:
          break
        else:
          a		:      Union[str, Any]       =  get_pairs(__snake_case			)
      a		:      List[Any]       =  '@@ '.join(__snake_case			)
      a		:      int       =  word[:-4]
      a		:      Optional[int]       =  word
      return word







    def     lowercase_ (   self				:						Dict		,    __snake_case				:						List[str]			):
      a		:      Optional[int]       =  []

      a		:      Optional[Any]       =  re.findall(r'\S+\n?'		,    __snake_case			)

      for token in words:
        split_tokens.extend(list(self.bpe(__snake_case			).split(' '			)			)			)
      return split_tokens







    def     lowercase_ (   self				:						List[Any]		,    __snake_case				:						Dict			):
      return self.encoder.get(__snake_case		,    self.encoder.get(self.unk_token			)			)







    def     lowercase_ (   self				:						Dict		,    __snake_case				:						Dict			):
      return self.decoder.get(__snake_case		,    self.unk_token			)







    def     lowercase_ (   self				:						List[Any]		,    __snake_case				:						Any			):
      a		:      Union[str, Any]       =  ' '.join(__snake_case			).replace('@@ '		,    ''			).strip()
      return out_string







    def     lowercase_ (   self				:						str		,    __snake_case				:						str		,    __snake_case				:						Optional[str] = None			):
      if not os.path.isdir(__snake_case			):
        logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""			)
        return
      a		:      Dict       =  os.path.join(
          __snake_case		,    (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']			)
      a		:      int       =  os.path.join(
          __snake_case		,    (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']			)

      if os.path.abspath(self.vocab_file			) != os.path.abspath(__snake_case			):
        copyfile(self.vocab_file		,    __snake_case			)

      if os.path.abspath(self.merges_file			) != os.path.abspath(__snake_case			):
        copyfile(self.merges_file		,    __snake_case			)

      return out_vocab_file, out_merge_file







    def     lowercase_ (   self				:						Tuple		,    __snake_case				:						List[str]			):
      if isinstance(__snake_case		,    __snake_case			):
        try:
          with open(__snake_case		,    'r'		,    encoding='utf-8'			) as fd:
            self.add_from_file(__snake_case			)
        except FileNotFoundError as fnfe:
          raise fnfe
        except UnicodeError:
          raise Exception(F"""Incorrect encoding detected in {f}, please rebuild the dataset"""			)
        return

      a		:      List[str]       =  f.readlines()
      for lineTmp in lines:
        a		:      Any       =  lineTmp.strip()
        a		:      Dict       =  line.rfind(' '			)
        if idx == -1:
          raise ValueError('Incorrect dictionary format, expected \'<token> <cnt>\''			)
        a		:      str       =  line[:idx]
        a		:      Optional[int]       =  len(self.encoder			)