
"""simple docstring"""






from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block

@dataclass
class 					__a     ( __a       ):



							'''simple docstring'''







							_lowerCamelCase					:			torch.FloatTensor





class 					__a     ( nn.Module       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase=3  ,    _lowerCamelCase=3  ,    _lowerCamelCase=("DownEncoderBlock2D",)  ,    _lowerCamelCase=(64,)  ,    _lowerCamelCase=2  ,    _lowerCamelCase=32  ,    _lowerCamelCase="silu"  ,    _lowerCamelCase=True  ,    )			->							int:




										'''simple docstring'''
										super().__init__()
										__lowercase       =       layers_per_block

										__lowercase       =       torch.nn.Convad(
										    _lowerCamelCase  ,    block_out_channels[0]  ,    kernel_size=3  ,    stride=1  ,    padding=1  ,    )

										__lowercase       =       None
										__lowercase       =       nn.ModuleList([]       )

										# down
										__lowercase       =       block_out_channels[0]
										for i, down_block_type in enumerate(_lowerCamelCase       ):
													__lowercase       =       output_channel
													__lowercase       =       block_out_channels[i]
													__lowercase       =       i == len(_lowerCamelCase       ) - 1

													__lowercase       =       get_down_block(
													    _lowerCamelCase  ,    num_layers=self.layers_per_block  ,    in_channels=_lowerCamelCase  ,    out_channels=_lowerCamelCase  ,    add_downsample=not is_final_block  ,    resnet_eps=1e-6  ,    downsample_padding=0  ,    resnet_act_fn=_lowerCamelCase  ,    resnet_groups=_lowerCamelCase  ,    attention_head_dim=_lowerCamelCase  ,    temb_channels=_lowerCamelCase  ,    )
													self.down_blocks.append(_lowerCamelCase       )

										# mid
										__lowercase       =       UNetMidBlockaD(
										    in_channels=block_out_channels[-1]  ,    resnet_eps=1e-6  ,    resnet_act_fn=_lowerCamelCase  ,    output_scale_factor=1  ,    resnet_time_scale_shift="default"  ,    attention_head_dim=block_out_channels[-1]  ,    resnet_groups=_lowerCamelCase  ,    temb_channels=_lowerCamelCase  ,    )

										# out
										__lowercase       =       nn.GroupNorm(num_channels=block_out_channels[-1]  ,    num_groups=_lowerCamelCase  ,    eps=1e-6       )
										__lowercase       =       nn.SiLU()

										__lowercase       =       2 * out_channels if double_z else out_channels
										__lowercase       =       nn.Convad(block_out_channels[-1]  ,    _lowerCamelCase  ,    3  ,    padding=1       )

										__lowercase       =       False





							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							Union[str, Any]:




										'''simple docstring'''
										__lowercase       =       x
										__lowercase       =       self.conv_in(_lowerCamelCase       )

										if self.training and self.gradient_checkpointing:

													def create_custom_forward(_lowerCamelCase       ):
																def custom_forward(*_lowerCamelCase       ):
																			return module(*_lowerCamelCase       )

																return custom_forward

													# down
													if is_torch_version(">="  ,    "1.11.0"       ):
																for down_block in self.down_blocks:
																			__lowercase       =       torch.utils.checkpoint.checkpoint(
																			    create_custom_forward(_lowerCamelCase       )  ,    _lowerCamelCase  ,    use_reentrant=_lowerCamelCase       )
																# middle
																__lowercase       =       torch.utils.checkpoint.checkpoint(
																    create_custom_forward(self.mid_block       )  ,    _lowerCamelCase  ,    use_reentrant=_lowerCamelCase       )
													else:
																for down_block in self.down_blocks:
																			__lowercase       =       torch.utils.checkpoint.checkpoint(create_custom_forward(_lowerCamelCase       )  ,    _lowerCamelCase       )
																# middle
																__lowercase       =       torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block       )  ,    _lowerCamelCase       )

										else:
													# down
													for down_block in self.down_blocks:
																__lowercase       =       down_block(_lowerCamelCase       )

													# middle
													__lowercase       =       self.mid_block(_lowerCamelCase       )

										# post-process
										__lowercase       =       self.conv_norm_out(_lowerCamelCase       )
										__lowercase       =       self.conv_act(_lowerCamelCase       )
										__lowercase       =       self.conv_out(_lowerCamelCase       )

										return sample





