




import os
import re
import unicodedata
from shutil import copyfile
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import is_torch_available, logging


if is_torch_available():
				import torch


if TYPE_CHECKING:
				from transformers.pipelines.conversational import Conversation


a													=			logging.get_logger(__name__)
a													=			{"vocab_file": "spiece.model"}

a													=			{
    "vocab_file": {
        "AI-Sweden/gpt-sw3-126m": "https://huggingface.co/AI-Sweden/gpt-sw3-126m/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-350m": "https://huggingface.co/AI-Sweden/gpt-sw3-350m/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-1.6b": "https://huggingface.co/AI-Sweden/gpt-sw3-1.6b/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-6.7b": "https://huggingface.co/AI-Sweden/gpt-sw3-6.7b/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-20b": "https://huggingface.co/AI-Sweden/gpt-sw3-20b/resolve/main/spiece.model",
    }
}

a													=			{
    "AI-Sweden/gpt-sw3-126m": 2048,
    "AI-Sweden/gpt-sw3-350m": 2048,
    "AI-Sweden/gpt-sw3-1.6b": 2048,
    "AI-Sweden/gpt-sw3-6.7b": 2048,
    "AI-Sweden/gpt-sw3-20b": 2048,
}




class       _A				(		__lowercase     ):
		__a					    =       VOCAB_FILES_NAMES
		__a					    =       PRETRAINED_VOCAB_FILES_MAP
		__a					    =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		__a					    =       ["""input_ids""", """attention_mask"""]






		def __init__(    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=False						,		_SCREAMING_SNAKE_CASE=False						,		_SCREAMING_SNAKE_CASE=False						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE = None						,		**_SCREAMING_SNAKE_CASE						,		):
								_UpperCAmelCase      =			{} if sp_model_kwargs is None else sp_model_kwargs

								_UpperCAmelCase      =			kwargs.get("""name_or_path"""			)
								if name_or_path is None:
														logger.warning(
														    """name_or_path not provided, will work for all GPTSw3 models except gpt-sw3-7b,"""
														    """ you are testing the model, this can safely be ignored"""			)
														_UpperCAmelCase      =			"""None"""

								# Default definitions for our 2 tokenizer versions, with None-checks to enable proper testing
								_UpperCAmelCase      =			"""<|endoftext|>""" if eos_token is None else eos_token
								_UpperCAmelCase      =			"""<unk>""" if unk_token is None else unk_token
								if "gpt-sw3-7b" in name_or_path:
														_UpperCAmelCase      =			unk_token if pad_token is None else pad_token
														_UpperCAmelCase      =			eos_token if bos_token is None else bos_token
								else:
														_UpperCAmelCase      =			"""<pad>""" if pad_token is None else pad_token
														_UpperCAmelCase      =			"""<s>""" if bos_token is None else bos_token

								super().__init__(
								    do_lower_case=_SCREAMING_SNAKE_CASE						,		remove_space=_SCREAMING_SNAKE_CASE						,		keep_accents=_SCREAMING_SNAKE_CASE						,		bos_token=_SCREAMING_SNAKE_CASE						,		eos_token=_SCREAMING_SNAKE_CASE						,		unk_token=_SCREAMING_SNAKE_CASE						,		pad_token=_SCREAMING_SNAKE_CASE						,		sp_model_kwargs=self.sp_model_kwargs						,		**_SCREAMING_SNAKE_CASE						,		)

								_UpperCAmelCase      =			do_lower_case
								_UpperCAmelCase      =			remove_space
								_UpperCAmelCase      =			keep_accents
								_UpperCAmelCase      =			vocab_file

								_UpperCAmelCase      =			spm.SentencePieceProcessor(**self.sp_model_kwargs			)
								self.sp_model.Load(_SCREAMING_SNAKE_CASE			)

								# Used for whitespace normalization in input texts
								# fmt : off
								_UpperCAmelCase      =			{""" """, """ """, """ """, """ """, """ """, """　""", """ """, """ """, """ """, """ """, """￼""", """"""}
								# fmt : on

								# Regular expression to remove non-printing characters (e.g. some unicode control chars) in preprocessing
								_UpperCAmelCase      =			re.compile(
								    F"[{''.join(map(_SCREAMING_SNAKE_CASE						,		list(range(0						,		9			)			) + list(range(11						,		32			)			) + list(range(127						,		160			)			) + [160, 173, 8203]			)			)}]"			)






		def __getstate__(    self			):
								_UpperCAmelCase      =			self.__dict__.copy()
								_UpperCAmelCase      =			None
								return state






		def __setstate__(    self						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			d

								# for backward compatibility
								if not hasattr(self						,		"""sp_model_kwargs"""			):
														_UpperCAmelCase      =			{}

								_UpperCAmelCase      =			spm.SentencePieceProcessor(**self.sp_model_kwargs			)
								self.sp_model.Load(self.vocab_file			)






		@property
		# Copied from transformers.models.albert.tokenization_albert.AlbertTokenizer.vocab_size
		def 				UpperCAmelCase (    self			):
								return len(self.sp_model			)






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			self.non_printing_characters_re.sub(""""""						,		_SCREAMING_SNAKE_CASE			)

								# Normalize whitespaces
								_UpperCAmelCase      =			"""""".join([char if char not in self.whitespaces else """ """ for char in text]			)

								# NFC Unicode normalization
								_UpperCAmelCase      =			unicodedata.normalize("""NFC"""						,		_SCREAMING_SNAKE_CASE			)
								return text






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			self.preprocess_text(_SCREAMING_SNAKE_CASE			)
								return self.sp_model.encode(_SCREAMING_SNAKE_CASE						,		out_type=_SCREAMING_SNAKE_CASE			)






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								return self.sp_model.PieceToId(_SCREAMING_SNAKE_CASE			)






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								return self.sp_model.IdToPiece(_SCREAMING_SNAKE_CASE			)






		@staticmethod
		def 				UpperCAmelCase (    _SCREAMING_SNAKE_CASE			):
								return out_string






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			[]
								_UpperCAmelCase      =			""""""
								_UpperCAmelCase      =			False
								for token in tokens:
														# make sure that special tokens are not decoded using sentencepiece model
														if token in self.all_special_tokens:
																				# TODO: Check if this is needed, as it ensures that decode(encode(doc)) != doc by adding extra whitespace in the decoded document
																				if not prev_is_special:
																										out_string += " "

																				out_string += self.sp_model.decode(_SCREAMING_SNAKE_CASE			) + token
																				_UpperCAmelCase      =			True
																				_UpperCAmelCase      =			[]
														else:
																				current_sub_tokens.append(_SCREAMING_SNAKE_CASE			)
																				_UpperCAmelCase      =			False
								out_string += self.sp_model.decode(_SCREAMING_SNAKE_CASE			)

								return out_string






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			{self.convert_ids_to_tokens(_SCREAMING_SNAKE_CASE			): i for i in range(self.vocab_size			)}
								vocab.update(self.added_tokens_encoder			)
								return vocab






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE = None			):
								if not os.path.isdir(_SCREAMING_SNAKE_CASE			):
														logger.error(F"Vocabulary path ({save_directory}) should be a directory"			)
														return
								_UpperCAmelCase      =			os.path.join(
								    _SCREAMING_SNAKE_CASE						,		(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]			)

								if os.path.abspath(self.vocab_file			) != os.path.abspath(_SCREAMING_SNAKE_CASE			) and os.path.isfile(self.vocab_file			):
														copyfile(self.vocab_file						,		_SCREAMING_SNAKE_CASE			)
								elif not os.path.isfile(self.vocab_file			):
														with open(_SCREAMING_SNAKE_CASE						,		"""wb"""			) as fi:
																				_UpperCAmelCase      =			self.sp_model.serialized_model_proto()
																				fi.write(_SCREAMING_SNAKE_CASE			)

								return (out_vocab_file,)






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE = False			):

								if isinstance(_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE			):
														_UpperCAmelCase      =			self.preprocess_text(_SCREAMING_SNAKE_CASE			)
														_UpperCAmelCase      =			self.sp_model.encode(_SCREAMING_SNAKE_CASE			)
								else:
														_UpperCAmelCase      =			[self.preprocess_text(_SCREAMING_SNAKE_CASE			) for t in text]
														_UpperCAmelCase      =			self.sp_model.encode(_SCREAMING_SNAKE_CASE			)

								if return_tensors is True or return_tensors == "pt":
														_UpperCAmelCase      =			torch.tensor(_SCREAMING_SNAKE_CASE			)

								return token_ids






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								return self.sp_model.decode(_SCREAMING_SNAKE_CASE			)






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			[F"User: {text}" if is_user else F"Bot: {text}" for is_user, text in conversation.iter_texts()]
								_UpperCAmelCase      =			(
								    F"{self.eos_token}{self.bos_token}" + F"{self.bos_token}".join(_SCREAMING_SNAKE_CASE			) + F"{self.bos_token}Bot:"
								)
								return self.encode(text=_SCREAMING_SNAKE_CASE			)





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




