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





class       __A:

					def __init__(	self					:     List[Any]   ,					__UpperCamelCase					:     Any = None   ,					__UpperCamelCase					:     Any = None   ,					__UpperCamelCase					:     List[Any]=None   ,					__UpperCamelCase					:     List[str]=None   ):
												if not conversation_id:
																			lowerCamelCase_  =					uuid.uuida()
												if past_user_inputs is None:
																			lowerCamelCase_  =					[]
												if generated_responses is None:
																			lowerCamelCase_  =					[]

												lowerCamelCase_  =					conversation_id
												lowerCamelCase_  =					past_user_inputs
												lowerCamelCase_  =					generated_responses
												lowerCamelCase_  =					text

					def __eq__(	self					:     Union[str, Any]   ,					__UpperCamelCase					:     str   ):
												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					:     Tuple   ,					__UpperCamelCase					:     Any   ,					__UpperCamelCase					:     Dict = 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_  =					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_  =					text

					def    lowercase__      (	self					:     Union[str, Any]   ):
												if self.new_user_input:
																			self.past_user_inputs.append(self.new_user_input   )
												lowerCamelCase_  =					None

					def    lowercase__      (	self					:     Optional[Any]   ,					__UpperCamelCase					:     Any   ):
												self.generated_responses.append(lowercase_   )

					def    lowercase__      (	self					:     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					:     Dict   ):
												lowerCamelCase_  =					F'''Conversation id: {self.uuid} \n'''
												for is_user, text in self.iter_texts():
																			lowerCamelCase_  =					"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       __A(				snake_case__       ):

					def __init__(	self					:     Tuple   ,					*__UpperCamelCase					:     List[Any]   ,					**__UpperCamelCase					:     Optional[int]   ):
												super().__init__(*lowercase_   ,					**lowercase_   )
												if self.tokenizer.pad_token_id is None:
																			lowerCamelCase_  =					self.tokenizer.eos_token

					def    lowercase__      (	self					:     Union[str, Any]   ,					__UpperCamelCase					:     str=None   ,					__UpperCamelCase					:     Optional[Any]=None   ,					__UpperCamelCase					:     Tuple=None   ,					**__UpperCamelCase					:     int   ):
												lowerCamelCase_  =					{}
												lowerCamelCase_  =					{}
												lowerCamelCase_  =					{}

												if min_length_for_response is not None:
																			lowerCamelCase_  =					min_length_for_response
												if minimum_tokens is not None:
																			lowerCamelCase_  =					minimum_tokens

												if "max_length" in generate_kwargs:
																			lowerCamelCase_  =					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_  =					clean_up_tokenization_spaces

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

					def __call__(	self					:     Optional[Any]   ,					__UpperCamelCase					:     List[Any]   ,					__UpperCamelCase					:     Optional[Any]=0   ,					**__UpperCamelCase					:     Tuple   ):
												lowerCamelCase_  =					super().__call__(lowercase_   ,					num_workers=lowercase_   ,					**lowercase_   )
												if isinstance(lowercase_   ,					lowercase_   ) and len(lowercase_   ) == 1:
																			return outputs[0]
												return outputs

					def    lowercase__      (	self					:     Union[str, Any]   ,					__UpperCamelCase					:     List[str]   ,					__UpperCamelCase					:     Optional[int]=3_2   ):
												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"""   ):
																			lowerCamelCase_  =					self.tokenizer._build_conversation_input_ids(lowercase_   )
												else:
																			# If the tokenizer cannot handle conversations, we default to only the old version
																			lowerCamelCase_  =					self._legacy_parse_and_tokenize(lowercase_   )

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

					def    lowercase__      (	self					:     Union[str, Any]   ,					__UpperCamelCase					:     Optional[Any]   ,					__UpperCamelCase					:     Optional[Any]=1_0   ,					**__UpperCamelCase					:     Optional[Any]   ):
												lowerCamelCase_  =					generate_kwargs.get("""max_length"""   ,					self.model.config.max_length   )

												lowerCamelCase_  =					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_  =					max_length - minimum_tokens
																			lowerCamelCase_  =					model_inputs["input_ids"][:, -trim:]
																			if "attention_mask" in model_inputs:
																										lowerCamelCase_  =					model_inputs["attention_mask"][:, -trim:]
												lowerCamelCase_  =					model_inputs.pop("""conversation"""   )
												lowerCamelCase_  =					max_length
												lowerCamelCase_  =					self.model.generate(**lowercase_   ,					**lowercase_   )
												if self.model.config.is_encoder_decoder:
																			lowerCamelCase_  =					1
												else:
																			lowerCamelCase_  =					n
												return {"output_ids": output_ids[:, start_position:], "conversation": conversation}

					def    lowercase__      (	self					:     str   ,					__UpperCamelCase					:     Optional[Any]   ,					__UpperCamelCase					:     Tuple=True   ):
												lowerCamelCase_  =					model_outputs["output_ids"]
												lowerCamelCase_  =					self.tokenizer.decode(
												    output_ids[0]   ,					skip_special_tokens=lowercase_   ,					clean_up_tokenization_spaces=lowercase_   ,					)
												lowerCamelCase_  =					model_outputs["conversation"]
												conversation.mark_processed()
												conversation.append_response(lowercase_   )
												return conversation



					def    lowercase__      (	self					:     List[str]   ,					__UpperCamelCase					:     List[str]   ):
												lowerCamelCase_  =					self.tokenizer.eos_token_id
												lowerCamelCase_  =					[]
												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:
																			lowerCamelCase_  =					input_ids[-self.tokenizer.model_max_length :]
												return input_ids







import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
		import jieba

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


lowercase        =  logging.get_logger(__name__)

lowercase        =  {'''vocab_file''': '''vocab.txt'''}

lowercase        =  {
    '''vocab_file''': {
        '''openbmb/cpm-ant-10b''': '''https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt''',
    },
}

lowercase        =  {
    '''openbmb/cpm-ant-10b''': 1_0_2_4,
}



def      __lowerCAmelCase						(							UpperCAmelCase__				:      List[str]       )  ->    Dict:
							lowerCamelCase_							  =					collections.OrderedDict()
							with open(UpperCAmelCase__			,				"""r"""			,				encoding="""utf-8"""       ) as reader:
														lowerCamelCase_							  =					reader.readlines()
							for index, token in enumerate(UpperCAmelCase__       ):
														lowerCamelCase_							  =					token.rstrip("""\n"""       )
														lowerCamelCase_							  =					index
							return vocab





class       __A(				UpperCAmelCase       ):

					def __init__(	self					:     Optional[Any]   ,					__UpperCamelCase					:     Tuple   ,					__UpperCamelCase					:     List[str]="<unk>"   ,					__UpperCamelCase					:     List[str]=2_0_0   ):
												lowerCamelCase_							  =					vocab
												lowerCamelCase_							  =					unk_token
												lowerCamelCase_							  =					max_input_chars_per_word



					def    lowercase__      (	self					:     int   ,					__UpperCamelCase					:     int   ):
												lowerCamelCase_							  =					list(__UpperCamelCase   )
												if len(__UpperCamelCase   ) > self.max_input_chars_per_word:
																			return [self.unk_token]

												lowerCamelCase_							  =					0
												lowerCamelCase_							  =					[]
												while start < len(__UpperCamelCase   ):
																			lowerCamelCase_							  =					len(__UpperCamelCase   )
																			lowerCamelCase_							  =					None
																			while start < end:
																										lowerCamelCase_							  =					"""""".join(chars[start:end]   )
																										if substr in self.vocab:
																																	lowerCamelCase_							  =					substr
																																	break
																										end -= 1
																			if cur_substr is None:
																										sub_tokens.append(self.unk_token   )
																										start += 1
																			else:
																										sub_tokens.append(__UpperCamelCase   )
																										lowerCamelCase_							  =					end

												return sub_tokens





class       __A(				UpperCAmelCase       ):
					SCREAMING_SNAKE_CASE			    =    VOCAB_FILES_NAMES
					SCREAMING_SNAKE_CASE			    =    PRETRAINED_VOCAB_FILES_MAP
					SCREAMING_SNAKE_CASE			    =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					SCREAMING_SNAKE_CASE			    =    ['''input_ids''', '''attention_mask''']
					SCREAMING_SNAKE_CASE			    =    False

					def __init__(	self					:     int   ,					__UpperCamelCase					:     Optional[int]   ,					__UpperCamelCase					:     Optional[int]="<d>"   ,					__UpperCamelCase					:     List[str]="</d>"   ,					__UpperCamelCase					:     Optional[Any]="<s>"   ,					__UpperCamelCase					:     Union[str, Any]="</s>"   ,					__UpperCamelCase					:     List[str]="<pad>"   ,					__UpperCamelCase					:     Union[str, Any]="<unk>"   ,					__UpperCamelCase					:     List[Any]="</n>"   ,					__UpperCamelCase					:     Tuple="</_>"   ,					__UpperCamelCase					:     Optional[Any]="left"   ,					**__UpperCamelCase					:     List[str]   ,					):
												requires_backends(self   ,					["""jieba"""]   )
												super().__init__(
												    bod_token=__UpperCamelCase   ,					eod_token=__UpperCamelCase   ,					bos_token=__UpperCamelCase   ,					eos_token=__UpperCamelCase   ,					pad_token=__UpperCamelCase   ,					unk_token=__UpperCamelCase   ,					line_token=__UpperCamelCase   ,					space_token=__UpperCamelCase   ,					padding_side=__UpperCamelCase   ,					**__UpperCamelCase   ,					)
												lowerCamelCase_							  =					bod_token
												lowerCamelCase_							  =					eod_token
												lowerCamelCase_							  =					load_vocab(__UpperCamelCase   )
												lowerCamelCase_							  =					self.encoder[space_token]
												lowerCamelCase_							  =					self.encoder[line_token]

												del self.encoder[space_token]
												del self.encoder[line_token]

												lowerCamelCase_							  =					collections.OrderedDict(sorted(self.encoder.items()   ,					key=lambda __UpperCamelCase   : x[1]   )   )
												lowerCamelCase_							  =					{v: k for k, v in self.encoder.items()}

												lowerCamelCase_							  =					WordpieceTokenizer(vocab=self.encoder   ,					unk_token=self.unk_token   )

					@property
					def    lowercase__      (	self					:     Any   ):
												return self.encoder[self.bod_token]

					@property
					def    lowercase__      (	self					:     Any   ):
												return self.encoder[self.eod_token]

					@property
					def    lowercase__      (	self					:     int   ):
												return self.encoder["\n"]

					@property
					def    lowercase__      (	self					:     str   ):
												return len(self.encoder   )

					def    lowercase__      (	self					:     Any   ):
												return dict(self.encoder   ,					**self.added_tokens_encoder   )

					def    lowercase__      (	self					:     Any   ,					__UpperCamelCase					:     Optional[Any]   ):
												lowerCamelCase_							  =					[]
												for x in jieba.cut(__UpperCamelCase   ,					cut_all=__UpperCamelCase   ):
																			output_tokens.extend(self.wordpiece_tokenizer.tokenize(__UpperCamelCase   )   )
												return output_tokens

					def    lowercase__      (	self					:     Optional[Any]   ,					__UpperCamelCase					:     Dict   ,					**__UpperCamelCase					:     str   ):
												lowerCamelCase_							  =					[i for i in token_ids if i >= 0]
												lowerCamelCase_							  =					[
												    x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
												]
												return super()._decode(__UpperCamelCase   ,					**__UpperCamelCase   )

					def    lowercase__      (	self					:     Dict   ,					__UpperCamelCase					:     Optional[Any]   ):
												return token in self.encoder

					def    lowercase__      (	self					:     int   ,					__UpperCamelCase					:     List[str]   ):
												return "".join(__UpperCamelCase   )

					def    lowercase__      (	self					:     List[str]   ,					__UpperCamelCase					:     int   ):
												return self.encoder.get(__UpperCamelCase   ,					self.encoder.get(self.unk_token   )   )

					def    lowercase__      (	self					:     Any   ,					__UpperCamelCase					:     Union[str, Any]   ):
												return self.decoder.get(__UpperCamelCase   ,					self.unk_token   )

					def    lowercase__      (	self					:     Union[str, Any]   ,					__UpperCamelCase					:     str   ,					__UpperCamelCase					:     Optional[str] = None   ):
												if os.path.isdir(__UpperCamelCase   ):
																			lowerCamelCase_							  =					os.path.join(
																			    __UpperCamelCase   ,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]   )
												else:
																			lowerCamelCase_							  =					(filename_prefix + """-""" if filename_prefix else """""") + save_directory
												lowerCamelCase_							  =					0
												if " " in self.encoder:
																			lowerCamelCase_							  =					self.encoder[""" """]
																			del self.encoder[" "]
												if "\n" in self.encoder:
																			lowerCamelCase_							  =					self.encoder["""\n"""]
																			del self.encoder["\n"]
												lowerCamelCase_							  =					collections.OrderedDict(sorted(self.encoder.items()   ,					key=lambda __UpperCamelCase   : x[1]   )   )
												with open(__UpperCamelCase   ,					"""w"""   ,					encoding="""utf-8"""   ) as writer:
																			for token, token_index in self.encoder.items():
																										if index != token_index:
																																	logger.warning(
																																	    F'''Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive.'''
																																	    """ Please check that the vocabulary is not corrupted!"""   )
																																	lowerCamelCase_							  =					token_index
																										writer.write(token + """\n"""   )
																										index += 1
												return (vocab_file,)

					def    lowercase__      (	self					:     Any   ,					__UpperCamelCase					:     List[int]   ,					__UpperCamelCase					:     List[int] = None   ):
												if token_ids_a is None:
																			return [self.bos_token_id] + token_ids_a
												return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a



					def    lowercase__      (	self					:     int   ,					__UpperCamelCase					:     List[int]   ,					__UpperCamelCase					:     Optional[List[int]] = None   ,					__UpperCamelCase					:     bool = False   ):

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

												if token_ids_a is not None:
																			return [1] + ([0] * len(__UpperCamelCase   )) + [1] + ([0] * len(__UpperCamelCase   ))
												return [1] + ([0] * len(__UpperCamelCase   ))





