mistral_common.tokens.tokenizers.instruct

def decode(self, tokens: list[int], special_token_policy: SpecialTokenPolicy = SpecialTokenPolicy.IGNORE) -> str:
    r"""Decode tokens to a string.

    Args:
        tokens: The token IDs to decode.
        special_token_policy: The policy to use for special tokens
            (IGNORE, KEEP, or RAISE).

    Returns:
        The decoded string.
    """
    return self.tokenizer.decode(tokens, special_token_policy=special_token_policy)

@abstractmethod
def encode_assistant_message(self, message: AssistantMessage, is_before_last_user_message: bool) -> list[int]:
    r"""Encode an assistant message.

    Args:
        message: The assistant message to encode.
        is_before_last_user_message: `True` if this message comes before the
            last user message of the conversation.

    Returns:
        The encoded token IDs for this message.

    Raises:
        NotImplementedError: The assistant message is not implemented for the base tokenizer.
    """
    raise NotImplementedError("Assistant message not implemented")

def encode_instruct(
    self,
    request: InstructRequest[ChatMessage, Tool],
) -> Tokenized:
    r"""Encode an instruct request.

    Walks the message list, encoding each message with its version-specific
    encoder, and aggregates the resulting tokens, images, and audio into a
    single Tokenized object. Applies truncation when
    `request.truncate_at_max_tokens` is set.

    Args:
        request: The request to encode.

    Returns:
        The tokenized request, with all images and audio collected across
        the messages.

    Raises:
        TokenizerException: If a message has an unknown type.
    """
    # init at bos
    images: list[np.ndarray] = []
    audios: list[Audio] = []
    prefix_ids: list[int] | None = None
    tokens_list: list[list[int] | None] = []

    # validate messages
    self.validate_messages(request.messages)

    # find last user message
    first_user_idx, last_user_idx = self.find_first_last_user(request)
    for msg_idx, msg in enumerate(request.messages):
        if isinstance(msg, UserMessage):
            new_tokens, new_images, new_audios = self.encode_user_message(
                msg,
                request.available_tools,
                msg_idx == last_user_idx,
                msg_idx == first_user_idx,
                system_prompt=request.system_prompt,
                force_img_first=True,  # img is always first when providing text/img chunk pair
                settings=request.settings,
            )
            images.extend(new_images)
            audios.extend(new_audios)
        elif isinstance(msg, ToolMessage):
            new_tokens, new_images, new_audios = self.encode_tool_message(msg, msg_idx < last_user_idx)
            images.extend(new_images)
            audios.extend(new_audios)
        elif isinstance(msg, AssistantMessage):
            new_tokens = self.encode_assistant_message(
                message=msg, is_before_last_user_message=msg_idx < last_user_idx
            )
            if msg.prefix:
                assert msg_idx == len(request.messages) - 1
                prefix_ids = new_tokens
        elif isinstance(msg, SystemMessage):
            new_tokens, new_audios = self.encode_system_message(msg)
            audios.extend(new_audios)
        else:
            raise TokenizerException(f"Unknown message type {type(msg)}")

        tokens_list.append(new_tokens)

    if request.truncate_at_max_tokens is not None:
        self._truncate_for_max_tokens(
            tokens_list,
            request.messages,
            request.truncate_at_max_tokens,
            last_user_idx,
        )
    tokens = self.start()

    for tok in tokens_list:
        if tok is not None:
            tokens.extend(tok)

    tokenized = Tokenized(
        tokens=tokens,
        prefix_ids=prefix_ids,
        images=images,
        audios=audios,
    )
    # TODO: remove once 1.13.0 lands.
    tokenized._text = self.decode(tokens=tokens, special_token_policy=SpecialTokenPolicy.KEEP)
    return tokenized

@abstractmethod
def encode_think(self, chunk: ThinkChunk) -> list[int]:
    r"""Encode a think chunk.

    Args:
        chunk: The think chunk to encode.

    Returns:
        The encoded token IDs for this chunk.

    Raises:
        NotImplementedError: The think chunk is not implemented for the base tokenizer.
    """
    raise NotImplementedError("Think chunk not implemented")

@abstractmethod
def encode_tool_message(
    self, message: ToolMessage, is_before_last_user_message: bool
) -> tuple[list[int], list[np.ndarray], list[Audio]]:
    r"""Encode a tool message.

    Args:
        message: The tool message to encode.
        is_before_last_user_message: `True` if this message comes before the
            last user message of the conversation.

    Returns:
        A tuple of (tokens, images, audios): the encoded token IDs, images,
        and audio for this message.

    Raises:
        NotImplementedError: The tool message is not implemented for the base tokenizer.
    """
    raise NotImplementedError("Tool message not implemented")