class 					__a     ( nn.Module       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase=3  ,    _lowerCamelCase=3  ,    _lowerCamelCase=("UpDecoderBlock2D",)  ,    _lowerCamelCase=(64,)  ,    _lowerCamelCase=2  ,    _lowerCamelCase=32  ,    _lowerCamelCase="silu"  ,    _lowerCamelCase="group"  ,    )			->							Dict:




										'''simple docstring'''
										super().__init__()
										__lowercase       =       layers_per_block

										__lowercase       =       nn.Convad(
										    _lowerCamelCase  ,    block_out_channels[-1]  ,    kernel_size=3  ,    stride=1  ,    padding=1  ,    )

										__lowercase       =       None
										__lowercase       =       nn.ModuleList([]       )

										__lowercase       =       in_channels if norm_type == "spatial" else None

										# mid
										__lowercase       =       UNetMidBlockaD(
										    in_channels=block_out_channels[-1]  ,    resnet_eps=1e-6  ,    resnet_act_fn=_lowerCamelCase  ,    output_scale_factor=1  ,    resnet_time_scale_shift="default" if norm_type == "group" else norm_type  ,    attention_head_dim=block_out_channels[-1]  ,    resnet_groups=_lowerCamelCase  ,    temb_channels=_lowerCamelCase  ,    )

										# up
										__lowercase       =       list(reversed(_lowerCamelCase       )       )
										__lowercase       =       reversed_block_out_channels[0]
										for i, up_block_type in enumerate(_lowerCamelCase       ):
													__lowercase       =       output_channel
													__lowercase       =       reversed_block_out_channels[i]

													__lowercase       =       i == len(_lowerCamelCase       ) - 1

													__lowercase       =       get_up_block(
													    _lowerCamelCase  ,    num_layers=self.layers_per_block + 1  ,    in_channels=_lowerCamelCase  ,    out_channels=_lowerCamelCase  ,    prev_output_channel=_lowerCamelCase  ,    add_upsample=not is_final_block  ,    resnet_eps=1e-6  ,    resnet_act_fn=_lowerCamelCase  ,    resnet_groups=_lowerCamelCase  ,    attention_head_dim=_lowerCamelCase  ,    temb_channels=_lowerCamelCase  ,    resnet_time_scale_shift=_lowerCamelCase  ,    )
													self.up_blocks.append(_lowerCamelCase       )
													__lowercase       =       output_channel

										# out
										if norm_type == "spatial":
													__lowercase       =       SpatialNorm(block_out_channels[0]  ,    _lowerCamelCase       )
										else:
													__lowercase       =       nn.GroupNorm(num_channels=block_out_channels[0]  ,    num_groups=_lowerCamelCase  ,    eps=1e-6       )
										__lowercase       =       nn.SiLU()
										__lowercase       =       nn.Convad(block_out_channels[0]  ,    _lowerCamelCase  ,    3  ,    padding=1       )

										__lowercase       =       False





							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase=None       )			->							str:




										'''simple docstring'''
										__lowercase       =       z
										__lowercase       =       self.conv_in(_lowerCamelCase       )

										__lowercase       =       next(iter(self.up_blocks.parameters()       )       ).dtype
										if self.training and self.gradient_checkpointing:

													def create_custom_forward(_lowerCamelCase       ):
																def custom_forward(*_lowerCamelCase       ):
																			return module(*_lowerCamelCase       )

																return custom_forward

													if is_torch_version(">="  ,    "1.11.0"       ):
																# middle
																__lowercase       =       torch.utils.checkpoint.checkpoint(
																    create_custom_forward(self.mid_block       )  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    use_reentrant=_lowerCamelCase       )
																__lowercase       =       sample.to(_lowerCamelCase       )

																# up
																for up_block in self.up_blocks:
																			__lowercase       =       torch.utils.checkpoint.checkpoint(
																			    create_custom_forward(_lowerCamelCase       )  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    use_reentrant=_lowerCamelCase       )
													else:
																# middle
																__lowercase       =       torch.utils.checkpoint.checkpoint(
																    create_custom_forward(self.mid_block       )  ,    _lowerCamelCase  ,    _lowerCamelCase       )
																__lowercase       =       sample.to(_lowerCamelCase       )

																# up
																for up_block in self.up_blocks:
																			__lowercase       =       torch.utils.checkpoint.checkpoint(create_custom_forward(_lowerCamelCase       )  ,    _lowerCamelCase  ,    _lowerCamelCase       )
										else:
													# middle
													__lowercase       =       self.mid_block(_lowerCamelCase  ,    _lowerCamelCase       )
													__lowercase       =       sample.to(_lowerCamelCase       )

													# up
													for up_block in self.up_blocks:
																__lowercase       =       up_block(_lowerCamelCase  ,    _lowerCamelCase       )

        # post-process
										if latent_embeds is None:
													__lowercase       =       self.conv_norm_out(_lowerCamelCase       )
										else:
													__lowercase       =       self.conv_norm_out(_lowerCamelCase  ,    _lowerCamelCase       )
										__lowercase       =       self.conv_act(_lowerCamelCase       )
										__lowercase       =       self.conv_out(_lowerCamelCase       )

										return sample





