






from abc import ABC, abstractmethod
from typing import List, Optional
class 							A_    (						__a			):






 '''simple docstring'''




 def __init__(	self) ->		List[str]:

     """simple docstring"""



     self.test()


 def  snake_case__				(	self) ->		Dict:

     """simple docstring"""



     _UpperCAmelCase				:      Any			  =				0
     _UpperCAmelCase				:      Any			  =				False
     while not completed:
         if counter == 1:
             self.reset()
         _UpperCAmelCase				:      List[str]			  =				self.advance()
         if not self.does_advance(A__):
             raise Exception(
                 '''Custom Constraint is not defined correctly. self.does_advance(self.advance()) must be true.''')

         _UpperCAmelCase  , _UpperCAmelCase  , _UpperCAmelCase				:      Union[str, Any]			  =				self.update(A__)
         counter += 1

         if counter > 10000:
             raise Exception('''update() does not fulfill the constraint.''')

     if self.remaining() != 0:
         raise Exception('''Custom Constraint is not defined correctly.''')


 @abstractmethod
 def  snake_case__				(	self) ->		List[str]:

     """simple docstring"""



     raise NotImplementedError(
         f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.''')


 @abstractmethod
 def  snake_case__				(	self							,  _A) ->		int:

     """simple docstring"""



     raise NotImplementedError(
         f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.''')


 @abstractmethod
 def  snake_case__				(	self							,  _A) ->		Tuple:

     """simple docstring"""



     raise NotImplementedError(
         f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.''')


 @abstractmethod
 def  snake_case__				(	self) ->		Any:

     """simple docstring"""



     raise NotImplementedError(
         f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.''')


 @abstractmethod
 def  snake_case__				(	self) ->		Tuple:

     """simple docstring"""



     raise NotImplementedError(
         f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.''')


 @abstractmethod
 def  snake_case__				(	self							,  _A=False) ->		Optional[Any]:

     """simple docstring"""



     raise NotImplementedError(
         f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.''')



class 							A_    (						__a			):






 '''simple docstring'''




 def __init__(	self							,  _A) ->		Dict:

     """simple docstring"""



     super(A__							,  self).__init__()

     if not isinstance(A__							,  A__) or len(A__) == 0:
         raise ValueError(f'''`token_ids` has to be a non-empty list, but is {token_ids}.''')
     if any((not isinstance(A__							,  A__) or token_id < 0) for token_id in token_ids):
         raise ValueError(f'''Each list in `token_ids` has to be a list of positive integers, but is {token_ids}.''')

     _UpperCAmelCase				:      str			  =				token_ids

     _UpperCAmelCase				:      str			  =				len(self.token_ids)
     _UpperCAmelCase				:      Dict			  =				-1  # the index of the currently fulfilled step
     _UpperCAmelCase				:      List[str]			  =				False


 def  snake_case__				(	self) ->		Any:

     """simple docstring"""



     if self.completed:
         return None
     return self.token_ids[self.fulfilled_idx + 1]


 def  snake_case__				(	self							,  _A) ->		Optional[int]:

     """simple docstring"""



     if not isinstance(A__							,  A__):
         raise ValueError(f'''`token_id` has to be an `int`, but is {token_id} of type {type(A__)}''')

     if self.completed:
         return False

     return token_id == self.token_ids[self.fulfilled_idx + 1]


 def  snake_case__				(	self							,  _A) ->		int:

     """simple docstring"""



     if not isinstance(A__							,  A__):
         raise ValueError(f'''`token_id` has to be an `int`, but is {token_id} of type {type(A__)}''')

     _UpperCAmelCase				:      Optional[int]			  =				False
     _UpperCAmelCase				:      Optional[Any]			  =				False
     _UpperCAmelCase				:      List[str]			  =				False

     if self.does_advance(A__):
         self.fulfilled_idx += 1
         _UpperCAmelCase				:      Optional[int]			  =				True
         if self.fulfilled_idx == (self.seqlen - 1):
             _UpperCAmelCase				:      List[str]			  =				True
         _UpperCAmelCase				:      Tuple			  =				completed
     else:
         # failed to make progress.
         _UpperCAmelCase				:      List[str]			  =				True
         self.reset()
     return stepped, completed, reset


 def  snake_case__				(	self) ->		int:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				False
     _UpperCAmelCase				:      Dict			  =				0


 def  snake_case__				(	self) ->		Tuple:

     """simple docstring"""



     return self.seqlen - (self.fulfilled_idx + 1)


 def  snake_case__				(	self							,  _A=False) ->		int:

     """simple docstring"""



     _UpperCAmelCase				:      Dict			  =				PhrasalConstraint(self.token_ids)

     if stateful:
         _UpperCAmelCase				:      Optional[int]			  =				self.seqlen
         _UpperCAmelCase				:      int			  =				self.fulfilled_idx
         _UpperCAmelCase				:      Any			  =				self.completed

     return new_constraint