@staticmethod
def find_first_last_user(request: InstructRequest) -> tuple[int, int]:
    r"""Find the first and last user message in the request.

    Args:
        request: The request to search for user messages.

    Returns:
        A tuple of (`first_user_idx`, `last_user_idx`), the message indexes of
        the first and last UserMessage. Both are -1 if the request has no
        user messages.
    """
    last_user_idx = -1
    first_user_idx = -1
    for i, msg in list(enumerate(request.messages)):
        if isinstance(msg, UserMessage):
            if first_user_idx == -1:
                first_user_idx = i
            last_user_idx = i
    return first_user_idx, last_user_idx

def start(self) -> list[int]:
    r"""Return the start tokens of every encoded request.

    Returns:
        The token IDs prepended to every encoded conversation (the BOS token).
    """
    return [self.tokenizer.bos_id]

@classmethod
def validate_messages(cls, messages: list[UATS]) -> None:
    r"""Validate messages before tokenization.

    No-op for this tokenizer version; message validation starts at v7.

    Args:
        messages: The messages that were validated.
    """
    # We start validating messages for v7
    return

def encode_assistant_message(self, message: AssistantMessage, is_before_last_user_message: bool) -> list[int]:
    r"""Encode an assistant message.

    Args:
        message: The message to encode.
        is_before_last_user_message: Not used.
    Returns:
        The encoded tokens.
    """
    assert isinstance(message, AssistantMessage), message
    if message.tool_calls is not None and len(message.tool_calls) > 0:
        raise TokenizerException("Tools not implemented for tokenizer V1")
    elif message.content:
        assert isinstance(message.content, str), "Message content must be a string for tokenizer < V13"
        curr_tokens = self.tokenizer.encode(message.content, bos=False, eos=False)
    else:
        raise TokenizerException(f"{message.content} // {message.tool_calls}")
    if not message.prefix:
        curr_tokens.append(self.tokenizer.eos_id)
    return curr_tokens

def encode_fim(self, request: FIMRequest) -> Tokenized:
    r"""Encode a FIM request.

    Raises:
       TokenizerException: The FIM request is not implemented for this version.
    """
    raise TokenizerException(f"FIM not available for {self.tokenizer.version}")

def encode_speech_request(self, request: SpeechRequest) -> Tokenized:
    r"""Encode a speech synthesis request.

    Raises:
        TokenizerException: Always; speech requests are not supported by this version.
    """
    raise TokenizerException(f"Speech request not available for tokenizer {self.tokenizer.version.value}")

def encode_system_message(self, message: SystemMessage) -> tuple[list[int], list[Audio]]:
    r"""Encode a system message.

    Raises:
        NotImplementedError: Always; system messages are not supported by this version.
    """
    raise NotImplementedError(f"System message encoding not implemented for {self.__class__.__name__}")

def encode_think(self, chunk: ThinkChunk) -> list[int]:
    r"""Encode a think chunk.

    Raises:
        TokenizerException: The think chunk is not implemented for this version.
    """
    raise TokenizerException("Think not implemented for tokenizer < V13.")

def encode_tool_message(
    self, message: ToolMessage, is_before_last_user_message: bool
) -> tuple[list[int], list[np.ndarray], list[Audio]]:
    r"""Encode a tool message.

    Raises:
        TokenizerException: The tool message is not implemented for this version.
    """
    raise TokenizerException("Tools not implemented for tokenizer V1")

def encode_transcription(self, request: TranscriptionRequest) -> Tokenized:
    r"""Encode a transcription request.

    Raises:
        TokenizerException: Always; transcription is not supported by this version.
    """
    raise TokenizerException(f"Transcription not available for {self.tokenizer.version}")

def encode_user_content(
    self,
    content: str | list[ContentChunk],
    is_last: bool,
    system_prompt: str | None = None,
    force_img_first: bool = False,
) -> tuple[list[int], list[np.ndarray], list[Audio]]:
    r"""Encode a user content.

    Args:
        content: The content to encode.
        is_last: Whether the message is the last one.
        system_prompt: The system prompt.
        force_img_first: Not used.

    Returns:
        The encoded tokens and empty list.
    """
    assert isinstance(content, str)

    if is_last and system_prompt:
        content = system_prompt + "\n\n" + content

    tokens = self.tokenizer.encode(content, bos=False, eos=False)
    return tokens, [], []