class 					__a     ( nn.Module       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase=None  ,    _lowerCamelCase="random"  ,    _lowerCamelCase=False  ,    _lowerCamelCase=True       )			->							Tuple:




										'''simple docstring'''
										super().__init__()
										__lowercase       =       n_e
										__lowercase       =       vq_embed_dim
										__lowercase       =       beta
										__lowercase       =       legacy

										__lowercase       =       nn.Embedding(self.n_e  ,    self.vq_embed_dim       )
										self.embedding.weight.data.uniform_(-1.0 / self.n_e  ,    1.0 / self.n_e       )

										__lowercase       =       remap
										if self.remap is not None:
													self.register_buffer("used"  ,    torch.tensor(np.load(self.remap       )       )       )
													__lowercase       =       self.used.shape[0]
													__lowercase       =       unknown_index  # "random" or "extra" or integer
													if self.unknown_index == "extra":
																__lowercase       =       self.re_embed
																__lowercase       =       self.re_embed + 1
													print(
													    f'''Remapping {self.n_e} indices to {self.re_embed} indices. '''
													    f'''Using {self.unknown_index} for unknown indices.'''       )
										else:
													__lowercase       =       n_e

										__lowercase       =       sane_index_shape
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							Dict:




										'''simple docstring'''
										__lowercase       =       inds.shape
										assert len(_lowerCamelCase       ) > 1
										__lowercase       =       inds.reshape(ishape[0]  ,    -1       )
										__lowercase       =       self.used.to(_lowerCamelCase       )
										__lowercase       =       (inds[:, :, None] == used[None, None, ...]).long()
										__lowercase       =       match.argmax(-1       )
										__lowercase       =       match.sum(2       ) < 1
										if self.unknown_index == "random":
													__lowercase       =       torch.randint(0  ,    self.re_embed  ,    size=new[unknown].shape       ).to(device=new.device       )
										else:
													__lowercase       =       self.unknown_index
										return new.reshape(_lowerCamelCase       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							Tuple:




										'''simple docstring'''
										__lowercase       =       inds.shape
										assert len(_lowerCamelCase       ) > 1
										__lowercase       =       inds.reshape(ishape[0]  ,    -1       )
										__lowercase       =       self.used.to(_lowerCamelCase       )
										if self.re_embed > self.used.shape[0]:  # extra token
													__lowercase       =       0  # simply set to zero
										__lowercase       =       torch.gather(used[None, :][inds.shape[0] * [0], :]  ,    1  ,    _lowerCamelCase       )
										return back.reshape(_lowerCamelCase       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							List[Any]:




										'''simple docstring'''
										# reshape z -> (batch, height, width, channel) and flatten
										__lowercase       =       z.permute(0  ,    2  ,    3  ,    1       ).contiguous()
										__lowercase       =       z.view(-1  ,    self.vq_embed_dim       )

										# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
										__lowercase       =       torch.argmin(torch.cdist(_lowerCamelCase  ,    self.embedding.weight       )  ,    dim=1       )

										__lowercase       =       self.embedding(_lowerCamelCase       ).view(z.shape       )
										__lowercase       =       None
										__lowercase       =       None

										# compute loss for embedding
										if not self.legacy:
													__lowercase       =       self.beta * torch.mean((z_q.detach() - z) ** 2       ) + torch.mean((z_q - z.detach()) ** 2       )
										else:
													__lowercase       =       torch.mean((z_q.detach() - z) ** 2       ) + self.beta * torch.mean((z_q - z.detach()) ** 2       )

										# preserve gradients
										__lowercase       =       z + (z_q - z).detach()

										# reshape back to match original input shape
										__lowercase       =       z_q.permute(0  ,    3  ,    1  ,    2       ).contiguous()

										if self.remap is not None:
													__lowercase       =       min_encoding_indices.reshape(z.shape[0]  ,    -1       )  # add batch axis
													__lowercase       =       self.remap_to_used(_lowerCamelCase       )
													__lowercase       =       min_encoding_indices.reshape(-1  ,    1       )  # flatten

										if self.sane_index_shape:
													__lowercase       =       min_encoding_indices.reshape(z_q.shape[0]  ,    z_q.shape[2]  ,    z_q.shape[3]       )

										return z_q, loss, (perplexity, min_encodings, min_encoding_indices)





							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase       )			->							Union[str, Any]:




										'''simple docstring'''
										# shape specifying (batch, height, width, channel)
										if self.remap is not None:
													__lowercase       =       indices.reshape(shape[0]  ,    -1       )  # add batch axis
													__lowercase       =       self.unmap_to_all(_lowerCamelCase       )
													__lowercase       =       indices.reshape(-1       )  # flatten again

										# get quantized latent vectors
										__lowercase       =       self.embedding(_lowerCamelCase       )

										if shape is not None:
													__lowercase       =       z_q.view(_lowerCamelCase       )
													# reshape back to match original input shape
													__lowercase       =       z_q.permute(0  ,    3  ,    1  ,    2       ).contiguous()

										return z_q





class 					__a     ( __a       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase=False       )			->							str:




										'''simple docstring'''
										__lowercase       =       parameters
										__lowercase   ,					__lowercase       =       torch.chunk(_lowerCamelCase  ,    2  ,    dim=1       )
										__lowercase       =       torch.clamp(self.logvar  ,    -30.0  ,    20.0       )
										__lowercase       =       deterministic
										__lowercase       =       torch.exp(0.5 * self.logvar       )
										__lowercase       =       torch.exp(self.logvar       )
										if self.deterministic:
													__lowercase       =       __lowercase       =       torch.zeros_like(
													    self.mean  ,    device=self.parameters.device  ,    dtype=self.parameters.dtype       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase = None       )			->							torch.FloatTensor:




										'''simple docstring'''
										# make sure sample is on the same device as the parameters and has same dtype
										__lowercase       =       randn_tensor(
										    self.mean.shape  ,    generator=_lowerCamelCase  ,    device=self.parameters.device  ,    dtype=self.parameters.dtype       )
										__lowercase       =       self.mean + self.std * sample
										return x
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase=None       )			->							str:




										'''simple docstring'''
										if self.deterministic:
													return torch.Tensor([0.0]       )
										else:
													if other is None:
																return 0.5 * torch.sum(torch.pow(self.mean  ,    2       ) + self.var - 1.0 - self.logvar  ,    dim=[1, 2, 3]       )
													else:
																return 0.5 * torch.sum(
																    torch.pow(self.mean - other.mean  ,    2       ) / other.var
																    + self.var / other.var
																    - 1.0
																    - self.logvar
																    + other.logvar  ,    dim=[1, 2, 3]  ,    )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase=[1, 2, 3]       )			->							List[Any]:




										'''simple docstring'''
										if self.deterministic:
													return torch.Tensor([0.0]       )
										__lowercase       =       np.log(2.0 * np.pi       )
										return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean  ,    2       ) / self.var  ,    dim=_lowerCamelCase       )





							def   SCREAMING_SNAKE_CASE						(     self       )			->							Optional[Any]:




										'''simple docstring'''
										return self.mean







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

