





'''simple docstring'''




import os
from shutil import copyfile
from typing import TYPE_CHECKING, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging
from ...utils.versions import require_version


if TYPE_CHECKING:
  from transformers.pipelines.conversational import Conversation

require_version('tokenizers>=0.13.3')

if is_sentencepiece_available():
  from .tokenization_llama import LlamaTokenizer
else:
  _a			:	Any								=  None

_a			:	str								=  logging.get_logger(__name__)
_a			:	List[str]								=  {'vocab_file': 'tokenizer.model', 'tokenizer_file': 'tokenizer.json'}

_a			:	str								=  '[INST]', '[/INST]'
_a			:	Any								=  '<<SYS>>\n', '\n<</SYS>>\n\n'

# fmt: off
_a			:	Any								=  'You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your\\nanswers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure\\nthat your responses are socially unbiased and positive in nature.\n\nIf a question does not make any sense, or is not factually coherent, explain why instead of answering something not\\ncorrect. If you don\'t know the answer to a question, please don\'t share false information.'



class     lowercase_		(						a	):

  '''simple docstring'''





  __lowerCAmelCase   :	Any      					=				VOCAB_FILES_NAMES
  __lowerCAmelCase   :	int      					=				LlamaTokenizer
  __lowerCAmelCase   :	Optional[int]      					=				"left"
  __lowerCAmelCase   :	Any      					=				["input_ids", "attention_mask"]
  def __init__(       self				,	a_=None				,	a_=None				,	a_=False				,	a_="<unk>"				,	a_="<s>"				,	a_="</s>"				,	a_=True				,	a_=False				,	**a_				,	)							->			Any:


    """simple docstring"""
    super().__init__(
        vocab_file=a_				,	tokenizer_file=a_				,	clean_up_tokenization_spaces=a_				,	unk_token=a_				,	bos_token=a_				,	eos_token=a_				,	**a_				,	)
    UpperCAmelCase  							=		add_bos_token
    UpperCAmelCase  							=		add_eos_token
    self.update_post_processor()

    UpperCAmelCase  							=		vocab_file
    UpperCAmelCase  							=		False if not self.vocab_file else True
  def   snake_case_     (       self       )							->			Union[str, Any]:


    """simple docstring"""
    UpperCAmelCase  							=		self.bos_token
    UpperCAmelCase  							=		self.bos_token_id

    UpperCAmelCase  							=		self.eos_token
    UpperCAmelCase  							=		self.eos_token_id

    UpperCAmelCase  							=		F'''{(bos+':0 ') * self.add_bos_token}$A:0{(' '+eos+':0') * self.add_eos_token}'''
    UpperCAmelCase  							=		F'''{single}{(' '+bos+':1') * self.add_bos_token} $B:1{(' '+eos+':1') * self.add_eos_token}'''

    UpperCAmelCase  							=		[]
    if self.add_bos_token:
      special_tokens.append((bos, bos_token_id)       )
    if self.add_eos_token:
      special_tokens.append((eos, eos_token_id)       )
    UpperCAmelCase  							=		processors.TemplateProcessing(
        single=a_				,	pair=a_				,	special_tokens=a_       )
  @property
  def   snake_case_     (       self       )							->			Any:


    """simple docstring"""
    return self._add_eos_token
  @property
  def   snake_case_     (       self       )							->			Optional[int]:


    """simple docstring"""
    return self._add_bos_token
  @add_eos_token.setter
  def   snake_case_     (       self				,	a_       )							->			int:


    """simple docstring"""
    UpperCAmelCase  							=		value
    self.update_post_processor()
  @add_bos_token.setter
  def   snake_case_     (       self				,	a_       )							->			Union[str, Any]:


    """simple docstring"""
    UpperCAmelCase  							=		value
    self.update_post_processor()
  def   snake_case_     (       self				,	a_				,	a_ = None       )							->			Tuple[str]:


    """simple docstring"""
    if not self.can_save_slow_tokenizer:
      raise ValueError(
          'Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '
          'tokenizer.'       )

    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']       )

    if os.path.abspath(self.vocab_file       ) != os.path.abspath(a_       ):
      copyfile(self.vocab_file				,	a_       )

    return (out_vocab_file,)





  def   snake_case_     (       self				,	a_       )							->			Tuple:


    """simple docstring"""
    UpperCAmelCase  							=		list(conversation.iter_texts()       )
    if not all([is_user for is_user, msg in dialogue[::2]]       ) or not all(
        [not is_user for is_user, msg in dialogue[1::2]]       ):
      raise ValueError(
          'The model only supports \'user\' and \'assistant\' roles, starting with user and alternating (u/a/u/a/u...)'       )

    UpperCAmelCase  							=		[]
    if len(conversation.past_user_inputs       ) > 0:
      if not conversation.past_user_inputs[0].startswith(a_       ) or E_SYS not in conversation.past_user_inputs[0]:
        UpperCAmelCase  							=		(
            B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
        )
    elif not dialogue[0][1].startswith(a_       ) or E_SYS not in dialogue[0][1]:
      UpperCAmelCase  							=		(dialogue[0][0], B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + dialogue[0][1])

    dialog_tokens += sum(
        [
            [self.bos_token_id]
            + self.encode(
                F'''{B_INST} {(prompt[1]).strip()} {E_INST} {(answer[1]).strip()} '''				,	add_special_tokens=a_       )
            + [self.eos_token_id]
            for prompt, answer in zip(dialogue[::2]				,	dialogue[1::2]       )
        ]				,	[]				,	)
    if not (dialogue[-1][0]):
      raise ValueError(F'''Last message must be from user, got {dialogue[-1]['role']}'''       )
    dialog_tokens += [self.bos_token_id] + self.encode(
        F'''{B_INST} {(dialogue[-1][1]).strip()} {E_INST}'''				,	add_special_tokens=a_       )
    return dialog_tokens





