


import contextlib
import copy
import random
from typing import Any, Dict, Iterable, Optional, Union

import numpy as np
import torch

from .utils import deprecate, is_transformers_available


if is_transformers_available():
	import transformers




def 	lowerCamelCase_							(       UpperCAmelCase__   ):



						"""simple docstring"""




						random.seed(UpperCamelCase__   )
						np.random.seed(UpperCamelCase__   )
						torch.manual_seed(UpperCamelCase__   )
						torch.cuda.manual_seed_all(UpperCamelCase__   )
						# ^^ safe to call this function even if cuda is not available





class 							lowercase_					:






							"""simple docstring"""







							def __init__( self      ,     _UpperCAmelCase      ,     _UpperCAmelCase = 0.9_9_9_9      ,     _UpperCAmelCase = 0.0      ,     _UpperCAmelCase = 0      ,     _UpperCAmelCase = False      ,     _UpperCAmelCase = 1.0      ,     _UpperCAmelCase = 2 / 3      ,     _UpperCAmelCase = None      ,     _UpperCAmelCase = None      ,     **_UpperCAmelCase      ,     ):







													"""simple docstring"""


													if isinstance(__a      ,     torch.nn.Module		):
																			a_		      =		(
																			    """Passing a `torch.nn.Module` to `ExponentialMovingAverage` is deprecated. """
																			    """Please pass the parameters of the module instead."""
																			)
																			deprecate(
																			    """passing a `torch.nn.Module` to `ExponentialMovingAverage`"""      ,     """1.0.0"""      ,     __a      ,     standard_warn=__a      ,     )
																			a_		      =		parameters.parameters()

																			# set use_ema_warmup to True if a torch.nn.Module is passed for backwards compatibility
																			a_		      =		True

													if kwargs.get("""max_value"""      ,     __a		) is not None:
																			a_		      =		"""The `max_value` argument is deprecated. Please use `decay` instead."""
																			deprecate("""max_value"""      ,     """1.0.0"""      ,     __a      ,     standard_warn=__a		)
																			a_		      =		kwargs["""max_value"""]

													if kwargs.get("""min_value"""      ,     __a		) is not None:
																			a_		      =		"""The `min_value` argument is deprecated. Please use `min_decay` instead."""
																			deprecate("""min_value"""      ,     """1.0.0"""      ,     __a      ,     standard_warn=__a		)
																			a_		      =		kwargs["""min_value"""]

													a_		      =		list(__a		)
													a_		      =		[p.clone().detach() for p in parameters]

													if kwargs.get("""device"""      ,     __a		) is not None:
																			a_		      =		"""The `device` argument is deprecated. Please use `to` instead."""
																			deprecate("""device"""      ,     """1.0.0"""      ,     __a      ,     standard_warn=__a		)
																			self.to(device=kwargs["""device"""]		)

													a_		      =		None

													a_		      =		decay
													a_		      =		min_decay
													a_		      =		update_after_step
													a_		      =		use_ema_warmup
													a_		      =		inv_gamma
													a_		      =		power
													a_		      =		0
													a_		      =		None  # set in `step()`

													a_		      =		model_cls
													a_		      =		model_config


							@classmethod
							def     lowercase__  ( cls      ,     _UpperCAmelCase      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_      , a_		      =		model_cls.load_config(__a      ,     return_unused_kwargs=__a		)
													a_		      =		model_cls.from_pretrained(__a		)

													a_		      =		cls(model.parameters()      ,     model_cls=__a      ,     model_config=model.config		)

													ema_model.load_state_dict(__a		)
													return ema_model


							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													if self.model_cls is None:
																			raise ValueError("""`save_pretrained` can only be used if `model_cls` was defined at __init__."""		)

													if self.model_config is None:
																			raise ValueError("""`save_pretrained` can only be used if `model_config` was defined at __init__."""		)

													a_		      =		self.model_cls.from_config(self.model_config		)
													a_		      =		self.state_dict()
													state_dict.pop("""shadow_params"""      ,     __a		)

													model.register_to_config(**__a		)
													self.copy_to(model.parameters()		)
													model.save_pretrained(__a		)


							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_		      =		max(0      ,     optimization_step - self.update_after_step - 1		)

													if step <= 0:
																			return 0.0

													if self.use_ema_warmup:
																			a_		      =		1 - (1 + step / self.inv_gamma) ** -self.power
													else:
																			a_		      =		(1 + step) / (10 + step)

													a_		      =		min(__a      ,     self.decay		)
													# make sure decay is not smaller than min_decay
													a_		      =		max(__a      ,     self.min_decay		)
													return cur_decay_value


							@torch.no_grad()
							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													if isinstance(__a      ,     torch.nn.Module		):
																			a_		      =		(
																			    """Passing a `torch.nn.Module` to `ExponentialMovingAverage.step` is deprecated. """
																			    """Please pass the parameters of the module instead."""
																			)
																			deprecate(
																			    """passing a `torch.nn.Module` to `ExponentialMovingAverage.step`"""      ,     """1.0.0"""      ,     __a      ,     standard_warn=__a      ,     )
																			a_		      =		parameters.parameters()

													a_		      =		list(__a		)

													self.optimization_step += 1

													# Compute the decay factor for the exponential moving average.
													a_		      =		self.get_decay(self.optimization_step		)
													a_		      =		decay
													a_		      =		1 - decay

													a_		      =		contextlib.nullcontext
													if is_transformers_available() and transformers.deepspeed.is_deepspeed_zeroa_enabled():
																			import deepspeed

													for s_param, param in zip(self.shadow_params      ,     __a		):
																			if is_transformers_available() and transformers.deepspeed.is_deepspeed_zeroa_enabled():
																									a_		      =		deepspeed.zero.GatheredParameters(__a      ,     modifier_rank=__a		)

																			with context_manager():
																									if param.requires_grad:
																															s_param.sub_(one_minus_decay * (s_param - param)		)
																									else:
																															s_param.copy_(__a		)


							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_		      =		list(__a		)
													for s_param, param in zip(self.shadow_params      ,     __a		):
																			param.data.copy_(s_param.to(param.device		).data		)


							def     lowercase__  ( self      ,     _UpperCAmelCase=None      ,     _UpperCAmelCase=None		):







													"""simple docstring"""


													a_		      =		[
													    p.to(device=__a      ,     dtype=__a		) if p.is_floating_point() else p.to(device=__a		)
													    for p in self.shadow_params
													]


							def     lowercase__  ( self		):







													"""simple docstring"""


													return {
													    "decay": self.decay,
													    "min_decay": self.min_decay,
													    "optimization_step": self.optimization_step,
													    "update_after_step": self.update_after_step,
													    "use_ema_warmup": self.use_ema_warmup,
													    "inv_gamma": self.inv_gamma,
													    "power": self.power,
													    "shadow_params": self.shadow_params,
													}


							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_		      =		[param.detach().cpu().clone() for param in parameters]


							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													if self.temp_stored_params is None:
																			raise RuntimeError("""This ExponentialMovingAverage has no `store()`ed weights """ """to `restore()`"""		)
													for c_param, param in zip(self.temp_stored_params      ,     __a		):
																			param.data.copy_(c_param.data		)

													# Better memory-wise.
													a_		      =		None






							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_		      =		copy.deepcopy(__a		)

													a_		      =		state_dict.get("""decay"""      ,     self.decay		)
													if self.decay < 0.0 or self.decay > 1.0:
																			raise ValueError("""Decay must be between 0 and 1"""		)

													a_		      =		state_dict.get("""min_decay"""      ,     self.min_decay		)
													if not isinstance(self.min_decay      ,     __a		):
																			raise ValueError("""Invalid min_decay"""		)

													a_		      =		state_dict.get("""optimization_step"""      ,     self.optimization_step		)
													if not isinstance(self.optimization_step      ,     __a		):
																			raise ValueError("""Invalid optimization_step"""		)

													a_		      =		state_dict.get("""update_after_step"""      ,     self.update_after_step		)
													if not isinstance(self.update_after_step      ,     __a		):
																			raise ValueError("""Invalid update_after_step"""		)

													a_		      =		state_dict.get("""use_ema_warmup"""      ,     self.use_ema_warmup		)
													if not isinstance(self.use_ema_warmup      ,     __a		):
																			raise ValueError("""Invalid use_ema_warmup"""		)

													a_		      =		state_dict.get("""inv_gamma"""      ,     self.inv_gamma		)
													if not isinstance(self.inv_gamma      ,     (float, int)		):
																			raise ValueError("""Invalid inv_gamma"""		)

													a_		      =		state_dict.get("""power"""      ,     self.power		)
													if not isinstance(self.power      ,     (float, int)		):
																			raise ValueError("""Invalid power"""		)

													a_		      =		state_dict.get("""shadow_params"""      ,     __a		)
													if shadow_params is not None:
																			a_		      =		shadow_params
																			if not isinstance(self.shadow_params      ,     __a		):
																									raise ValueError("""shadow_params must be a list"""		)
																			if not all(isinstance(__a      ,     torch.Tensor		) for p in self.shadow_params		):
																									raise ValueError("""shadow_params must all be Tensors"""		)
'''simple docstring'''


import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

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


if TYPE_CHECKING:
					from transformers.pipelines.conversational import Conversation

__lowerCAmelCase						:							Dict    							=						logging.get_logger(__name__)


__lowerCAmelCase						:							Optional[int]    							=						{
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
    "tokenizer_config_file": "tokenizer_config.json",
}

__lowerCAmelCase						:							int    							=						{
    "vocab_file": {"facebook/blenderbot-3B": "https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json"},
    "merges_file": {"facebook/blenderbot-3B": "https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt"},
    "tokenizer_config_file": {
        "facebook/blenderbot-3B": "https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json"
    },
}

__lowerCAmelCase						:							List[str]    							=						{"facebook/blenderbot-3B": 128}





@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def    lowerCAmelCase       (		):


	"""simple docstring"""
	__UpperCAmelCase      					=      (
	    list(range(ord('''!'''							)     ,       ord('''~'''							) + 1							)							) + list(range(ord('''¡'''							)     ,       ord('''¬'''							) + 1							)							) + list(range(ord('''®'''							)     ,       ord('''ÿ'''							) + 1							)							)
	)
	__UpperCAmelCase      					=      bs[:]
	__UpperCAmelCase      					=      0
	for b in range(2**8							):
		if b not in bs:
			bs.append(UpperCamelCase__							)
			cs.append(2**8 + n							)
			n += 1
	__UpperCAmelCase      					=      [chr(UpperCamelCase__							) for n in cs]
	return dict(zip(UpperCamelCase__     ,       UpperCamelCase__							)							)





def    lowerCAmelCase       (		UpperCamelCase__					: int							):


	"""simple docstring"""
	__UpperCAmelCase      					=      set()
	__UpperCAmelCase      					=      word[0]
	for char in word[1:]:
		pairs.add((prev_char, char)							)
		__UpperCAmelCase      					=      char
	return pairs






class     A ( UpperCAmelCase ):
							a_										=   VOCAB_FILES_NAMES
							a_										=   PRETRAINED_VOCAB_FILES_MAP
							a_										=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							a_										=   ['''input_ids''', '''attention_mask''']





							def __init__(   self					:				str  ,		__a					:				Union[str, Any]  ,		__a					:				Optional[Any]  ,		__a					:				List[Any]="replace"  ,		__a					:				Union[str, Any]="<s>"  ,		__a					:				Any="</s>"  ,		__a					:				Dict="</s>"  ,		__a					:				Dict="<s>"  ,		__a					:				Tuple="<unk>"  ,		__a					:				List[str]="<pad>"  ,		__a					:				Any="<mask>"  ,		__a					:				Dict=False  ,		**__a					:				Union[str, Any]  ,		)				->						Optional[int]:
								__UpperCAmelCase      					=      AddedToken(__a  ,		lstrip=__a  ,		rstrip=__a						) if isinstance(__a  ,		__a						) else bos_token
								__UpperCAmelCase      					=      AddedToken(__a  ,		lstrip=__a  ,		rstrip=__a						) if isinstance(__a  ,		__a						) else eos_token
								__UpperCAmelCase      					=      AddedToken(__a  ,		lstrip=__a  ,		rstrip=__a						) if isinstance(__a  ,		__a						) else sep_token
								__UpperCAmelCase      					=      AddedToken(__a  ,		lstrip=__a  ,		rstrip=__a						) if isinstance(__a  ,		__a						) else cls_token
								__UpperCAmelCase      					=      AddedToken(__a  ,		lstrip=__a  ,		rstrip=__a						) if isinstance(__a  ,		__a						) else unk_token
								__UpperCAmelCase      					=      AddedToken(__a  ,		lstrip=__a  ,		rstrip=__a						) if isinstance(__a  ,		__a						) else pad_token

								# Mask token behave like a normal word, i.e. include the space before it
								__UpperCAmelCase      					=      AddedToken(__a  ,		lstrip=__a  ,		rstrip=__a						) if isinstance(__a  ,		__a						) else mask_token

								super().__init__(
								    errors=__a  ,		bos_token=__a  ,		eos_token=__a  ,		unk_token=__a  ,		sep_token=__a  ,		cls_token=__a  ,		pad_token=__a  ,		mask_token=__a  ,		add_prefix_space=__a  ,		**__a  ,		)

								with open(__a  ,		encoding='''utf-8'''						) as vocab_handle:
									__UpperCAmelCase      					=      json.load(__a						)
								__UpperCAmelCase      					=      {v: k for k, v in self.encoder.items()}
								__UpperCAmelCase      					=      errors  # how to handle errors in decoding
								__UpperCAmelCase      					=      bytes_to_unicode()
								__UpperCAmelCase      					=      {v: k for k, v in self.byte_encoder.items()}
								with open(__a  ,		encoding='''utf-8'''						) as merges_handle:
									__UpperCAmelCase      					=      merges_handle.read().split('''\n'''						)[1:-1]
								__UpperCAmelCase      					=      [tuple(merge.split()						) for merge in bpe_merges]
								__UpperCAmelCase      					=      dict(zip(__a  ,		range(len(__a						)						)						)						)
								__UpperCAmelCase      					=      {}
								__UpperCAmelCase      					=      add_prefix_space

								# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
								__UpperCAmelCase      					=      re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''						)





							@property
							# Copied from transformers.models.roberta.tokenization_roberta.RobertaTokenizer.vocab_size with Roberta->Blenderbot, RoBERTa->Blenderbot
							def    snake_case__    (   self					:				List[Any]						)				->						Union[str, Any]:
								return len(self.encoder						)





							def    snake_case__    (   self					:				str						)				->						int:
								return dict(self.encoder  ,		**self.added_tokens_encoder						)





							def    snake_case__    (   self					:				List[Any]  ,		__a					:				Tuple						)				->						List[Any]:
								if token in self.cache:
									return self.cache[token]
								__UpperCAmelCase      					=      tuple(__a						)
								__UpperCAmelCase      					=      get_pairs(__a						)

								if not pairs:
									return token

								while True:
									__UpperCAmelCase      					=      min(__a  ,		key=lambda __a						: self.bpe_ranks.get(__a  ,		float('''inf'''						)						)						)
									if bigram not in self.bpe_ranks:
										break
									__UpperCAmelCase					,      __UpperCAmelCase      					=      bigram
									__UpperCAmelCase      					=      []
									__UpperCAmelCase      					=      0
									while i < len(__a						):
										try:
											__UpperCAmelCase      					=      word.index(__a  ,		__a						)
										except ValueError:
											new_word.extend(word[i:]						)
											break
										else:
											new_word.extend(word[i:j]						)
											__UpperCAmelCase      					=      j

										if word[i] == first and i < len(__a						) - 1 and word[i + 1] == second:
											new_word.append(first + second						)
											i += 2
										else:
											new_word.append(word[i]						)
											i += 1
									__UpperCAmelCase      					=      tuple(__a						)
									__UpperCAmelCase      					=      new_word
									if len(__a						) == 1:
										break
									else:
										__UpperCAmelCase      					=      get_pairs(__a						)
								__UpperCAmelCase      					=      ''' '''.join(__a						)
								__UpperCAmelCase      					=      word
								return word





							def    snake_case__    (   self					:				int  ,		__a					:				int						)				->						List[Any]:
								__UpperCAmelCase      					=      []
								for token in re.findall(self.pat  ,		__a						):
									__UpperCAmelCase      					=      ''''''.join(
									    self.byte_encoder[b] for b in token.encode('''utf-8'''						)						)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
									bpe_tokens.extend(bpe_token for bpe_token in self.bpe(__a						).split(''' '''						)						)
								return bpe_tokens





							def    snake_case__    (   self					:				Optional[Any]  ,		__a					:				Tuple						)				->						str:
								return self.encoder.get(__a  ,		self.encoder.get(self.unk_token						)						)





							def    snake_case__    (   self					:				Optional[int]  ,		__a					:				Any						)				->						List[str]:
								return self.decoder.get(__a						)





							def    snake_case__    (   self					:				Union[str, Any]  ,		__a					:				List[str]						)				->						List[Any]:
								__UpperCAmelCase      					=      ''''''.join(__a						)
								__UpperCAmelCase      					=      bytearray([self.byte_decoder[c] for c in text]						).decode('''utf-8'''  ,		errors=self.errors						)
								return text





							def    snake_case__    (   self					:				Union[str, Any]  ,		__a					:				str  ,		__a					:				Optional[str] = None						)				->						Tuple[str]:
								if not os.path.isdir(__a						):
									logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""						)
									return
								__UpperCAmelCase      					=      os.path.join(
								    __a  ,		(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']						)
								__UpperCAmelCase      					=      os.path.join(
								    __a  ,		(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']						)

								with open(__a  ,		'''w'''  ,		encoding='''utf-8'''						) as f:
									f.write(json.dumps(self.encoder  ,		indent=2  ,		sort_keys=__a  ,		ensure_ascii=__a						) + '''\n'''						)

								__UpperCAmelCase      					=      0
								with open(__a  ,		'''w'''  ,		encoding='''utf-8'''						) as writer:
									writer.write('''#version: 0.2\n'''						)
									for bpe_tokens, token_index in sorted(self.bpe_ranks.items()  ,		key=lambda __a						: kv[1]						):
										if index != token_index:
											logger.warning(
											    f"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
											    ''' Please check that the tokenizer is not corrupted!'''						)
											__UpperCAmelCase      					=      token_index
										writer.write(''' '''.join(__a						) + '''\n'''						)
										index += 1

								return vocab_file, merge_file





							def    snake_case__    (   self					:				Union[str, Any]  ,		__a					:				List[int]  ,		__a					:				Optional[List[int]] = None  ,		__a					:				bool = False						)				->						List[int]:
								if already_has_special_tokens:
									return super().get_special_tokens_mask(
									    token_ids_a=__a  ,		token_ids_a=__a  ,		already_has_special_tokens=__a						)

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





							def    snake_case__    (   self					:				Dict  ,		__a					:				List[int]  ,		__a					:				Optional[List[int]] = None						)				->						List[int]:
								__UpperCAmelCase      					=      [self.sep_token_id]
								__UpperCAmelCase      					=      [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]





							def    snake_case__    (   self					:				int  ,		__a					:				Optional[int]  ,		__a					:				int=False  ,		**__a					:				Union[str, Any]						)				->						Union[str, Any]:
								__UpperCAmelCase      					=      kwargs.pop('''add_prefix_space'''  ,		self.add_prefix_space						)
								if (is_split_into_words or add_prefix_space) and (len(__a						) > 0 and not text[0].isspace()):
									__UpperCAmelCase      					=      ''' ''' + text
								return (text, kwargs)





							def    snake_case__    (   self					:				List[str]  ,		__a					:				List[int]  ,		__a					:				Optional[List[int]] = None						)				->						Dict:
								return token_ids_a + [self.eos_token_id]






							def    snake_case__    (   self					:				Optional[Any]  ,		__a					:				"Conversation"						)				->						List[int]:
								__UpperCAmelCase      					=      []
								for is_user, text in conversation.iter_texts():
									if is_user:
										# We need to space prefix as it's being done within blenderbot
										inputs.append(''' ''' + text						)
									else:
										# Generated responses should contain them already.
										inputs.append(__a						)

								__UpperCAmelCase      					=      '''  '''.join(__a						)
								__UpperCAmelCase      					=      self.encode(__a						)
								if len(__a						) > self.model_max_length:
									__UpperCAmelCase      					=      input_ids[-self.model_max_length :]
									logger.warning(f"""Trimmed input from conversation as it was longer than {self.model_max_length} tokens."""						)
								return input_ids