class 							A_    :






 '''simple docstring'''




 def __init__(	self							,  _A							,  _A=True) ->		Optional[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				max([len(A__) for one in nested_token_ids])

     _UpperCAmelCase				:      int			  =				{}
     for token_ids in nested_token_ids:
         _UpperCAmelCase				:      Dict			  =				root
         for tidx, token_id in enumerate(A__):
             if token_id not in level:
                 _UpperCAmelCase				:      str			  =				{}

             _UpperCAmelCase				:      Any			  =				level[token_id]

     if no_subsets and self.has_subsets(A__							,  A__):
         raise ValueError(
             '''Each list in `nested_token_ids` can\'t be a complete subset of another list, but is'''
             f''' {nested_token_ids}.''')

     _UpperCAmelCase				:      Optional[Any]			  =				root


 def  snake_case__				(	self							,  _A) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Any			  =				self.trie

     for current_token in current_seq:
         _UpperCAmelCase				:      Tuple			  =				start[current_token]

     _UpperCAmelCase				:      int			  =				list(start.keys())

     return next_tokens


 def  snake_case__				(	self							,  _A) ->		Any:

     """simple docstring"""



     _UpperCAmelCase				:      Tuple			  =				self.next_tokens(A__)

     return len(A__) == 0


 def  snake_case__				(	self							,  _A) ->		Union[str, Any]:

     """simple docstring"""



     _UpperCAmelCase				:      List[str]			  =				list(root.values())
     if len(A__) == 0:
         return 1
     else:
         return sum([self.count_leaves(A__) for nn in next_nodes])


 def  snake_case__				(	self							,  _A							,  _A) ->		Optional[int]:

     """simple docstring"""



     _UpperCAmelCase				:      Union[str, Any]			  =				self.count_leaves(A__)
     return len(A__) != leaf_count



class 							A_    (						__a			):






 '''simple docstring'''




 def __init__(	self							,  _A) ->		str:

     """simple docstring"""



     super(A__							,  self).__init__()

     if not isinstance(A__							,  A__) or len(A__) == 0:
         raise ValueError(f'''`nested_token_ids` has to be a non-empty list, but is {nested_token_ids}.''')
     if any(not isinstance(A__							,  A__) for token_ids in nested_token_ids):
         raise ValueError(f'''`nested_token_ids` has to be a list of lists, but is {nested_token_ids}.''')
     if any(
         any((not isinstance(A__							,  A__) or token_id < 0) for token_id in token_ids)
         for token_ids in nested_token_ids):
         raise ValueError(
             f'''Each list in `nested_token_ids` has to be a list of positive integers, but is {nested_token_ids}.''')

     _UpperCAmelCase				:      List[str]			  =				DisjunctiveTrie(A__)
     _UpperCAmelCase				:      Dict			  =				nested_token_ids

     _UpperCAmelCase				:      Optional[Any]			  =				self.trie.max_height
     _UpperCAmelCase				:      Any			  =				[]
     _UpperCAmelCase				:      Optional[int]			  =				False


 def  snake_case__				(	self) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Tuple			  =				self.trie.next_tokens(self.current_seq)

     if len(A__) == 0:
         return None
     else:
         return token_list


 def  snake_case__				(	self							,  _A) ->		List[str]:

     """simple docstring"""



     if not isinstance(A__							,  A__):
         raise ValueError(f'''`token_id` is supposed to be type `int`, but is {token_id} of type {type(A__)}''')

     _UpperCAmelCase				:      Optional[int]			  =				self.trie.next_tokens(self.current_seq)

     return token_id in next_tokens


 def  snake_case__				(	self							,  _A) ->		Union[str, Any]:

     """simple docstring"""



     if not isinstance(A__							,  A__):
         raise ValueError(f'''`token_id` is supposed to be type `int`, but is {token_id} of type {type(A__)}''')

     _UpperCAmelCase				:      Optional[int]			  =				False
     _UpperCAmelCase				:      Optional[int]			  =				False
     _UpperCAmelCase				:      int			  =				False

     if self.does_advance(A__):
         self.current_seq.append(A__)
         _UpperCAmelCase				:      Optional[Any]			  =				True
     else:
         _UpperCAmelCase				:      Optional[int]			  =				True
         self.reset()

     _UpperCAmelCase				:      int			  =				self.trie.reached_leaf(self.current_seq)
     _UpperCAmelCase				:      List[Any]			  =				completed

     return stepped, completed, reset


 def  snake_case__				(	self) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Tuple			  =				False
     _UpperCAmelCase				:      Tuple			  =				[]


 def  snake_case__				(	self) ->		Tuple:

     """simple docstring"""



     if self.completed:
         # since this can be completed without reaching max height
         return 0
     else:
         return self.seqlen - len(self.current_seq)


 def  snake_case__				(	self							,  _A=False) ->		Optional[int]:

     """simple docstring"""



     _UpperCAmelCase				:      Union[str, Any]			  =				DisjunctiveConstraint(self.token_ids)

     if stateful:
         _UpperCAmelCase				:      int			  =				self.seqlen
         _UpperCAmelCase				:      str			  =				self.current_seq
         _UpperCAmelCase				:      Dict			  =				self.completed

     return new_constraint



class 							A_    :






 '''simple docstring'''




 def __init__(	self							,  _A) ->		Union[str, Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Any			  =				constraints

     # max # of steps required to fulfill a given constraint
     _UpperCAmelCase				:      List[Any]			  =				max([c.seqlen for c in constraints])
     _UpperCAmelCase				:      Optional[Any]			  =				len(A__)
     _UpperCAmelCase				:      List[Any]			  =				False

     self.init_state()


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

     """simple docstring"""



     _UpperCAmelCase				:      Any			  =				[]
     _UpperCAmelCase				:      Union[str, Any]			  =				None
     _UpperCAmelCase				:      List[Any]			  =				[constraint.copy(stateful=A__) for constraint in self.constraints]


 def  snake_case__				(	self) ->		Any:

     """simple docstring"""



     _UpperCAmelCase				:      List[Any]			  =				0
     if self.inprogress_constraint:
         # extra points for having a constraint mid-fulfilled
         add += self.max_seqlen - self.inprogress_constraint.remaining()

     return (len(self.complete_constraints) * self.max_seqlen) + add


 def  snake_case__				(	self) ->		Optional[int]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[Any]			  =				[]
     if self.inprogress_constraint is None:
         for constraint in self.pending_constraints:  # "pending" == "unfulfilled yet"
             _UpperCAmelCase				:      Dict			  =				constraint.advance()
             if isinstance(A__							,  A__):
                 token_list.append(A__)
             elif isinstance(A__							,  A__):
                 token_list.extend(A__)
     else:
         _UpperCAmelCase				:      List[Any]			  =				self.inprogress_constraint.advance()
         if isinstance(A__							,  A__):
             token_list.append(A__)
         elif isinstance(A__							,  A__):
             token_list.extend(A__)

     if len(A__) == 0:
         return None
     else:
         return token_list


 def  snake_case__				(	self							,  _A) ->		Dict:

     """simple docstring"""



     self.init_state()

     if token_ids is not None:
         for token in token_ids:
             # completes or steps **one** constraint
             _UpperCAmelCase  , _UpperCAmelCase				:      Union[str, Any]			  =				self.add(A__)

             # the entire list of constraints are fulfilled
             if self.completed:
                 break


 def  snake_case__				(	self							,  _A) ->		Tuple:

     """simple docstring"""



     if not isinstance(A__							,  A__):
         raise ValueError(f'''`token_id` should be an `int`, but is `{token_id}`.''')

     _UpperCAmelCase  , _UpperCAmelCase				:      Optional[int]			  =				False, False

     if self.completed:
         _UpperCAmelCase				:      Optional[int]			  =				True
         _UpperCAmelCase				:      Optional[Any]			  =				False
         return complete, stepped

     if self.inprogress_constraint is not None:
         # In the middle of fulfilling a constraint. If the `token_id` *does* makes an incremental progress to current
         # job, simply update the state

         _UpperCAmelCase  , _UpperCAmelCase  , _UpperCAmelCase				:      str			  =				self.inprogress_constraint.update(A__)
         if reset:
             # 1. If the next token breaks the progress, then we must restart.
             #     e.g. constraint = "I love pies" and sequence so far is "I love" but `token_id` == "books".

             #     But that doesn't mean we self.init_state(), since we only reset the state for this particular
             #     constraint, not the full list of constraints.

             self.pending_constraints.append(self.inprogress_constraint.copy(stateful=A__))
             _UpperCAmelCase				:      Optional[int]			  =				None

         if complete:
             # 2. If the next token completes the constraint, move it to completed list, set
             #     inprogress to None. If there are no pending constraints either, then this full list of constraints
             #     is complete.

             self.complete_constraints.append(self.inprogress_constraint)
             _UpperCAmelCase				:      List[str]			  =				None

             if len(self.pending_constraints) == 0:
                 # we're done!
                 _UpperCAmelCase				:      List[str]			  =				True

     else:
         # Not in the middle of fulfilling a constraint. So does this `token_id` helps us step towards any of our list
         # of constraints?

         for cidx, pending_constraint in enumerate(self.pending_constraints):
             if pending_constraint.does_advance(A__):
                 _UpperCAmelCase  , _UpperCAmelCase  , _UpperCAmelCase				:      Tuple			  =				pending_constraint.update(A__)

                 if not stepped:
                     raise Exception(
                         '''`constraint.update(token_id)` is not yielding incremental progress, '''
                         '''even though `constraint.does_advance(token_id)` is true.''')

                 if complete:
                     self.complete_constraints.append(A__)
                     _UpperCAmelCase				:      Optional[int]			  =				None

                 if not complete and stepped:
                     _UpperCAmelCase				:      Union[str, Any]			  =				pending_constraint

                 if complete or stepped:
                     # If we made any progress at all, then it's at least not a "pending constraint".

                     _UpperCAmelCase				:      Union[str, Any]			  =				(
                         self.pending_constraints[:cidx] + self.pending_constraints[cidx + 1 :]
                     )

                     if len(self.pending_constraints) == 0 and self.inprogress_constraint is None:
                         # If there's no longer any pending after this and no inprogress either, then we must be
                         # complete.

                         _UpperCAmelCase				:      int			  =				True

                     break  # prevent accidentally stepping through multiple constraints with just one token.

     return complete, stepped


 def  snake_case__				(	self							,  _A=True) ->		Union[str, Any]:

     """simple docstring"""



     _UpperCAmelCase				:      int			  =				ConstraintListState(self.constraints)  # we actually never though self.constraints objects
     # throughout this process. So it's at initialization state.

     if stateful:
         _UpperCAmelCase				:      Union[str, Any]			  =				[
             constraint.copy(stateful=A__) for constraint in self.complete_constraints
         ]
         if self.inprogress_constraint is not None:
             _UpperCAmelCase				:      List[Any]			  =				self.inprogress_constraint.copy(stateful=A__)
         _UpperCAmelCase				:      str			  =				[constraint.copy() for constraint in self.pending_constraints]

     return new_state













import json
from typing import TYPE_CHECKING, List, Optional, Tuple

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_blenderbot import BlenderbotTokenizer


if TYPE_CHECKING:
  from transformers.pipelines.conversational import Conversation

SCREAMING_SNAKE_CASE 							=						logging.get_logger(__name__)


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

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

SCREAMING_SNAKE_CASE 							=						{'facebook/blenderbot-3B': 128}
class 							A_    (						__lowercase			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			Tuple =					VOCAB_FILES_NAMES
 _SCREAMING_SNAKE_CASE							:			Optional[Any] =					PRETRAINED_VOCAB_FILES_MAP
 _SCREAMING_SNAKE_CASE							:			Optional[int] =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _SCREAMING_SNAKE_CASE							:			List[Any] =					["input_ids", "attention_mask"]
 _SCREAMING_SNAKE_CASE							:			Dict =					BlenderbotTokenizer


 def __init__(	self							,  _A=None							,  _A=None							,  _A=None							,  _A="replace"							,  _A="<s>"							,  _A="</s>"							,  _A="</s>"							,  _A="<s>"							,  _A="<unk>"							,  _A="<pad>"							,  _A="<mask>"							,  _A=False							,  _A=True							,  **_A							,  ) ->		str:

     """simple docstring"""



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

     _UpperCAmelCase				:      Any			  =				json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__())
     if pre_tok_state.get('''add_prefix_space'''							,  _A) != add_prefix_space:
         _UpperCAmelCase				:      Optional[int]			  =				getattr(_A							,  pre_tok_state.pop('''type'''))
         _UpperCAmelCase				:      Optional[int]			  =				add_prefix_space
         _UpperCAmelCase				:      Dict			  =				pre_tok_class(**_A)

     _UpperCAmelCase				:      Dict			  =				add_prefix_space

     _UpperCAmelCase				:      Tuple			  =				'''post_processor'''
     _UpperCAmelCase				:      str			  =				getattr(self.backend_tokenizer							,  _A							,  _A)
     if tokenizer_component_instance:
         _UpperCAmelCase				:      str			  =				json.loads(tokenizer_component_instance.__getstate__())

         # The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
         if "sep" in state:
             _UpperCAmelCase				:      str			  =				tuple(state['''sep'''])
         if "cls" in state:
             _UpperCAmelCase				:      List[str]			  =				tuple(state['''cls'''])

         _UpperCAmelCase				:      List[str]			  =				False

         if state.get('''add_prefix_space'''							,  _A) != add_prefix_space:
             _UpperCAmelCase				:      int			  =				add_prefix_space
             _UpperCAmelCase				:      Optional[int]			  =				True

         if state.get('''trim_offsets'''							,  _A) != trim_offsets:
             _UpperCAmelCase				:      List[Any]			  =				trim_offsets
             _UpperCAmelCase				:      Any			  =				True

         if changes_to_apply:
             _UpperCAmelCase				:      Optional[Any]			  =				getattr(_A							,  state.pop('''type'''))
             _UpperCAmelCase				:      Union[str, Any]			  =				component_class(**_A)
             setattr(self.backend_tokenizer							,  _A							,  _A)


 @property
 # Copied from transformers.models.roberta.tokenization_roberta_fast.RobertaTokenizerFast.mask_token with Roberta->Blenderbot, RoBERTa->Blenderbot
 def  snake_case__				(	self) ->		str:

     """simple docstring"""



     if self._mask_token is None:
         if self.verbose:
             logger.error('''Using mask_token, but it is not set yet.''')
         return None
     return str(self._mask_token)


 @mask_token.setter
 def  snake_case__				(	self							,  _A) ->		str:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				AddedToken(_A							,  lstrip=_A							,  rstrip=_A) if isinstance(_A							,  _A) else value
     _UpperCAmelCase				:      Union[str, Any]			  =				value


 def  snake_case__				(	self							,  *_A							,  **_A) ->		BatchEncoding:

     """simple docstring"""



     _UpperCAmelCase				:      List[Any]			  =				kwargs.get('''is_split_into_words'''							,  _A)
     assert self.add_prefix_space or not is_split_into_words, (
         f'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True '''
         "to use it with pretokenized inputs."
     )

     return super()._batch_encode_plus(*_A							,  **_A)


 def  snake_case__				(	self							,  *_A							,  **_A) ->		BatchEncoding:

     """simple docstring"""



     _UpperCAmelCase				:      Tuple			  =				kwargs.get('''is_split_into_words'''							,  _A)

     assert self.add_prefix_space or not is_split_into_words, (
         f'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True '''
         "to use it with pretokenized inputs."
     )

     return super()._encode_plus(*_A							,  **_A)


 def  snake_case__				(	self							,  _A							,  _A = None) ->		Tuple[str]:

     """simple docstring"""



     _UpperCAmelCase				:      str			  =				self._tokenizer.model.save(_A							,  name=_A)
     return tuple(_A)


 def  snake_case__				(	self							,  _A							,  _A = None) ->		List[int]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				[self.sep_token_id]
     _UpperCAmelCase				:      str			  =				[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							,  _A							,  _A = None) ->		Optional[Any]:

     """simple docstring"""



     return token_ids_a + [self.eos_token_id]


 def  snake_case__				(	self							,  _A) ->		List[int]:

     """simple docstring"""



     _UpperCAmelCase				:      Dict			  =				[]
     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				:      int			  =				'''  '''.join(_A)
     _UpperCAmelCase				:      Optional[int]			  =				self.encode(_A)
     if len(_A) > self.model_max_length:
         _UpperCAmelCase				:      str			  =				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