'''simple docstring'''




import os
import pathlib
import tempfile
import uuid

import numpy as np

from ..utils import is_soundfile_availble, is_torch_available, is_vision_available, logging


_a			:	int								=  logging.get_logger(__name__)

if is_vision_available():
  import PIL.Image
  from PIL import Image
  from PIL.Image import Image as ImageType
else:
  _a			:	List[str]								=  object

if is_torch_available():
  import torch

if is_soundfile_availble():
  import soundfile as sf



class     lowercase_		:

  '''simple docstring'''





  def __init__(       self				,	a_       )							->			Dict:


    """simple docstring"""
    UpperCAmelCase  							=		value
  def __str__(       self       )							->			Any:


    """simple docstring"""
    return self.to_string()
  def   snake_case_     (       self       )							->			Union[str, Any]:


    """simple docstring"""
    logger.error(
        'This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable'       )
    return self._value





  def   snake_case_     (       self       )							->			str:


    """simple docstring"""
    logger.error(
        'This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable'       )
    return str(self._value       )




class     lowercase_		(						a	,    a	):

  '''simple docstring'''





  def   snake_case_     (       self       )							->			Dict:


    """simple docstring"""
    return self._value





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


    """simple docstring"""
    return self._value




class     lowercase_		(						a	,    a	):

  '''simple docstring'''





  def __init__(       self				,	a_       )							->			int:


    """simple docstring"""
    super().__init__(a_       )

    if not is_vision_available():
      raise ImportError('PIL must be installed in order to handle images.'       )

    UpperCAmelCase  							=		None
    UpperCAmelCase  							=		None
    UpperCAmelCase  							=		None

    if isinstance(a_				,	a_       ):
      UpperCAmelCase  							=		value
    elif isinstance(a_				,	(str, pathlib.Path)       ):
      UpperCAmelCase  							=		value
    elif isinstance(a_				,	torch.Tensor       ):
      UpperCAmelCase  							=		value
    else:
      raise ValueError(F'''Unsupported type for {self.__class__.__name__}: {type(a_       )}'''       )
  def   snake_case_     (       self				,	a_=None				,	a_=None       )							->			Dict:


    """simple docstring"""
    from IPython.display import Image, display

    display(Image(self.to_string()       )       )
  def   snake_case_     (       self       )							->			List[str]:


    """simple docstring"""
    if self._raw is not None:
      return self._raw

    if self._path is not None:
      UpperCAmelCase  							=		Image.open(self._path       )
      return self._raw





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


    """simple docstring"""
    if self._path is not None:
      return self._path

    if self._raw is not None:
      UpperCAmelCase  							=		tempfile.mkdtemp()
      UpperCAmelCase  							=		os.path.join(a_				,	str(uuid.uuida()       ) + '.png'       )
      self._raw.save(self._path       )

      return self._path

    if self._tensor is not None:
      UpperCAmelCase  							=		self._tensor.cpu().detach().numpy()

      # There is likely simpler than load into image into save
      UpperCAmelCase  							=		Image.fromarray((array * 2_5_5).astype(np.uinta       )       )

      UpperCAmelCase  							=		tempfile.mkdtemp()
      UpperCAmelCase  							=		os.path.join(a_				,	str(uuid.uuida()       ) + '.png'       )

      img.save(self._path       )

      return self._path