def encode_user_message(
    self,
    message: UserMessage,
    available_tools: list[Tool] | None,
    is_last: bool,
    is_first: bool,
    system_prompt: str | None = None,
    force_img_first: bool = False,
    settings: ModelSettings = ModelSettings.none(),
) -> tuple[list[int], list[np.ndarray], list[Audio]]:
    r"""Encode a user message.

    Args:
        message: The message to encode.
        available_tools: Not used.
        is_last: Not used.
        is_first: Whether the message is the first one.
        system_prompt: The system prompt.
        force_img_first: Not used.
        settings: Not used.

    Returns:
        The encoded tokens and empty list.
    """
    assert isinstance(message.content, str), "Message content must be normalized"
    assert self.image_encoder is None, "InstructTokenizerV1 cannot encode images"

    content = ""
    if is_first and system_prompt:
        content = system_prompt + "\n\n" + message.content
    else:
        content = message.content

    message_txt = f"[INST] {content} [/INST]"
    curr_tokens, image, audio = self.encode_user_content(content=message_txt, is_last=False, system_prompt=None)
    return curr_tokens, image, audio

def encode_think(self, chunk: ThinkChunk) -> list[int]:
    r"""Encode a thinking chunk.

    Args:
        chunk: The thinking chunk to encode.
    Returns:
        The encoded tokens.
    """
    assert self.BEGIN_THINK is not None, "think tokens are not available for this tokenizer."
    assert self.END_THINK is not None, "think tokens are not available for this tokenizer."
    tokens = self.tokenizer.encode(chunk.thinking, bos=False, eos=False)
    think_tokens = [self.BEGIN_THINK, *tokens]
    if chunk.closed:
        think_tokens.append(self.END_THINK)
    return think_tokens

def encode_tool_message(
    self, message: ToolMessage, is_before_last_user_message: bool
) -> tuple[list[int], list[np.ndarray], list[Audio]]:
    r"""Encode a tool message.

    Args:
        message: The message to encode.
        is_before_last_user_message: Not used.
    Returns:
        The encoded tokens, images, and audios.
    """
    assert message.tool_call_id is not None, "Tool call id must be provided for tokenizer >= v13"

    if isinstance(message.content, str):
        content_tokens = self.tokenizer.encode(message.content, bos=False, eos=False)
        images: list[np.ndarray] = []
        audios: list[Audio] = []
    else:
        content_tokens, images, audios = self._encode_content_chunks(message.content)

    curr_tokens = [
        self.BEGIN_TOOL_RESULTS,
        *content_tokens,
        self.END_TOOL_RESULTS,
    ]
    return curr_tokens, images, audios

@classmethod
def validate_messages(cls, messages: list[UATS]) -> None:
    r"""Validate messages before tokenization.

    No-op: unlike v7, this version allows system prompts and audio chunks
    to coexist.

    Args:
        messages: The messages that were validated.
    """
    return

def encode_system_message(self, message: SystemMessage) -> tuple[list[int], list[Audio]]:
    r"""Encode a system message, rejecting ThinkChunk content."""
    if isinstance(message.content, list):
        if any(isinstance(chunk, ThinkChunk) for chunk in message.content):
            raise TokenizerException("ThinkChunk in system message is not supported for this model")
    return super().encode_system_message(message)

def encode_assistant_message(self, message: AssistantMessage, is_before_last_user_message: bool) -> list[int]:
    r"""Encode an assistant message.

    Args:
        message: The message to encode.
        is_before_last_user_message: Whether the message is before the last user message. If has tools and true, the
            message is not encoded.
    Returns:
        The encoded tokens.
    """
    if message.tool_calls and message.content:
        raise ValueError(f"Cannot have tool calls and content defined in the same assistant message {message}")
    if message.tool_calls:
        if is_before_last_user_message:
            # don't tokenize tool call before last user message
            return []
        curr_tokens = self._encode_tool_calls_in_assistant_message(message)
    elif message.content:
        assert isinstance(message.content, str), "Message content must be a string for tokenizer < V7"
        curr_tokens = self._encode_normal_content_assistant_message(message)
    else:
        raise TokenizerException(f"Invalid assistant message: {message.content}")
    if not message.prefix:
        curr_tokens.append(self.tokenizer.eos_id)
    return curr_tokens

def encode_fim(self, request: FIMRequest) -> Tokenized:
    r"""Encode a FIM request.

    Args:
        request: The request to encode.

    Returns:
        The encoded tokens.
    """
    prefix_tokens = self.tokenizer.encode(request.prompt, bos=False, eos=False)
    suffix_tokens = self._encode_infilling(request.suffix) if request.suffix else []
    tokens = [
        self.BOS,
        self.SUFFIX,
        *suffix_tokens,
        self.PREFIX,
        *prefix_tokens,
    ]
    tokenized = Tokenized(tokens=tokens)
    # TODO: remove once 1.13.0 lands.
    tokenized._text = self.decode(tokens=tokens, special_token_policy=SpecialTokenPolicy.KEEP)
    return tokenized