_lowercase							=				logging.get_logger(__name__)


_lowercase							=				{
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
}

_lowercase							=				{
    '''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'''
    },
}

_lowercase							=				{'''facebook/blenderbot-3B''': 128}



@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def      lowerCAmelCase__			(		)						->Any:
			__lowercase       =       (
			    list(range(ord("!"      )			,   ord("~"      ) + 1      )      ) + list(range(ord("¡"      )			,   ord("¬"      ) + 1      )      ) + list(range(ord("®"      )			,   ord("ÿ"      ) + 1      )      )
			)
			__lowercase       =       bs[:]
			__lowercase       =       0
			for b in range(2**8      ):
						if b not in bs:
									bs.append(__magic_name__      )
									cs.append(2**8 + n      )
									n += 1
			__lowercase       =       [chr(__magic_name__      ) for n in cs]
			return dict(zip(__magic_name__			,   __magic_name__      )      )



def      lowerCAmelCase__			(		__magic_name__      )						->Tuple:
			__lowercase       =       set()
			__lowercase       =       word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)      )
						__lowercase       =       char
			return pairs







class 					__a     ( __a       ):



							'''simple docstring'''







							_lowerCamelCase					:			List[str]          =   VOCAB_FILES_NAMES
							_lowerCamelCase					:			Tuple          =   PRETRAINED_VOCAB_FILES_MAP
							_lowerCamelCase					:			int          =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							_lowerCamelCase					:			Dict          =   ["""input_ids""", """attention_mask"""]
							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase="replace"  ,    _lowerCamelCase="<s>"  ,    _lowerCamelCase="</s>"  ,    _lowerCamelCase="</s>"  ,    _lowerCamelCase="<s>"  ,    _lowerCamelCase="<unk>"  ,    _lowerCamelCase="<pad>"  ,    _lowerCamelCase="<mask>"  ,    _lowerCamelCase=False  ,    **_lowerCamelCase  ,    )			->							int:




										'''simple docstring'''
										__lowercase       =       AddedToken(_lowerCamelCase  ,    lstrip=_lowerCamelCase  ,    rstrip=_lowerCamelCase       ) if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ) else bos_token
										__lowercase       =       AddedToken(_lowerCamelCase  ,    lstrip=_lowerCamelCase  ,    rstrip=_lowerCamelCase       ) if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ) else eos_token
										__lowercase       =       AddedToken(_lowerCamelCase  ,    lstrip=_lowerCamelCase  ,    rstrip=_lowerCamelCase       ) if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ) else sep_token
										__lowercase       =       AddedToken(_lowerCamelCase  ,    lstrip=_lowerCamelCase  ,    rstrip=_lowerCamelCase       ) if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ) else cls_token
										__lowercase       =       AddedToken(_lowerCamelCase  ,    lstrip=_lowerCamelCase  ,    rstrip=_lowerCamelCase       ) if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ) else unk_token
										__lowercase       =       AddedToken(_lowerCamelCase  ,    lstrip=_lowerCamelCase  ,    rstrip=_lowerCamelCase       ) if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ) else pad_token

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

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

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

										# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
										__lowercase       =       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   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										return len(self.encoder       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										return dict(self.encoder  ,    **self.added_tokens_encoder       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							int:




										'''simple docstring'''
										if token in self.cache:
													return self.cache[token]
										__lowercase       =       tuple(_lowerCamelCase       )
										__lowercase       =       get_pairs(_lowerCamelCase       )

										if not pairs:
													return token

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

																if word[i] == first and i < len(_lowerCamelCase       ) - 1 and word[i + 1] == second:
																			new_word.append(first + second       )
																			i += 2
																else:
																			new_word.append(word[i]       )
																			i += 1
													__lowercase       =       tuple(_lowerCamelCase       )
													__lowercase       =       new_word
													if len(_lowerCamelCase       ) == 1:
																break
													else:
																__lowercase       =       get_pairs(_lowerCamelCase       )
										__lowercase       =       " ".join(_lowerCamelCase       )
										__lowercase       =       word
										return word
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							Union[str, Any]:




										'''simple docstring'''
										__lowercase       =       []
										for token in re.findall(self.pat  ,    _lowerCamelCase       ):
													__lowercase       =       "".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(_lowerCamelCase       ).split(" "       )       )
										return bpe_tokens
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							Tuple:




										'''simple docstring'''
										return self.encoder.get(_lowerCamelCase  ,    self.encoder.get(self.unk_token       )       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							str:




										'''simple docstring'''
										return self.decoder.get(_lowerCamelCase       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							List[str]:




										'''simple docstring'''
										__lowercase       =       "".join(_lowerCamelCase       )
										__lowercase       =       bytearray([self.byte_decoder[c] for c in text]       ).decode("utf-8"  ,    errors=self.errors       )
										return text
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase = None       )			->							Tuple[str]:




										'''simple docstring'''
										if not os.path.isdir(_lowerCamelCase       ):
													logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''       )
													return
										__lowercase       =       os.path.join(
										    _lowerCamelCase  ,    (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]       )
										__lowercase       =       os.path.join(
										    _lowerCamelCase  ,    (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]       )

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

										__lowercase       =       0
										with open(_lowerCamelCase  ,    "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 _lowerCamelCase       : 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!"       )
																			__lowercase       =       token_index
																writer.write(" ".join(_lowerCamelCase       ) + "\n"       )
																index += 1

										return vocab_file, merge_file
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase = None  ,    _lowerCamelCase = False       )			->							List[int]:




										'''simple docstring'''
										if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=_lowerCamelCase  ,    token_ids_a=_lowerCamelCase  ,    already_has_special_tokens=_lowerCamelCase       )

										if token_ids_a is None:
													return [1] + ([0] * len(_lowerCamelCase       )) + [1]
										return [1] + ([0] * len(_lowerCamelCase       )) + [1, 1] + ([0] * len(_lowerCamelCase       )) + [1]
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase = None       )			->							List[int]:




										'''simple docstring'''
										__lowercase       =       [self.sep_token_id]
										__lowercase       =       [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   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase=False  ,    **_lowerCamelCase       )			->							int:




										'''simple docstring'''
										__lowercase       =       kwargs.pop("add_prefix_space"  ,    self.add_prefix_space       )
										if (is_split_into_words or add_prefix_space) and (len(_lowerCamelCase       ) > 0 and not text[0].isspace()):
													__lowercase       =       " " + text
										return (text, kwargs)
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase = None       )			->							Tuple:




										'''simple docstring'''
										return token_ids_a + [self.eos_token_id]





							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							List[int]:




										'''simple docstring'''
										__lowercase       =       []
										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(_lowerCamelCase       )

										__lowercase       =       "  ".join(_lowerCamelCase       )
										__lowercase       =       self.encode(_lowerCamelCase       )
										if len(_lowerCamelCase       ) > self.model_max_length:
													__lowercase       =       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