class     lowercase_		(						a	):

  '''simple docstring'''





  def __init__(       self				,	a_				,	a_=1_6_0_0_0       )							->			str:


    """simple docstring"""
    super().__init__(a_       )

    if not is_soundfile_availble():
      raise ImportError('soundfile must be installed in order to handle audio.'       )

    UpperCAmelCase  							=		None
    UpperCAmelCase  							=		None

    UpperCAmelCase  							=		samplerate

    if isinstance(a_				,	(str, pathlib.Path)       ):
      UpperCAmelCase  							=		value
    elif isinstance(a_				,	torch.Tensor       ):
      UpperCAmelCase  							=		value
    else:
      raise ValueError(F'''Unsupported audio type: {type(a_       )}'''       )
  def   snake_case_     (       self				,	a_=None				,	a_=None       )							->			Dict:


    """simple docstring"""
    from IPython.display import Audio, display

    display(Audio(self.to_string()				,	rate=self.samplerate       )       )
  def   snake_case_     (       self       )							->			List[str]:


    """simple docstring"""
    if self._tensor is not None:
      return self._tensor

    if self._path is not None:
      UpperCAmelCase							, UpperCAmelCase  							=		sf.read(self._path       )
      UpperCAmelCase  							=		torch.tensor(a_       )
      return self._tensor





  def   snake_case_     (       self       )							->			int:


    """simple docstring"""
    if self._path is not None:
      return self._path

    if self._tensor is not None:
      UpperCAmelCase  							=		tempfile.mkdtemp()
      UpperCAmelCase  							=		os.path.join(a_				,	str(uuid.uuida()       ) + '.wav'       )
      sf.write(self._path				,	self._tensor				,	samplerate=self.samplerate       )
      return self._path




_a			:	Optional[int]								=  {'text': AgentText, 'image': AgentImage, 'audio': AgentAudio}
_a			:	Tuple								=  {str: AgentText}

if is_vision_available():
  _a			:	List[str]								=  AgentImage


def  lowerCamelCase__						(  *SCREAMING_SNAKE_CASE :				List[str]    ,				**SCREAMING_SNAKE_CASE :				Tuple  ):
  UpperCAmelCase  							=		[(arg.to_raw() if isinstance(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE  ) else arg) for arg in args]
  UpperCAmelCase  							=		{k: (v.to_raw() if isinstance(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE  ) else v) for k, v in kwargs.items()}
  return args, kwargs



def  lowerCamelCase__						(  SCREAMING_SNAKE_CASE :				Tuple    ,				SCREAMING_SNAKE_CASE :				Optional[Any]=None  ):
  if isinstance(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE  ):
    UpperCAmelCase  							=		{}
    for i, (k, v) in enumerate(outputs.items()  ):
      if output_types is not None:
        # If the class has defined outputs, we can map directly according to the class definition
        if output_types[i] in AGENT_TYPE_MAPPING:
          UpperCAmelCase  							=		AGENT_TYPE_MAPPING[output_types[i]](SCREAMING_SNAKE_CASE  )
        else:
          UpperCAmelCase  							=		AgentType(SCREAMING_SNAKE_CASE  )

      else:
        # If the class does not have defined output, then we map according to the type
        for _k, _v in INSTANCE_TYPE_MAPPING.items():
          if isinstance(SCREAMING_SNAKE_CASE    ,				_k  ):
            UpperCAmelCase  							=		_v(SCREAMING_SNAKE_CASE  )
        if k not in decoded_outputs:
          UpperCAmelCase  							=		AgentType[v]

  elif isinstance(SCREAMING_SNAKE_CASE    ,				(list, tuple)  ):
    UpperCAmelCase  							=		type(SCREAMING_SNAKE_CASE  )()
    for i, v in enumerate(SCREAMING_SNAKE_CASE  ):
      if output_types is not None:
        # If the class has defined outputs, we can map directly according to the class definition
        if output_types[i] in AGENT_TYPE_MAPPING:
          decoded_outputs.append(AGENT_TYPE_MAPPING[output_types[i]](SCREAMING_SNAKE_CASE  )  )
        else:
          decoded_outputs.append(AgentType(SCREAMING_SNAKE_CASE  )  )
      else:
        # If the class does not have defined output, then we map according to the type
        UpperCAmelCase  							=		False
        for _k, _v in INSTANCE_TYPE_MAPPING.items():
          if isinstance(SCREAMING_SNAKE_CASE    ,				_k  ):
            decoded_outputs.append(_v(SCREAMING_SNAKE_CASE  )  )
            UpperCAmelCase  							=		True

        if not found:
          decoded_outputs.append(AgentType(SCREAMING_SNAKE_CASE  )  )

  else:
    if output_types[0] in AGENT_TYPE_MAPPING:
      # If the class has defined outputs, we can map directly according to the class definition
      UpperCAmelCase  							=		AGENT_TYPE_MAPPING[output_types[0]](SCREAMING_SNAKE_CASE  )

    else:
      # If the class does not have defined output, then we map according to the type
      for _k, _v in INSTANCE_TYPE_MAPPING.items():
        if isinstance(SCREAMING_SNAKE_CASE    ,				_k  ):
          return _v(SCREAMING_SNAKE_CASE  )
      return AgentType(SCREAMING_SNAKE_CASE  )

  return decoded_outputs