def encode_instruct(self, request: InstructRequest[ChatMessage, Tool]) -> Tokenized:
    r"""Encode a request after validating tool-result limitations.

    Args:
        request: The instruct request to encode.

    Returns:
        The tokenized request.

    Raises:
        UnsupportedTokenizerFeatureException: If an assistant turn has
            multiple tool results.
    """
    self._validate_tool_results(request=request)
    return super().encode_instruct(request)

def encode_tool_message(
    self, message: ToolMessage, is_before_last_user_message: bool
) -> tuple[list[int], list[np.ndarray], list[Audio]]:
    r"""Encode a tool message.

    Args:
        message: The message to encode.
        is_before_last_user_message: Whether the message is before the last user message. If true, the message is
            not encoded.

    Returns:
        The encoded tokens, images, and audios.
    """
    if is_before_last_user_message:
        # don't tokenize last tool response before last user msg
        return [], [], []

    # Currently only supports single tool results
    tool_result_str = json.dumps([self._prepare_tool_result(message)], ensure_ascii=False)
    curr_tokens = [
        self.BEGIN_TOOL_RESULTS,
        *self.tokenizer.encode(tool_result_str, bos=False, eos=False),
        self.END_TOOL_RESULTS,
    ]
    return curr_tokens, [], []

def encode_user_message(
    self,
    message: UserMessage,
    available_tools: list[Tool] | None,
    is_last: bool,
    is_first: bool,
    system_prompt: str | None = None,
    force_img_first: bool = False,
    settings: ModelSettings = ModelSettings.none(),
) -> tuple[list[int], list[np.ndarray], list[Audio]]:
    r"""Encode a user message.

    Args:
        message: The message to encode.
        available_tools: The list of available tools if any.
        is_last: Whether the message is the last one.
        is_first: Not used.
        system_prompt: The system prompt.
        force_img_first: Whether to force the image to be first.
        settings: Not used.

    Returns:
        The encoded tokens and the list of images.
    """
    do_encode_tools_settings = False
    do_encode_tools_settings |= is_first and (
        self._message_position_to_encode_tools_settings == UserMessagePosition.first
    )
    do_encode_tools_settings |= is_last and (
        self._message_position_to_encode_tools_settings == UserMessagePosition.last
    )
    tools_settings_tokens: list[int] = []

    if do_encode_tools_settings and available_tools:
        tools = [tool.model_dump(exclude={"function": {"strict": True}}) for tool in available_tools]
        tools_json_tokens = self.tokenizer.encode(json.dumps(tools, ensure_ascii=False), bos=False, eos=False)
        tools_settings_tokens.extend(
            [
                self.BEGIN_AVAILABLE_TOOLS,
                *tools_json_tokens,
                self.END_AVAILABLE_TOOLS,
            ]
        )

    if do_encode_tools_settings:
        tools_settings_tokens.extend(self._encode_settings(settings=settings))

    tokens, image, audio = self.encode_user_content(
        content=message.content,
        is_last=is_last,
        system_prompt=system_prompt,
        force_img_first=force_img_first,
    )

    prefix_tokens = [*tools_settings_tokens, self.BEGIN_INST]
    suffix_tokens = [self.END_INST]

    curr_tokens = prefix_tokens + tokens + suffix_tokens

    return curr_tokens, image, audio

def encode_assistant_message(self, message: AssistantMessage, is_before_last_user_message: bool) -> list[int]:
    r"""Encode an assistant message.

    Note:
        Same as [V2][mistral_common.tokens.tokenizers.instruct.InstructTokenizerV2.encode_assistant_message] but
        always encode the tool history.

    Args:
        message: The message to encode.
        is_before_last_user_message: Not used.

    Returns:
        The encoded tokens.
    """
    return super().encode_assistant_message(message=message, is_before_last_user_message=False)

def encode_tool_message(
    self, message: ToolMessage, is_before_last_user_message: bool
) -> tuple[list[int], list[np.ndarray], list[Audio]]:
    r"""Encode a tool message.

    Note:
        Same as [V2][mistral_common.tokens.tokenizers.instruct.InstructTokenizerV2.encode_tool_message] but tools
        are not wrapped in a list and the history is also tokenized.

    Args:
        message: The message to encode.
        is_before_last_user_message: Whether the message is before the last user message. If true, the message is
            not encoded.

    Returns:
        The encoded tokens, images, and audios.
    """
    tool_result_str = json.dumps(self._prepare_tool_result(message), ensure_ascii=False)
    curr_tokens = [
        self.BEGIN_TOOL_RESULTS,
        *self.tokenizer.encode(tool_result_str, bos=False, eos=False),
        self.END_TOOL_RESULTS,
    ]
    return curr_tokens, [], []