class       _A				:






		def __init__(    self						,		_SCREAMING_SNAKE_CASE = None						,		_SCREAMING_SNAKE_CASE = None						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None			):
								if not conversation_id:
														_UpperCAmelCase      =			uuid.uuida()
								if past_user_inputs is None:
														_UpperCAmelCase      =			[]
								if generated_responses is None:
														_UpperCAmelCase      =			[]

								_UpperCAmelCase      =			conversation_id
								_UpperCAmelCase      =			past_user_inputs
								_UpperCAmelCase      =			generated_responses
								_UpperCAmelCase      =			text






		def __eq__(    self						,		_SCREAMING_SNAKE_CASE			):
								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 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE = 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}\"."			)
																				_UpperCAmelCase      =			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:
														_UpperCAmelCase      =			text






		def 				UpperCAmelCase (    self			):
								if self.new_user_input:
														self.past_user_inputs.append(self.new_user_input			)
								_UpperCAmelCase      =			None






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								self.generated_responses.append(_SCREAMING_SNAKE_CASE			)






		def 				UpperCAmelCase (    self			):
								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			):
								_UpperCAmelCase      =			F"Conversation id: {self.uuid} \n"
								for is_user, text in self.iter_texts():
														_UpperCAmelCase      =			"""user""" if is_user else """bot"""
														output += F"{name} >> {text} \n"
								return output




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






		def __init__(    self						,		*_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			):
								super().__init__(*_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			)
								if self.tokenizer.pad_token_id is None:
														_UpperCAmelCase      =			self.tokenizer.eos_token






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		**_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			{}
								_UpperCAmelCase      =			{}
								_UpperCAmelCase      =			{}

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

								if "max_length" in generate_kwargs:
														_UpperCAmelCase      =			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:
														_UpperCAmelCase      =			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			):
								_UpperCAmelCase      =			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 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=32			):
								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"""			):
														_UpperCAmelCase      =			self.tokenizer._build_conversation_input_ids(_SCREAMING_SNAKE_CASE			)
								else:
														# If the tokenizer cannot handle conversations, we default to only the old version
														_UpperCAmelCase      =			self._legacy_parse_and_tokenize(_SCREAMING_SNAKE_CASE			)

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






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=10						,		**_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			generate_kwargs.get("""max_length"""						,		self.model.config.max_length			)

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






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=True			):
								_UpperCAmelCase      =			model_outputs["""output_ids"""]
								_UpperCAmelCase      =			self.tokenizer.decode(
								    output_ids[0]						,		skip_special_tokens=_SCREAMING_SNAKE_CASE						,		clean_up_tokenization_spaces=_SCREAMING_SNAKE_CASE						,		)
								_UpperCAmelCase      =			model_outputs["""conversation"""]
								conversation.mark_processed()
								conversation.append_response(_SCREAMING_SNAKE_CASE			)
								return conversation






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			self.tokenizer.eos_token_id
								_UpperCAmelCase      =			[]
								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:
														_UpperCAmelCase      =			input_ids[-self.tokenizer.model_max_length :]
								return input_ids