/
githubmirror
/
transformers
Обзор
Документация
Войти
/
githubmirror
/
transformers
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
main
tests/test_tokenization_mistral_common.py
2 446 строк
233 KB
Julien Denize
[MistralCommonBackend] Stop relying on mistral-common's Tokenized.text field (#47646)
10 авг 2026, 10:41
Не верифицирован
10 авг 2026, 10:41
81df408
Код
Авторство
О чём код?
# Copyright 2025 Mistral AI and The HuggingFace Inc. team. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import base64 import gc import importlib.util import io import json import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch import numpy as np import torch from tests.integrations.mistral.tekken_fixtures import write_fake_tekken_json from transformers.image_utils import load_image from transformers.integrations.mistral import convert_tekken_tokenizer from transformers.models.auto.tokenization_auto import AutoTokenizer from transformers.testing_utils import ( require_mistral_common, ) from transformers.tokenization_mistral_common import MistralCommonBackend from transformers.tokenization_utils_base import BatchEncoding, TruncationStrategy from transformers.utils import PaddingStrategy, is_mistral_common_available if is_mistral_common_available(): import mistral_common.tokens.tokenizers from mistral_common.exceptions import InvalidMessageStructureException from mistral_common.protocol.instruct.request import ChatCompletionRequest from mistral_common.protocol.instruct.validator import ( ValidationMode, ) from mistral_common.protocol.transcription.request import TranscriptionRequest from mistral_common.tokens.tokenizers.base import SpecialTokenPolicy from mistral_common.tokens.tokenizers.mistral import MistralTokenizer from mistral_common.tokens.tokenizers.utils import list_local_hf_repo_files # To avoid unnecessary `httpx.get` calls which give us `Error: Too Many Requests for url` on CircleCI mistral_common.tokens.tokenizers.image.download_image = load_image from .test_processing_common import url_to_local_path IMG_URL = url_to_local_path( "https://huggingface.co/datasets/raushan-testing-hf/images_test/resolve/main/picsum_237_200x300.jpg" ) # Required by `mistral_common.tokens.tokenizers.image.image_from_chunk` to correctly use local file IMG_URL = f"file://{IMG_URL}" if not IMG_URL.startswith("http") else IMG_URL IMG_BASE_64 = """/9j/4QDeRXhpZgAASUkqAAgAAAAGABIBAwABAAAAAQAAABoBBQABAAAAVgAAABsBBQABAAAAXgAAACgBAwABAAAAAgAAABMCAwABAAAAAQAAAGmHBAABAAAAZgAAAAAAAABIAAAAAQAAAEgAAAABAAAABwAAkAcABAAAADAyMTABkQcABAAAAAE … [Строка слишком длинная. Вы можете скачать файл] AUDIO_NAMESPACE = "hf-internal-testing" AUDIO_REPO_NAME = "dummy-audio-samples" AUDIO_FILENAME = "bcn_weather.mp3" AUDIO_URL = url_to_local_path( f"https://huggingface.co/datasets/{AUDIO_NAMESPACE}/{AUDIO_REPO_NAME}/resolve/main/{AUDIO_FILENAME}" ) AUDIO_BASE_64 = """//uUxAAAAAAAAAAAAAAAAAAAAAAAWGluZwAAAA8AAAHNAAFFIAACAwQGBwkLDBAUGBodIiQnKy8yNjo9QEVIS1BTVlteYGNlZ2ltcHR5fH6AgoSGioyPkpOVl5mbnJ6goqWnqaqsra6wsbK0tbe6wMTIzM/S1tnd3+Ll6Ovt7/Dy8/X3+fr8/ … [Строка слишком длинная. Вы можете скачать файл] @require_mistral_common class TestMistralCommonBackend(unittest.TestCase): @classmethod def setUpClass(cls): super().setUpClass() cls.repo_id = "hf-internal-testing/namespace-mistralai-repo_name-Mistral-Small-3.1-24B-Instruct-2503" # determine if we already have this downloaded cls.local_files_only = len(list_local_hf_repo_files(cls.repo_id, revision=None)) > 0 cls.tokenizer = MistralCommonBackend.from_pretrained( cls.repo_id, local_files_only=cls.local_files_only, # This is a hack as `list_local_hf_repo_files` from `mistral_common` has a bug # TODO: Discuss with `mistral-common` maintainers: after a fix being done there, remove this `revision` hack revision=None, ) cls.ref_tokenizer: MistralTokenizer = MistralTokenizer.from_hf_hub( cls.repo_id, local_files_only=cls.local_files_only ) # Define SPM tokenizer to test the private methods that handle SPM and Tekken differencies. cls.spm_repo_id = "mistralai/Mistral-7B-v0.3" # cls.tokenizer_audio: MistralCommonBackend = AutoTokenizer.from_pretrained( # "hf-internal-testing/namesspace-mistralai-repo_name-Voxtral-Mini-3B-2507" # ) repo_id = "mistralai/Voxtral-Mini-3B-2507" local_files_only = len(list_local_hf_repo_files(repo_id, revision=None)) > 0 cls.tokenizer_audio: MistralCommonBackend = AutoTokenizer.from_pretrained( repo_id, local_files_only=local_files_only, revision=None, ) cls.ref_tokenizer_audio = MistralTokenizer.from_hf_hub(repo_id, local_files_only=local_files_only) cls.fixture_conversations = [ [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, ], [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can I help you?"}, {"role": "user", "content": "What is the temperature in Paris?"}, ], ] cls.tokenized_fixture_conversations = [ cls.ref_tokenizer.encode_chat_completion(ChatCompletionRequest.from_openai(conversation)) for conversation in cls.fixture_conversations ] cls.ref_special_ids = {t["rank"] for t in cls.ref_tokenizer.instruct_tokenizer.tokenizer._all_special_tokens} @classmethod def tearDownClass(cls): del cls.tokenizer del cls.ref_tokenizer del cls.tokenizer_audio del cls.ref_tokenizer_audio del cls.fixture_conversations del cls.tokenized_fixture_conversations del cls.ref_special_ids gc.collect() # Copy paste of `MistralCommonBackend._tekken_piece_to_id` def _ref_piece_to_id(self, piece: str) -> int: tekken_tokenizer = self.ref_tokenizer.instruct_tokenizer.tokenizer piece_bytes = piece.encode("utf-8") shift = tekken_tokenizer.num_special_tokens try: return shift + tekken_tokenizer._tekken_token2id_nospecial[piece_bytes] except KeyError: piece_str = piece_bytes.decode("utf-8") if piece_str in tekken_tokenizer._special_tokens_reverse_vocab: return tekken_tokenizer._special_tokens_reverse_vocab[piece_str] return tekken_tokenizer.unk_id def _get_spm_tokenizer(self, mode: str = "test") -> MistralCommonBackend: local_files_only = len(list_local_hf_repo_files(self.spm_repo_id, revision=None)) > 0 return MistralCommonBackend.from_pretrained( self.spm_repo_id, local_files_only=local_files_only, revision=None, mode=mode ) def test_spm_vs_tekken_piece_to_id(self): spm_tokenizer = self._get_spm_tokenizer() self.assertEqual(spm_tokenizer._piece_to_id("<s>", False), 1) self.assertEqual(spm_tokenizer._piece_to_id("h", False), 29484) self.assertEqual(self.tokenizer._piece_to_id("<s>", False), 1) self.assertEqual(self._ref_piece_to_id("<s>"), 1) self.assertEqual(self.tokenizer._piece_to_id("\u0000", False), 1000) self.assertEqual(self._ref_piece_to_id("\u0000"), 1000) self.assertEqual(self.tokenizer._piece_to_id(" String", False), 3000) self.assertEqual(self._ref_piece_to_id(" String"), 3000) self.assertEqual(self.tokenizer._piece_to_id("后汉书", False), 131071) self.assertEqual(self._ref_piece_to_id("后汉书"), 131071) def test_vocab_size(self): self.assertEqual(self.tokenizer.vocab_size, self.ref_tokenizer.instruct_tokenizer.tokenizer.n_words) def test_save_pretrained(self): with tempfile.TemporaryDirectory() as tmp_dir: self.tokenizer.save_pretrained(tmp_dir) loaded_tokenizer = MistralCommonBackend.from_pretrained(tmp_dir) self.assertIsNotNone(loaded_tokenizer) self.assertEqual(self.tokenizer.get_vocab(), loaded_tokenizer.get_vocab()) self.assertEqual( self.tokenizer.tokenizer.instruct_tokenizer.tokenizer.version, loaded_tokenizer.tokenizer.instruct_tokenizer.tokenizer.version, ) with self.assertRaises( ValueError, msg="Kwargs [unk_args] are not supported by `MistralCommonBackend.save_pretrained`." ): with tempfile.TemporaryDirectory() as tmp_dir: self.tokenizer.save_pretrained(tmp_dir, unk_args="") def test_save_pretrained_hf_format_produces_loadable_hf_tokenizer(self): """Saving in HF format writes tokenizer.json + tokenizer_config.json, and the reloaded tokenizer encodes identically to a direct convert_tekken_tokenizer call.""" with tempfile.TemporaryDirectory() as tmp_dir: tmp_path = Path(tmp_dir) src_dir = tmp_path / "src" src_dir.mkdir() tekken_path = write_fake_tekken_json(src_dir) out_dir = str(tmp_path / "out") backend = MistralCommonBackend(tokenizer_path=str(tekken_path)) backend.save_pretrained(out_dir, save_format="hf") out_path = Path(out_dir) self.assertTrue((out_path / "tokenizer.json").exists()) self.assertTrue((out_path / "tokenizer_config.json").exists()) text = "hello world" reloaded = AutoTokenizer.from_pretrained(out_dir, mistral_format=False) expected = convert_tekken_tokenizer(str(tekken_path)).encode(text, add_special_tokens=False) actual = reloaded.encode(text, add_special_tokens=False) self.assertEqual(actual, expected) def test_save_pretrained_hf_format_missing_source_raises(self): """save_pretrained(save_format='hf') raises OSError when the source tekken.json is gone.""" with tempfile.TemporaryDirectory() as src_dir: src_path = Path(src_dir) tekken_path = write_fake_tekken_json(src_path) backend = MistralCommonBackend(tokenizer_path=str(tekken_path)) # Delete the source file so the path is no longer valid. tekken_path.unlink() with tempfile.TemporaryDirectory() as out_dir: with self.assertRaises(OSError): backend.save_pretrained(out_dir, save_format="hf") def test_save_pretrained_hf_format_missing_source_leaves_no_directory(self): """A failed hf-format save (missing source tekken.json) must not create the output directory.""" with tempfile.TemporaryDirectory() as tmp_dir: tmp_path = Path(tmp_dir) tekken_path = write_fake_tekken_json(tmp_path) backend = MistralCommonBackend(tokenizer_path=str(tekken_path)) # Delete the source file so the path is no longer valid. tekken_path.unlink() out_dir = tmp_path / "does-not-exist-yet" with self.assertRaises(OSError): backend.save_pretrained(str(out_dir), save_format="hf") self.assertFalse(out_dir.exists()) def test_save_pretrained_mistral_format_copy_is_byte_identical(self): """Saving with save_format='mistral' writes the native tekken.json byte-for-byte.""" with tempfile.TemporaryDirectory() as tmp_dir: tmp_path = Path(tmp_dir) src_dir = tmp_path / "src" src_dir.mkdir() tekken_path = write_fake_tekken_json(src_dir) out_dir = str(tmp_path / "out") backend = MistralCommonBackend(tokenizer_path=str(tekken_path)) backend.save_pretrained(out_dir, save_format="mistral") saved = Path(out_dir) / "tekken.json" self.assertTrue(saved.exists()) with open(tekken_path, encoding="utf-8") as f: original = json.load(f) with open(saved, encoding="utf-8") as f: copied = json.load(f) self.assertEqual(original, copied) def test_save_pretrained_mistral_format_in_place_resave_succeeds(self): """Resaving into the same directory the tokenizer was loaded from is idempotent: no `shutil.SameFileError`, and the source tekken.json is left byte-for-byte unchanged.""" with tempfile.TemporaryDirectory() as tmp_dir: tekken_path = write_fake_tekken_json(Path(tmp_dir)) original_bytes = tekken_path.read_bytes() backend = MistralCommonBackend(tokenizer_path=str(tekken_path)) result = backend.save_pretrained(tmp_dir, save_format="mistral") self.assertEqual(result, (str(tekken_path),)) self.assertEqual(tekken_path.read_bytes(), original_bytes) def test_save_pretrained_mistral_format_push_to_hub_uploads_copied_tekken_json(self): """The files-timestamps snapshot for push_to_hub must be taken before tekken.json is copied into the (not yet existing) output directory, otherwise `_upload_modified_files` sees it as already-present and never uploads it.""" with tempfile.TemporaryDirectory() as tmp_dir: tmp_path = Path(tmp_dir) src_dir = tmp_path / "src" src_dir.mkdir() tekken_path = write_fake_tekken_json(src_dir) out_dir = tmp_path / "does-not-exist-yet" backend = MistralCommonBackend(tokenizer_path=str(tekken_path)) with ( patch("transformers.tokenization_mistral_common.hf_api") as mock_hf_api, patch.object(MistralCommonBackend, "_upload_modified_files") as mock_upload, ): mock_hf_api.return_value.create_repo.return_value.repo_id = "fake-repo" backend.save_pretrained(str(out_dir), save_format="mistral", push_to_hub=True) mock_upload.assert_called_once() files_timestamps = mock_upload.call_args.args[2] self.assertNotIn("tekken.json", files_timestamps) def test_save_pretrained_rejects_file_path_as_save_directory(self): """save_pretrained on a path that is already a file logs an error and returns without writing anything, matching `PreTrainedTokenizerBase.save_pretrained`.""" with tempfile.TemporaryDirectory() as tmp_dir: tmp_path = Path(tmp_dir) tekken_path = write_fake_tekken_json(tmp_path) backend = MistralCommonBackend(tokenizer_path=str(tekken_path)) existing_file = tmp_path / "already-a-file" existing_file.write_text("sentinel", encoding="utf-8") result = backend.save_pretrained(str(existing_file), save_format="mistral") self.assertIsNone(result) self.assertEqual(existing_file.read_text(encoding="utf-8"), "sentinel") def test_save_pretrained_unknown_format_raises_value_error(self): """save_pretrained rejects any save_format outside 'hf'/'mistral'/None.""" with tempfile.TemporaryDirectory() as tmp_dir: tmp_path = Path(tmp_dir) tekken_path = write_fake_tekken_json(tmp_path) backend = MistralCommonBackend(tokenizer_path=str(tekken_path)) with tempfile.TemporaryDirectory() as out_dir: with self.assertRaises(ValueError) as ctx: backend.save_pretrained(out_dir, save_format="bogus") self.assertIn("Unknown save_format", str(ctx.exception)) def test_encode(self): string = "Hello, world!" # Test 1: # encode with add_special_tokens expected_with_special = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string, bos=True, eos=False) tokens_with_special = self.tokenizer.encode(string, add_special_tokens=True) self.assertEqual(tokens_with_special, expected_with_special) # Test 2: # encode without add_special_tokens expected_without_special = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string, bos=False, eos=False) tokens_without_special = self.tokenizer.encode(string, add_special_tokens=False) self.assertEqual(tokens_without_special, expected_without_special) # Test 3: # encode with add_special_tokens and mode finetuning expected_with_special_ft = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string, bos=True, eos=True) with patch.object(self.tokenizer, "_mode", ValidationMode.finetuning): tokens_with_special_ft = self.tokenizer.encode(string, add_special_tokens=True) self.assertEqual(tokens_with_special_ft, expected_with_special_ft) # Test 4: # encode with return_tensors tokens_with_return_tensors = self.tokenizer.encode(string, add_special_tokens=False, return_tensors="pt") self.assertIsInstance(tokens_with_return_tensors, torch.Tensor) self.assertEqual(tokens_with_return_tensors.tolist()[0], expected_without_special) # Test 5: # encode with max_length tokens_with_max_length = self.tokenizer.encode(string, add_special_tokens=False, max_length=3) self.assertEqual(tokens_with_max_length, expected_without_special[:3]) # Test 6: # encode with padding tokens_with_padding = self.tokenizer.encode( string, add_special_tokens=False, padding=True, pad_to_multiple_of=6 ) expected_padding = [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * ( 6 - len(expected_without_special) % 6 ) + expected_without_special self.assertEqual(tokens_with_padding, expected_padding) for padding in [ False, True, "longest", "max_length", "do_not_pad", PaddingStrategy.LONGEST, PaddingStrategy.MAX_LENGTH, PaddingStrategy.DO_NOT_PAD, ]: tokens_with_padding = self.tokenizer.encode(string, add_special_tokens=False, padding=padding) self.assertEqual(tokens_with_padding, expected_without_special) # For truncation, we use a longer string string_long = ( "Hello world! It is a beautiful day today. The sun is shining brightly and the birds are singing." ) expected_long = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string_long, bos=False, eos=False) # Test 7: # encode with truncation tokens_with_truncation = self.tokenizer.encode( string_long, add_special_tokens=False, truncation=True, max_length=12 ) self.assertEqual(tokens_with_truncation, expected_long[:12]) # Test 8: # encode with padding and truncation tokens_with_padding_and_truncation = self.tokenizer.encode( string_long, add_special_tokens=False, padding=True, pad_to_multiple_of=12, truncation=True, max_length=36 ) expected_long_padding = [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * ( 12 - len(expected_long) % 12 ) + expected_long self.assertEqual(tokens_with_padding_and_truncation, expected_long_padding) # Test 9: # encode empty string self.assertEqual(self.tokenizer.encode("", add_special_tokens=False), []) with self.assertRaises( ValueError, msg="Kwargs [unk_args] are not supported by `MistralCommonBackend.encode`." ): self.tokenizer.encode("Hello, world!", add_special_tokens=True, unk_args="") def test_decode(self): string = "Hello, world!" string_with_space = "Hello, world !" tokens_ids = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string, bos=True, eos=True) tokens_ids_with_space = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode( string_with_space, bos=True, eos=True ) # Test 1: # decode with and without skip_special_tokens self.assertEqual(self.tokenizer.decode(tokens_ids, skip_special_tokens=True), string) self.assertEqual(self.tokenizer.decode(tokens_ids, skip_special_tokens=False), "<s>" + string + "</s>") self.assertEqual(self.tokenizer.decode(tokens_ids_with_space, skip_special_tokens=True), string_with_space) # Test 2: # decode with clean_up_tokenization_spaces self.assertEqual( self.tokenizer.decode(tokens_ids_with_space, skip_special_tokens=True, clean_up_tokenization_spaces=True), "Hello, world!", ) # Test 3: # decode one token self.assertEqual(self.tokenizer.decode(tokens_ids[0], skip_special_tokens=False), "<s>") # Test 4: # decode numpy self.assertEqual(self.tokenizer.decode(np.array(tokens_ids), skip_special_tokens=True), string) # Test 5: # decode empty string self.assertEqual(self.tokenizer.decode([], skip_special_tokens=True), "") # Test 6: # decode with unsupported kwargs with self.assertRaises( ValueError, msg="Kwargs [unk_args] are not supported by `MistralCommonBackend.decode`." ): self.tokenizer.decode(tokens_ids, skip_special_tokens=False, unk_args="") def test_decode_on_batch(self): string = "Hello, world!" string_with_space = "Hello, world !" batch_tokens_ids = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string, bos=True, eos=True), self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string_with_space, bos=True, eos=True), ] # Test 1: # batch_decode with and without skip_special_tokens self.assertEqual( self.tokenizer.decode(batch_tokens_ids, skip_special_tokens=True), [string, string_with_space], ) self.assertEqual( self.tokenizer.decode(batch_tokens_ids, skip_special_tokens=False), ["<s>" + string + "</s>", "<s>" + string_with_space + "</s>"], ) self.assertEqual( self.tokenizer.decode(batch_tokens_ids, skip_special_tokens=True, clean_up_tokenization_spaces=True), ["Hello, world!", "Hello, world!"], ) # Test 3: # decode numpy self.assertEqual( self.tokenizer.decode( np.array(batch_tokens_ids), skip_special_tokens=True, clean_up_tokenization_spaces=True ), ["Hello, world!", "Hello, world!"], ) def test_decode_transcription_mode(self): # in the specific case of Voxtral, the added f"lang:xx" (always a two char language code since it follows ISO 639-1 alpha-2 format) # is not considered as a special token by mistral-common and is encoded/ decoded as normal text. # we made the explicit choice of skipping "lang:xx" it to ease users life, see `[~MistralCommonBackend.decode]` expected_string = "lang:en[TRANSCRIBE]" openai_transcription_request = { "model": None, "language": "en", "file": io.BytesIO(base64.b64decode(AUDIO_BASE_64)), } transcription_request = TranscriptionRequest.from_openai(openai_transcription_request) tokenized_transcription_request = self.ref_tokenizer_audio.encode_transcription(transcription_request) # without skip_special_tokens self.assertEqual( self.tokenizer_audio.decode(tokenized_transcription_request.tokens, skip_special_tokens=False)[ -len(expected_string) : ], expected_string, ) # with skip_special_tokens self.assertEqual(self.tokenizer.decode(tokenized_transcription_request.tokens, skip_special_tokens=True), "") def test_batch_decode(self): string = "Hello, world!" string_with_space = "Hello, world !" batch_tokens_ids = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string, bos=True, eos=True), self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string_with_space, bos=True, eos=True), ] # Test 1: # batch_decode with and without skip_special_tokens self.assertEqual( self.tokenizer.batch_decode(batch_tokens_ids, skip_special_tokens=True), [string, string_with_space], ) self.assertEqual( self.tokenizer.batch_decode(batch_tokens_ids, skip_special_tokens=False), ["<s>" + string + "</s>", "<s>" + string_with_space + "</s>"], ) self.assertEqual( self.tokenizer.batch_decode(batch_tokens_ids, skip_special_tokens=True, clean_up_tokenization_spaces=True), ["Hello, world!", "Hello, world!"], ) # Test 3: # decode numpy self.assertEqual( self.tokenizer.batch_decode( np.array(batch_tokens_ids), skip_special_tokens=True, clean_up_tokenization_spaces=True ), ["Hello, world!", "Hello, world!"], ) # Test 4: # decode empty list self.assertEqual(self.tokenizer.batch_decode([], skip_special_tokens=True), [""]) self.assertEqual( self.tokenizer.batch_decode([batch_tokens_ids[0], []], skip_special_tokens=True), [string, ""] ) # Test 5: # batch_decode with unsupported kwargs with self.assertRaises( ValueError, msg="Kwargs [unk_args] are not supported by `MistralCommonBackend.batch_decode`." ): self.tokenizer.batch_decode(batch_tokens_ids, skip_special_tokens=False, unk_args="") def test_convert_ids_to_tokens(self): # Test 1: # with skip_special_tokens=False ids = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode("Hello world!", bos=True, eos=True) expected_tokens = [self.ref_tokenizer.instruct_tokenizer.tokenizer.id_to_piece(id) for id in ids] tokens = self.tokenizer.convert_ids_to_tokens(ids, skip_special_tokens=False) self.assertEqual(tokens, expected_tokens) token = self.tokenizer.convert_ids_to_tokens(ids[0], skip_special_tokens=False) self.assertEqual(token, expected_tokens[0]) # Test 2: # with skip_special_tokens=True expected_tokens = expected_tokens[1:-1] tokens = self.tokenizer.convert_ids_to_tokens(ids, skip_special_tokens=True) self.assertEqual(tokens, expected_tokens) # Test 3: # with empty list tokens = self.tokenizer.convert_ids_to_tokens([]) self.assertEqual(tokens, []) with self.assertRaises(ValueError): self.tokenizer.convert_ids_to_tokens(ids[0], skip_special_tokens=True) token = self.tokenizer.convert_ids_to_tokens(ids[1], skip_special_tokens=True) self.assertEqual(token, expected_tokens[0]) def test_convert_tokens_to_ids(self): tokens = ["Hello", "world", "!"] expected_ids = [self._ref_piece_to_id(token) for token in tokens] # Test 1: # list of tokens ids = self.tokenizer.convert_tokens_to_ids(tokens) self.assertEqual(ids, expected_ids) # Test 2: # single token id = self.tokenizer.convert_tokens_to_ids(tokens[0]) self.assertEqual(id, expected_ids[0]) self.assertEqual(id, self.tokenizer.convert_tokens_to_ids(tokens[0])) # Test 3: # with empty list ids = self.tokenizer.convert_tokens_to_ids([]) self.assertEqual(ids, []) def test_tokenize(self): string = "Hello world!" # Test 1: # with string expected_tokens = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.id_to_piece(id) for id in self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(string, bos=False, eos=False) ] tokens = self.tokenizer.tokenize(string) self.assertEqual(tokens, expected_tokens) # Test 2: # with empty string tokens = self.tokenizer.tokenize("") self.assertEqual(tokens, []) with self.assertRaises( ValueError, msg="Kwargs [add_special_tokens] are not supported by `MistralCommonBackend.tokenize`." ): self.tokenizer.tokenize(string, add_special_tokens=True) def test_get_special_tokens_mask(self): # Test 1: # mode test with special ids = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode("Hello world!", bos=True, eos=False) expected_mask = [1 if id in self.ref_special_ids else 0 for id in ids] mask = self.tokenizer.get_special_tokens_mask(ids, already_has_special_tokens=True) self.assertEqual(mask, expected_mask) # Test 2: # mode finetuning with special with patch.object(self.tokenizer, "_mode", ValidationMode.finetuning): ids = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode("Hello world!", bos=True, eos=True) expected_mask = [1 if id in self.ref_special_ids else 0 for id in ids] mask = self.tokenizer.get_special_tokens_mask(ids, already_has_special_tokens=True) self.assertEqual(mask, expected_mask) # Test 3: # mode test without special ids = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode("Hello world!", bos=False, eos=False) expected_mask = [1] + [0 for id in ids] mask = self.tokenizer.get_special_tokens_mask(ids, already_has_special_tokens=False) self.assertEqual(mask, expected_mask) # Test 4: # mode finetuning without special with patch.object(self.tokenizer, "_mode", ValidationMode.finetuning): ids = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode("Hello world!", bos=False, eos=False) expected_mask = [1] + [0 for id in ids] + [1] mask = self.tokenizer.get_special_tokens_mask(ids, already_has_special_tokens=False) self.assertEqual(mask, expected_mask) # Test 5: # token_ids_1 not None should raise an error with self.assertRaises(ValueError): self.tokenizer.get_special_tokens_mask(ids, token_ids_1=ids) def test_pad_batch_encoding_input(self): # Test 1: # padding and default values def get_batch_encoding(): return self.tokenizer("Hello world!", return_special_tokens_mask=True) batch_encoding = get_batch_encoding() for padding in [ False, True, "longest", "max_length", "do_not_pad", PaddingStrategy.LONGEST, PaddingStrategy.MAX_LENGTH, PaddingStrategy.DO_NOT_PAD, ]: padded_batch_encoding = self.tokenizer.pad(get_batch_encoding(), padding=padding) self.assertEqual(padded_batch_encoding, batch_encoding) # Test 2: # padding_strategy="max_length" or PaddingStrategy.MAX_LENGTH and max_length for padding in ["max_length", PaddingStrategy.MAX_LENGTH]: padded_batch_encoding = self.tokenizer.pad(get_batch_encoding(), padding=padding, max_length=12) self.assertEqual( padded_batch_encoding["input_ids"], [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"])) + batch_encoding["input_ids"], ) self.assertEqual( padded_batch_encoding["attention_mask"], [0] * (12 - len(batch_encoding["input_ids"])) + batch_encoding["attention_mask"], ) self.assertEqual( padded_batch_encoding["special_tokens_mask"], [1] * (12 - len(batch_encoding["input_ids"])) + batch_encoding["special_tokens_mask"], ) # Test 3: # padding_strategy=True or "longest" or PaddingStrategy.LONGEST or "max_length" or PaddingStrategy.MAX_LENGTH and pad_to_multiple_of 16 for padding in [True, "longest", PaddingStrategy.LONGEST]: padded_batch_encoding = self.tokenizer.pad(get_batch_encoding(), padding=padding, pad_to_multiple_of=16) self.assertEqual( padded_batch_encoding["input_ids"], [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (16 - len(batch_encoding["input_ids"])) + batch_encoding["input_ids"], ) self.assertEqual( padded_batch_encoding["attention_mask"], [0] * (16 - len(batch_encoding["input_ids"])) + batch_encoding["attention_mask"], ) self.assertEqual( padded_batch_encoding["special_tokens_mask"], [1] * (16 - len(batch_encoding["input_ids"])) + batch_encoding["special_tokens_mask"], ) # Test 4: # padding_side="right" right_tokenizer = MistralCommonBackend.from_pretrained( self.repo_id, local_files_only=self.local_files_only, padding_side="right", revision=None, ) right_paddings = [ right_tokenizer.pad(get_batch_encoding(), padding="max_length", max_length=12), self.tokenizer.pad(get_batch_encoding(), padding="max_length", max_length=12, padding_side="right"), ] for padded_batch_encoding in right_paddings: self.assertEqual( padded_batch_encoding["input_ids"], batch_encoding["input_ids"] + [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"])), ) self.assertEqual( padded_batch_encoding["attention_mask"], batch_encoding["attention_mask"] + [0] * (12 - len(batch_encoding["input_ids"])), ) self.assertEqual( padded_batch_encoding["special_tokens_mask"], batch_encoding["special_tokens_mask"] + [1] * (12 - len(batch_encoding["input_ids"])), ) # Test 5: # return_attention_mask=False padded_batch_encoding = self.tokenizer.pad( get_batch_encoding(), padding="max_length", max_length=12, return_attention_mask=False ) self.assertEqual( padded_batch_encoding["input_ids"], [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"])) + batch_encoding["input_ids"], ) self.assertEqual(padded_batch_encoding["attention_mask"], batch_encoding["attention_mask"]) self.assertEqual( padded_batch_encoding["special_tokens_mask"], [1] * (12 - len(batch_encoding["input_ids"])) + batch_encoding["special_tokens_mask"], ) # Test 6: # return_tensors="pt" or "np" for return_tensors in ["pt", "np"]: padded_batch_encoding = self.tokenizer.pad( get_batch_encoding(), padding="max_length", max_length=12, return_tensors=return_tensors ) self.assertEqual(padded_batch_encoding["input_ids"].shape, torch.Size((12,))) self.assertEqual(padded_batch_encoding["attention_mask"].shape, torch.Size((12,))) self.assertEqual(padded_batch_encoding["special_tokens_mask"].shape, torch.Size((12,))) def test_list_batch_encoding_input(self): def get_batch_encoding(): return self.tokenizer(["Hello world!", "Hello world! Longer sentence."], return_special_tokens_mask=True) # Test 1: # padding=True or "longest" or PaddingStrategy.LONGEST batch_encoding = get_batch_encoding() for padding in [ True, "longest", PaddingStrategy.LONGEST, ]: padded_batch_encoding = self.tokenizer.pad(get_batch_encoding(), padding=padding) self.assertEqual( padded_batch_encoding["input_ids"], [ [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (len(batch_encoding["input_ids"][1]) - len(batch_encoding["input_ids"][0])) + batch_encoding["input_ids"][0], batch_encoding["input_ids"][1], ], ) self.assertEqual( padded_batch_encoding["attention_mask"], [ [0] * (len(batch_encoding["input_ids"][1]) - len(batch_encoding["input_ids"][0])) + batch_encoding["attention_mask"][0], batch_encoding["attention_mask"][1], ], ) self.assertEqual( padded_batch_encoding["special_tokens_mask"], [ [1] * (len(batch_encoding["input_ids"][1]) - len(batch_encoding["input_ids"][0])) + batch_encoding["special_tokens_mask"][0], batch_encoding["special_tokens_mask"][1], ], ) # Test 2: # padding_strategy="max_length" or PaddingStrategy.MAX_LENGTH and max_length for padding in ["max_length", PaddingStrategy.MAX_LENGTH]: padded_batch_encoding = self.tokenizer.pad(get_batch_encoding(), padding=padding, max_length=12) self.assertEqual( padded_batch_encoding["input_ids"], [ [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"][0])) + batch_encoding["input_ids"][0], [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"][1])) + batch_encoding["input_ids"][1], ], ) self.assertEqual( padded_batch_encoding["attention_mask"], [ [0] * (12 - len(batch_encoding["input_ids"][0])) + batch_encoding["attention_mask"][0], [0] * (12 - len(batch_encoding["input_ids"][1])) + batch_encoding["attention_mask"][1], ], ) self.assertEqual( padded_batch_encoding["special_tokens_mask"], [ [1] * (12 - len(batch_encoding["input_ids"][0])) + batch_encoding["special_tokens_mask"][0], [1] * (12 - len(batch_encoding["input_ids"][1])) + batch_encoding["special_tokens_mask"][1], ], ) # Test 3: # padding_strategy=True or "longest" or PaddingStrategy.LONGEST or "max_length" or PaddingStrategy.MAX_LENGTH and pad_to_multiple_of 16 for padding in [True, "longest", PaddingStrategy.LONGEST]: padded_batch_encoding = self.tokenizer.pad(get_batch_encoding(), padding=padding, pad_to_multiple_of=16) self.assertEqual( padded_batch_encoding["input_ids"], [ [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (16 - len(batch_encoding["input_ids"][0])) + batch_encoding["input_ids"][0], [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (16 - len(batch_encoding["input_ids"][1])) + batch_encoding["input_ids"][1], ], ) self.assertEqual( padded_batch_encoding["attention_mask"], [ [0] * (16 - len(batch_encoding["input_ids"][0])) + batch_encoding["attention_mask"][0], [0] * (16 - len(batch_encoding["input_ids"][1])) + batch_encoding["attention_mask"][1], ], ) self.assertEqual( padded_batch_encoding["special_tokens_mask"], [ [1] * (16 - len(batch_encoding["input_ids"][0])) + batch_encoding["special_tokens_mask"][0], [1] * (16 - len(batch_encoding["input_ids"][1])) + batch_encoding["special_tokens_mask"][1], ], ) # Test 4: # padding_side="right" right_tokenizer = MistralCommonBackend.from_pretrained( self.repo_id, local_files_only=self.local_files_only, padding_side="right", revision=None, ) right_paddings = [ right_tokenizer.pad(get_batch_encoding(), padding="max_length", max_length=12), self.tokenizer.pad(get_batch_encoding(), padding="max_length", max_length=12, padding_side="right"), ] for padded_batch_encoding in right_paddings: self.assertEqual( padded_batch_encoding["input_ids"], [ batch_encoding["input_ids"][0] + [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"][0])), batch_encoding["input_ids"][1] + [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"][1])), ], ) self.assertEqual( padded_batch_encoding["attention_mask"], [ batch_encoding["attention_mask"][0] + [0] * (12 - len(batch_encoding["input_ids"][0])), batch_encoding["attention_mask"][1] + [0] * (12 - len(batch_encoding["input_ids"][1])), ], ) self.assertEqual( padded_batch_encoding["special_tokens_mask"], [ batch_encoding["special_tokens_mask"][0] + [1] * (12 - len(batch_encoding["input_ids"][0])), batch_encoding["special_tokens_mask"][1] + [1] * (12 - len(batch_encoding["input_ids"][1])), ], ) # Test 5: # return_attention_mask=False padded_batch_encoding = self.tokenizer.pad( get_batch_encoding(), padding="max_length", max_length=12, return_attention_mask=False ) self.assertEqual( padded_batch_encoding["input_ids"], [ [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"][0])) + batch_encoding["input_ids"][0], [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] * (12 - len(batch_encoding["input_ids"][1])) + batch_encoding["input_ids"][1], ], ) self.assertEqual(padded_batch_encoding["attention_mask"], batch_encoding["attention_mask"]) self.assertEqual( padded_batch_encoding["special_tokens_mask"], [ [1] * (12 - len(batch_encoding["input_ids"][0])) + batch_encoding["special_tokens_mask"][0], [1] * (12 - len(batch_encoding["input_ids"][1])) + batch_encoding["special_tokens_mask"][1], ], ) # Test 6: # return_tensors="pt" or "np" for return_tensors in ["pt", "np"]: padded_batch_encoding = self.tokenizer.pad( get_batch_encoding(), padding="max_length", max_length=12, return_tensors=return_tensors ) self.assertEqual(padded_batch_encoding["input_ids"].shape, torch.Size((2, 12))) self.assertEqual(padded_batch_encoding["attention_mask"].shape, torch.Size((2, 12))) self.assertEqual(padded_batch_encoding["special_tokens_mask"].shape, torch.Size((2, 12))) def test_truncate_sequences(self): # Test 1: # truncation_strategy="longest_first" or TruncationStrategy.LONGEST_FIRST text = "Hello world!" ids = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(text, bos=True, eos=True) for truncation in ["longest_first", TruncationStrategy.LONGEST_FIRST]: for num_tokens_to_remove in [0, 2]: tokens, none, overflowing_tokens = self.tokenizer.truncate_sequences( ids, truncation_strategy=truncation, num_tokens_to_remove=num_tokens_to_remove ) self.assertEqual(tokens, ids[:-num_tokens_to_remove] if num_tokens_to_remove > 0 else ids) self.assertIsNone(none) self.assertEqual(overflowing_tokens, ids[-num_tokens_to_remove:] if num_tokens_to_remove > 0 else []) # Test 2: # truncation_strategy="only_first" or "only_second" or TruncationStrategy.ONLY_FIRST or TruncationStrategy.ONLY_SECOND # Should raise a ValueError for truncation in ["only_first", "only_second", TruncationStrategy.ONLY_FIRST, TruncationStrategy.ONLY_SECOND]: with self.assertRaises(ValueError): self.tokenizer.truncate_sequences(ids, truncation_strategy=truncation, num_tokens_to_remove=1) # Test 3: # truncation_strategy="do_not_truncate" or TruncationStrategy.DO_NOT_TRUNCATE for truncation in ["do_not_truncate", TruncationStrategy.DO_NOT_TRUNCATE]: tokens, none, overflowing_tokens = self.tokenizer.truncate_sequences( ids, truncation_strategy=truncation, num_tokens_to_remove=1 ) self.assertEqual(tokens, ids) self.assertIsNone(none) self.assertEqual(overflowing_tokens, []) # Test 4: # pair_ids is not None # Should raise a ValueError with self.assertRaises(ValueError): self.tokenizer.truncate_sequences( ids, pair_ids=ids, truncation_strategy="longest_first", num_tokens_to_remove=1 ) # Test 5: # stride for stride in [0, 2]: tokens, none, overflowing_tokens = self.tokenizer.truncate_sequences( ids, truncation_strategy="longest_first", num_tokens_to_remove=2, stride=stride ) self.assertEqual(tokens, ids[:-2]) self.assertIsNone(none) self.assertEqual(overflowing_tokens, ids[-2 - stride :]) # Test 6: # truncation_side="left" left_tokenizer = MistralCommonBackend.from_pretrained( self.repo_id, local_files_only=self.local_files_only, truncation_side="left", revision=None, ) tokens, none, overflowing_tokens = left_tokenizer.truncate_sequences( ids, truncation_strategy="longest_first", num_tokens_to_remove=2 ) self.assertEqual(tokens, ids[2:]) self.assertIsNone(none) self.assertEqual(overflowing_tokens, ids[:2]) def test_apply_chat_template_basic(self): conversation = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can I help you?"}, {"role": "user", "content": "What is the capital of France?"}, ] expected_tokenized = self.ref_tokenizer.encode_chat_completion(ChatCompletionRequest.from_openai(conversation)) # Test 1: # with tokenize self.assertEqual( self.tokenizer.apply_chat_template(conversation, tokenize=False, reasoning_effort=None), self.ref_tokenizer.decode(tokens=expected_tokenized.tokens, special_token_policy=SpecialTokenPolicy.KEEP), ) # Test 2: # without tokenize self.assertEqual( self.tokenizer.apply_chat_template(conversation, tokenize=True, reasoning_effort=None).input_ids, expected_tokenized.tokens, ) def test_apply_chat_template_continue_final_message(self): conversation = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can I help you?"}, {"role": "user", "content": "What is the capital of France?"}, {"role": "assistant", "content": "Paris"}, ] expected_tokenized = self.ref_tokenizer.encode_chat_completion( ChatCompletionRequest.from_openai(conversation, continue_final_message=True) ) self.assertEqual( self.tokenizer.apply_chat_template(conversation, tokenize=False, continue_final_message=True), self.ref_tokenizer.decode(tokens=expected_tokenized.tokens, special_token_policy=SpecialTokenPolicy.KEEP), ) self.assertEqual( self.tokenizer.apply_chat_template(conversation, tokenize=True, continue_final_message=True).input_ids, expected_tokenized.tokens, ) with self.assertRaises(InvalidMessageStructureException): self.tokenizer.apply_chat_template(conversation, tokenize=False, continue_final_message=False) def test_apply_chat_template_with_add_generation_prompt(self): conversation = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, ] # Test 1: # with add_generation_prompt for add_generation_prompt in [False, True]: expected_tokenized = self.ref_tokenizer.encode_chat_completion( ChatCompletionRequest.from_openai(conversation) ) token_outputs = self.tokenizer.apply_chat_template( conversation, tokenize=True, add_generation_prompt=add_generation_prompt ) self.assertEqual(token_outputs.input_ids, expected_tokenized.tokens) # Test 2: # with continue_final_message with self.assertRaises(ValueError): self.tokenizer.apply_chat_template( conversation, tokenize=True, add_generation_prompt=True, continue_final_message=True ) # Test 3: # with last message with assistant role conversation = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "asistant", "content": "Hey!"}, ] with self.assertRaises(ValueError): self.tokenizer.apply_chat_template(conversation, tokenize=True, add_generation_prompt=True) def test_apply_chat_template_with_tools(self): conversation = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can I help you?"}, {"role": "user", "content": "What is the temperature in Paris?"}, { "role": "assistant", "tool_calls": [ { "id": "azerty123", "function": { "name": "get_current_weather", "arguments": {"location": "Paris", "format": "text", "unit": "celsius"}, }, } ], }, {"role": "tool", "name": "get_current_weather", "content": "22", "tool_call_id": "azerty123"}, ] tools = [ { "type": "function", "function": { "name": "get_current_weather", "description": "Get the current weather in a given location", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "The city and state, e.g. San Francisco, CA", "required": ["location"], }, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}, "format": { "type": "string", "enum": ["text", "json"], "description": "The format of the response", "required": ["format"], }, }, }, }, } ] expected_tokenized = self.ref_tokenizer.encode_chat_completion( ChatCompletionRequest.from_openai(conversation, tools) ) self.assertEqual( self.tokenizer.apply_chat_template(conversation, tools=tools, tokenize=False), self.ref_tokenizer.decode(tokens=expected_tokenized.tokens, special_token_policy=SpecialTokenPolicy.KEEP), ) def test_apply_chat_template_with_image(self): ref_conversation = conversation = [ {"role": "system", "content": "You are a helpful assistant."}, { "role": "user", "content": [ {"type": "text", "text": "What is this?"}, { "type": "image_url", "image_url": {"url": IMG_URL}, }, ], }, ] expected_tokenized = self.ref_tokenizer.encode_chat_completion( ChatCompletionRequest.from_openai(ref_conversation) ) image_contents = [ { "type": "image_url", "image_url": {"url": IMG_URL}, }, { "type": "image", "url": IMG_URL, }, {"type": "image", "base64": IMG_BASE_64}, ] for image_content in image_contents: conversation = [ {"role": "system", "content": "You are a helpful assistant."}, { "role": "user", "content": [{"type": "text", "text": "What is this?"}, image_content], }, ] output = self.tokenizer.apply_chat_template(conversation).input_ids self.assertEqual(output, expected_tokenized.tokens) output_dict = self.tokenizer.apply_chat_template(conversation, tokenize=True) self.assertEqual(output_dict["input_ids"], expected_tokenized.tokens) self.assertEqual(len(output_dict["pixel_values"]), len(expected_tokenized.images)) for o, e in zip(output_dict["pixel_values"], expected_tokenized.images): self.assertTrue(np.allclose(o, e)) # Test with return_tensors="pt" output_dict_pt = self.tokenizer.apply_chat_template(conversation, tokenize=True, return_tensors="pt") self.assertEqual(output_dict_pt["input_ids"].tolist()[0], expected_tokenized.tokens) expected_images_pt_tensor = torch.from_numpy(np.stack(expected_tokenized.images)) self.assertTrue(torch.allclose(output_dict_pt["pixel_values"], expected_images_pt_tensor)) self.assertIn("image_sizes", output_dict_pt) self.assertIsInstance(output_dict_pt["image_sizes"], torch.Tensor) self.assertEqual(output_dict_pt["image_sizes"].shape, torch.Size([1, 2])) # Test with return_tensors="np" output_dict_np = self.tokenizer.apply_chat_template(conversation, tokenize=True, return_tensors="np") self.assertEqual(output_dict_np["input_ids"].tolist()[0], expected_tokenized.tokens) expected_images_np_array = np.stack(expected_tokenized.images) self.assertTrue(np.allclose(output_dict_np["pixel_values"], expected_images_np_array)) self.assertIn("image_sizes", output_dict_np) self.assertIsInstance(output_dict_np["image_sizes"], np.ndarray) self.assertEqual(output_dict_np["image_sizes"].shape, (1, 2)) # Test with return_tensors=None (default - Python lists) output_dict_default = self.tokenizer.apply_chat_template(conversation, tokenize=True) self.assertEqual(output_dict_default["input_ids"], expected_tokenized.tokens) self.assertEqual(len(output_dict_default["pixel_values"]), len(expected_tokenized.images)) for o, e in zip(output_dict_default["pixel_values"], expected_tokenized.images): self.assertTrue(np.allclose(o, e)) self.assertIn("image_sizes", output_dict_default) self.assertIsInstance(output_dict_default["image_sizes"], list) self.assertEqual(len(output_dict_default["image_sizes"]), 1) # Check pt, np, list are equals self.assertEqual(output_dict_pt["image_sizes"].tolist(), output_dict_np["image_sizes"].tolist()) self.assertEqual(output_dict_pt["image_sizes"].tolist(), output_dict_default["image_sizes"]) actual_height, actual_width = output_dict_pt["image_sizes"].tolist()[0] self.assertEqual(actual_height, 308, f"Expected height 308, got {actual_height}") self.assertEqual(actual_width, 224, f"Expected width 224, got {actual_width}") def test_apply_chat_template_with_audio(self): ref_conversation = conversation = [ { "role": "user", "content": [ {"type": "text", "text": "What is this?"}, { "type": "input_audio", "input_audio": { "data": AUDIO_BASE_64, "format": "wav", }, }, ], }, ] expected_tokenized = self.ref_tokenizer_audio.encode_chat_completion( ChatCompletionRequest.from_openai(ref_conversation) ) audio_contents = [ { "type": "audio", "url": AUDIO_URL, }, { "type": "audio", "path": AUDIO_URL, }, {"type": "audio", "base64": AUDIO_BASE_64}, ] for audio_content in audio_contents: conversation = [ { "role": "user", "content": [{"type": "text", "text": "What is this?"}, audio_content], }, ] output = self.tokenizer_audio.apply_chat_template(conversation, tokenize=True).input_ids self.assertEqual(output, expected_tokenized.tokens) output_dict = self.tokenizer_audio.apply_chat_template(conversation, tokenize=True, return_dict=True) self.assertEqual(output_dict["input_ids"], expected_tokenized.tokens) self.assertEqual(len(output_dict["audio"]), len(expected_tokenized.audios)) for o, e in zip(output_dict["audio"], expected_tokenized.audios): audio_array = e.audio_array self.assertTrue(np.allclose(o, audio_array)) with self.assertRaises(NotImplementedError): output_dict = self.tokenizer_audio.apply_chat_template( conversation, tokenize=True, return_dict=True, return_tensors="pt" ) def test_apply_chat_template_with_truncation(self): conversation = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can I help you?"}, {"role": "user", "content": "What is the capital of France?"}, ] expected_tokenized = self.ref_tokenizer.encode_chat_completion(ChatCompletionRequest.from_openai(conversation)) # Test 1: # with truncation self.assertEqual( self.tokenizer.apply_chat_template(conversation, tokenize=True, truncation=True, max_length=20).input_ids, expected_tokenized.tokens[:20], ) # Test 2: # without truncation self.assertEqual( self.tokenizer.apply_chat_template(conversation, tokenize=True, truncation=False, max_length=20).input_ids, expected_tokenized.tokens, ) # Test 3: # assert truncation is boolean with self.assertRaises(TypeError): self.tokenizer.apply_chat_template( conversation, tokenize=True, truncation=TruncationStrategy.LONGEST_FIRST, max_length=20 ) def test_batch_apply_chat_template(self): conversations = [ [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can I help you?"}, { "role": "user", "content": [ {"type": "text", "text": "What is this?"}, { "type": "image_url", "image_url": {"url": IMG_URL}, }, ], }, ], [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can I help you?"}, {"role": "user", "content": "What is the temperature in Paris?"}, { "role": "assistant", "tool_calls": [ { "id": "azerty123", "function": { "name": "get_current_weather", "arguments": {"location": "Paris", "format": "text", "unit": "celsius"}, }, } ], }, {"role": "tool", "name": "get_current_weather", "content": "22", "tool_call_id": "azerty123"}, ], ] tools = [ { "type": "function", "function": { "name": "get_current_weather", "description": "Get the current weather in a given location", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "The city and state, e.g. San Francisco, CA", "required": ["location"], }, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}, "format": { "type": "string", "enum": ["text", "json"], "description": "The format of the response", "required": ["format"], }, }, }, }, } ] expected_tokenized = [ self.ref_tokenizer.encode_chat_completion(ChatCompletionRequest.from_openai(conversation, tools=tools)) for conversation in conversations ] text_outputs = self.tokenizer.apply_chat_template(conversations, tools=tools, tokenize=False) token_outputs = self.tokenizer.apply_chat_template(conversations, tools=tools, tokenize=True).input_ids self.assertEqual(len(text_outputs), len(token_outputs)) self.assertEqual(len(text_outputs), len(expected_tokenized)) for text, token, expected in zip(text_outputs, token_outputs, expected_tokenized): self.assertEqual( text, self.ref_tokenizer.decode(tokens=expected.tokens, special_token_policy=SpecialTokenPolicy.KEEP) ) self.assertEqual(token, expected.tokens) def test_batch_apply_chat_template_images(self): conversations = [ [ {"role": "system", "content": "You are a helpful assistant."}, { "role": "user", "content": [ {"type": "text", "text": "What is this?"}, { "type": "image_url", "image_url": {"url": IMG_URL}, }, ], }, ], [ {"role": "system", "content": "You are a helpful assistant."}, { "role": "user", "content": [ {"type": "text", "text": "What is this?"}, { "type": "image", "url": IMG_URL, }, ], }, ], [ {"role": "system", "content": "You are a helpful assistant."}, { "role": "user", "content": [ {"type": "text", "text": "What is this?"}, {"type": "image", "base64": IMG_BASE_64}, ], }, ], ] ref_conversation = [ {"role": "system", "content": "You are a helpful assistant."}, { "role": "user", "content": [ {"type": "text", "text": "What is this?"}, { "type": "image_url", "image_url": {"url": IMG_URL}, }, ], }, ] expected_tokenized = self.ref_tokenizer.encode_chat_completion( ChatCompletionRequest.from_openai(ref_conversation) ) output = self.tokenizer.apply_chat_template(conversations, tokenize=True).input_ids self.assertEqual(output, [expected_tokenized.tokens] * 3) output = self.tokenizer.apply_chat_template(conversations, tokenize=True, return_dict=True) self.assertEqual(output["input_ids"], [expected_tokenized.tokens] * 3) self.assertEqual(len(output["pixel_values"]), len(expected_tokenized.images) * 3) for o, e in zip(output["pixel_values"], [expected_tokenized.images] * 3): self.assertTrue(np.allclose(o, e)) output = self.tokenizer.apply_chat_template( conversations, tokenize=True, return_dict=True, return_tensors="pt" ) self.assertEqual(output["input_ids"].tolist(), [expected_tokenized.tokens] * 3) self.assertEqual(output["input_ids"].shape[0], len(expected_tokenized.images) * 3) expected_images_pt_tensor = torch.from_numpy(np.stack([expected_tokenized.images] * 3)) self.assertTrue(torch.allclose(output["pixel_values"], expected_images_pt_tensor)) output = self.tokenizer.apply_chat_template( conversations, tokenize=True, return_dict=True, return_tensors="np" ) self.assertEqual(output["input_ids"].tolist(), [expected_tokenized.tokens] * 3) self.assertTrue(np.allclose(output["pixel_values"], np.array([expected_tokenized.images] * 3))) def test_batch_apply_chat_template_with_continue_final_message(self): conversations = [ [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can "}, ], [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "assistant", "content": "Hello! How can I help you? Ou préférez vous "}, ], ] # Test 1: # with continue_final_message expected_tokenized = [ self.ref_tokenizer.encode_chat_completion( ChatCompletionRequest.from_openai(conversation, continue_final_message=True) ) for conversation in conversations ] token_outputs = self.tokenizer.apply_chat_template( conversations, tokenize=True, continue_final_message=True ).input_ids for output, expected in zip(token_outputs, expected_tokenized): self.assertEqual(output, expected.tokens) # Test 2: # without continue_final_message with self.assertRaises(InvalidMessageStructureException): self.tokenizer.apply_chat_template( conversations, tokenize=False, continue_final_message=False, ) # Test 3: # with continue_final_message and last role is not assistant with self.assertRaises(InvalidMessageStructureException): self.tokenizer.apply_chat_template( conversation=[ [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, ] ], tokenize=True, continue_final_message=True, ) def test_batch_apply_chat_template_with_add_generation_prompt(self): conversations = [ [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, ], [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, ], ] # Test 1: # with add_generation_prompt for add_generation_prompt in [False, True]: expected_tokenized = [ self.ref_tokenizer.encode_chat_completion(ChatCompletionRequest.from_openai(conversation)) for conversation in conversations ] token_outputs = self.tokenizer.apply_chat_template( conversations, tokenize=True, add_generation_prompt=add_generation_prompt ).input_ids for output, expected in zip(token_outputs, expected_tokenized): self.assertEqual(output, expected.tokens) # Test 2: # with continue_final_message with self.assertRaises(ValueError): self.tokenizer.apply_chat_template( conversations, tokenize=True, add_generation_prompt=True, continue_final_message=True ) # Test 3: # with last message with assistant role conversations = [ [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, {"role": "asistant", "content": "Hey!"}, ], [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hi!"}, ], ] with self.assertRaises(ValueError): self.tokenizer.apply_chat_template(conversations, tokenize=True, add_generation_prompt=True) def test_batch_apply_chat_template_with_truncation( self, ): # Test 1: # with truncation token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=True, truncation=True, max_length=20 ).input_ids for output, expected in zip(token_outputs, self.tokenized_fixture_conversations): self.assertEqual(output, expected.tokens[:20]) # Test 2: # without truncation token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=True, truncation=False, max_length=20 ).input_ids self.assertEqual(len(token_outputs), len(self.tokenized_fixture_conversations)) for output, expected in zip(token_outputs, self.tokenized_fixture_conversations): self.assertEqual(output, expected.tokens) # Test 3: # assert truncation is boolean with self.assertRaises(TypeError): self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=True, truncation=TruncationStrategy.LONGEST_FIRST, max_length=20 ) def test_batch_apply_chat_template_with_padding( self, ): for padding in [True, "max_length", PaddingStrategy.LONGEST, PaddingStrategy.MAX_LENGTH]: if padding == PaddingStrategy.MAX_LENGTH: # No padding if no max length is provided token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, padding=padding, return_dict=False ) self.assertEqual(len(token_outputs), len(self.tokenized_fixture_conversations)) for output, expected in zip(token_outputs, self.tokenized_fixture_conversations): self.assertEqual(output, expected.tokens) max_length = 20 if padding == PaddingStrategy.MAX_LENGTH else None token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=True, padding=padding, max_length=max_length, return_dict=False ) if padding != PaddingStrategy.MAX_LENGTH: longest = max(len(tokenized.tokens) for tokenized in self.tokenized_fixture_conversations) self.assertEqual(len(token_outputs), len(self.tokenized_fixture_conversations)) for output, expected in zip(token_outputs, self.tokenized_fixture_conversations): self.assertEqual( output, [self.tokenizer.pad_token_id] * (longest - len(expected.tokens)) + expected.tokens, ) else: self.assertEqual(len(token_outputs), len(self.tokenized_fixture_conversations)) for output, expected in zip(token_outputs, self.tokenized_fixture_conversations): if len(expected.tokens) < max_length: self.assertEqual( output, [self.tokenizer.pad_token_id] * (20 - len(expected.tokens)) + expected.tokens, ) else: self.assertEqual(output, expected.tokens) for padding in [False, "do_not_pad", PaddingStrategy.DO_NOT_PAD]: token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=True, padding=padding, return_dict=False ) self.assertEqual(len(token_outputs), len(self.tokenized_fixture_conversations)) for output, expected in zip(token_outputs, self.tokenized_fixture_conversations): self.assertEqual(output, expected.tokens) def test_batch_apply_chat_template_with_padding_and_truncation( self, ): max_length = 20 for padding in [True, "max_length", PaddingStrategy.LONGEST, PaddingStrategy.MAX_LENGTH]: token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=True, truncation=True, padding=padding, max_length=max_length, return_dict=False, ) self.assertEqual(len(token_outputs), len(self.tokenized_fixture_conversations)) for output, expected in zip(token_outputs, self.tokenized_fixture_conversations): self.assertEqual( output, [self.tokenizer.pad_token_id] * (20 - len(expected.tokens)) + expected.tokens[:20] ) for padding in [False, "do_not_pad", PaddingStrategy.DO_NOT_PAD]: token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=True, truncation=True, padding=padding, max_length=max_length, return_dict=False, ) self.assertEqual(len(token_outputs), len(self.tokenized_fixture_conversations)) for output, expected in zip(token_outputs, self.tokenized_fixture_conversations): self.assertEqual(output, expected.tokens[:20]) def test_batch_apply_chat_template_return_tensors(self): # Test 1: # with tokenize token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=True, return_tensors="pt", padding=True, return_dict=False ) self.assertIsInstance(token_outputs, torch.Tensor) self.assertEqual( token_outputs.shape, (len(self.fixture_conversations), max(len(t.tokens) for t in self.tokenized_fixture_conversations)), ) # Test 2: # without tokenize, should ignore return_tensors token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=False, return_tensors="pt", padding=True, return_dict=False ) self.assertEqual( token_outputs, [ self.ref_tokenizer.decode(tokens=t.tokens, special_token_policy=SpecialTokenPolicy.KEEP) for t in self.tokenized_fixture_conversations ], ) def test_batch_apply_chat_template_return_dict(self): # Test 1: # with tokenize token_outputs = self.tokenizer.apply_chat_template(self.fixture_conversations, tokenize=True, return_dict=True) self.assertIn("input_ids", token_outputs) self.assertIn("attention_mask", token_outputs) self.assertEqual(token_outputs["input_ids"], [t.tokens for t in self.tokenized_fixture_conversations]) self.assertEqual( token_outputs["attention_mask"], [[1] * len(t.tokens) for t in self.tokenized_fixture_conversations] ) # Test 2: # without tokenize, should ignore return_dict token_outputs = self.tokenizer.apply_chat_template( self.fixture_conversations, tokenize=False, return_dict=True ) self.assertNotIsInstance(token_outputs, dict) self.assertEqual( token_outputs, [ self.ref_tokenizer.decode(tokens=t.tokens, special_token_policy=SpecialTokenPolicy.KEEP) for t in self.tokenized_fixture_conversations ], ) def test_call(self): # Test 1: # default case text = "Hello world!" expected_tokens = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(text, bos=True, eos=False) tokens = self.tokenizer(text) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [1] * len(expected_tokens)) # Test 2: # return_attention_mask=False tokens = self.tokenizer(text, return_attention_mask=False) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertNotIn("attention_mask", tokens) # Test 3: # return_tensors="pt" tokens = self.tokenizer(text, return_tensors="pt") self.assertIsInstance(tokens["input_ids"], torch.Tensor) self.assertTrue(torch.equal(tokens["input_ids"], torch.Tensor(expected_tokens).unsqueeze(0))) self.assertIsInstance(tokens["attention_mask"], torch.Tensor) self.assertTrue(torch.equal(tokens["attention_mask"], torch.ones(1, len(expected_tokens)))) # Test 4: # return_special_tokens_mask=True tokens = self.tokenizer(text, return_special_tokens_mask=True) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [1] * len(expected_tokens)) self.assertEqual(tokens["special_tokens_mask"], [1] + [0] * (len(expected_tokens) - 1)) # Test 5: # add_special_tokens=False expected_tokens = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(text, bos=False, eos=False) tokens = self.tokenizer(text, add_special_tokens=False, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [1] * len(expected_tokens)) self.assertEqual(tokens["special_tokens_mask"], [0] * len(expected_tokens)) # Test 6: # add_special_tokens=False and mode finetuning text = "Hello world!" expected_tokens = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(text, bos=True, eos=True) with patch.object(self.tokenizer, "_mode", ValidationMode.finetuning): tokens = self.tokenizer(text, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [1] * len(expected_tokens)) self.assertEqual(tokens["special_tokens_mask"], [1] + [0] * (len(expected_tokens) - 2) + [1]) # Test 7: # empty string tokens = self.tokenizer("", add_special_tokens=False) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], []) self.assertEqual(tokens["attention_mask"], []) with self.assertRaises( ValueError, msg="Kwargs [wrong_kwarg] are not supported by `MistralCommonBackend.__call__`." ): self.tokenizer(text, wrong_kwarg=True) with self.assertRaises( ValueError, msg="`text_pair`, `text_target` and `text_pair_target` are not supported by `MistralCommonBackend`.", ): self.tokenizer(text, text_pair="Hello world!") with self.assertRaises( ValueError, msg="`text_pair`, `text_target` and `text_pair_target` are not supported by `MistralCommonBackend`.", ): self.tokenizer(text, text_target="Hello world!") with self.assertRaises( ValueError, msg="`text_pair`, `text_target` and `text_pair_target` are not supported by `MistralCommonBackend`.", ): self.tokenizer(text, text_pair_target="Hello world!") def test_call_with_truncation(self): # Test 1: # truncation=True or "longest_first" or TruncationStrategy.LONGEST_FIRST text = "Hello world!" * 10 expected_tokens = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(text, bos=True, eos=False) for truncation in [True, "longest_first", TruncationStrategy.LONGEST_FIRST]: tokens = self.tokenizer(text, truncation=True, max_length=10, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens[:10]) self.assertEqual(tokens["attention_mask"], [1] * 10) self.assertEqual(tokens["special_tokens_mask"], [1, 0, 0, 0, 0, 0, 0, 0, 0, 0]) # Test 2: # truncation=False for truncation in [False, "do_not_truncate", TruncationStrategy.DO_NOT_TRUNCATE]: tokens = self.tokenizer(text, truncation=truncation, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [1] * len(expected_tokens)) self.assertEqual(tokens["special_tokens_mask"], [1] + [0] * (len(expected_tokens) - 1)) # Test 3: # truncation=True or "longest_first" or TruncationStrategy.LONGEST_FIRST with return_overflowing_tokens=True and stride for truncation in [True, "longest_first", TruncationStrategy.LONGEST_FIRST]: for stride in [0, 2]: tokens = self.tokenizer( text, truncation=truncation, max_length=10, return_overflowing_tokens=True, return_special_tokens_mask=True, stride=stride, ) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens[:10]) self.assertEqual(tokens["attention_mask"], [1] * 10) self.assertEqual(tokens["special_tokens_mask"], [1, 0, 0, 0, 0, 0, 0, 0, 0, 0]) self.assertEqual(tokens["overflowing_tokens"], expected_tokens[10 - stride :]) self.assertEqual(tokens["num_truncated_tokens"], len(expected_tokens) - 10) # Test 4: # truncation="only_first" or TruncationStrategy.ONLY_FIRST or "only_second" or TruncationStrategy.ONLY_SECOND # should raise an error for truncation in ["only_first", TruncationStrategy.ONLY_FIRST, "only_second", TruncationStrategy.ONLY_SECOND]: with self.assertRaises( ValueError, msg="Truncation strategy `only_first` and `only_second` are not supported by `MistralCommonBackend`.", ): self.tokenizer(text, truncation=truncation) def test_call_with_padding(self): text = "Hello world!" expected_tokens = self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(text, bos=True, eos=False) # Test 1: # padding=False or padding=True or "do_not_pad" or PaddingStrategy.DO_NOT_PAD or padding="longest" or PaddingStrategy.LONGEST for padding in [False, True, "do_not_pad", PaddingStrategy.DO_NOT_PAD, "longest", PaddingStrategy.LONGEST]: tokens = self.tokenizer(text, padding=padding, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [1] * len(expected_tokens)) self.assertEqual(tokens["special_tokens_mask"], [1] + [0] * (len(expected_tokens) - 1)) # Test 2: # padding="max_length" or PaddingStrategy.MAX_LENGTH for padding in ["max_length", PaddingStrategy.MAX_LENGTH]: tokens = self.tokenizer(text, padding=padding, max_length=20, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) num_padding = 20 - len(expected_tokens) self.assertEqual(tokens["input_ids"], num_padding * [self.tokenizer.pad_token_id] + expected_tokens) self.assertEqual(tokens["attention_mask"], num_padding * [0] + [1] * len(expected_tokens)) self.assertEqual(tokens["special_tokens_mask"], num_padding * [1] + [1] + [0] * (len(expected_tokens) - 1)) # Test 3: # pad_to_multiple_of tokens = self.tokenizer( text, padding=True, max_length=20, pad_to_multiple_of=16, return_special_tokens_mask=True ) self.assertIsInstance(tokens, BatchEncoding) num_padding = 16 - len(expected_tokens) self.assertEqual(tokens["input_ids"], num_padding * [self.tokenizer.pad_token_id] + expected_tokens) self.assertEqual(tokens["attention_mask"], num_padding * [0] + [1] * len(expected_tokens)) self.assertEqual(tokens["special_tokens_mask"], num_padding * [1] + [1] + [0] * (len(expected_tokens) - 1)) # Test 4: # padding="max_length" and padding_side="right" tokens = self.tokenizer( text, padding="max_length", max_length=20, padding_side="right", return_special_tokens_mask=True ) self.assertIsInstance(tokens, BatchEncoding) num_padding = 20 - len(expected_tokens) self.assertEqual(tokens["input_ids"], expected_tokens + num_padding * [self.tokenizer.pad_token_id]) self.assertEqual(tokens["attention_mask"], [1] * len(expected_tokens) + num_padding * [0]) self.assertEqual(tokens["special_tokens_mask"], [1] + [0] * (len(expected_tokens) - 1) + num_padding * [1]) def test_batch_call(self): # Test 1: # default case text = ["Hello world!", "Hello world! Longer"] expected_tokens = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(t, bos=True, eos=False) for t in text ] tokens = self.tokenizer(text) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [[1] * len(t) for t in expected_tokens]) # Test 2: # return_attention_mask=False tokens = self.tokenizer(text, return_attention_mask=False) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertNotIn("attention_mask", tokens) # Test 3: # return_tensors="pt" tokens = self.tokenizer(text, return_tensors="pt", padding="longest", return_special_tokens_mask=True) self.assertIsInstance(tokens["input_ids"], torch.Tensor) self.assertEqual(tokens["input_ids"].shape, torch.Size([2, len(expected_tokens[1])])) self.assertTrue( torch.equal( tokens["input_ids"][0], torch.Tensor( (len(expected_tokens[1]) - len(expected_tokens[0])) * [self.ref_tokenizer.instruct_tokenizer.tokenizer.pad_id] + expected_tokens[0] ), ) ) self.assertIsInstance(tokens["attention_mask"], torch.Tensor) self.assertEqual(tokens["attention_mask"].shape, torch.Size([2, len(expected_tokens[1])])) self.assertTrue( torch.equal( tokens["attention_mask"][0], torch.Tensor( [0] * (len(expected_tokens[1]) - len(expected_tokens[0])) + [1] * len(expected_tokens[0]) ), ) ) self.assertTrue(torch.equal(tokens["attention_mask"][1], torch.Tensor([1] * len(expected_tokens[1])))) self.assertIsInstance(tokens["special_tokens_mask"], torch.Tensor) self.assertEqual(tokens["special_tokens_mask"].shape, torch.Size([2, len(expected_tokens[1])])) self.assertTrue( torch.equal( tokens["special_tokens_mask"][0], torch.Tensor( (len(expected_tokens[1]) - len(expected_tokens[0])) * [1] + [1] + [0] * (len(expected_tokens[0]) - 1) ), ) ) self.assertTrue( torch.equal(tokens["special_tokens_mask"][1], torch.Tensor([1] + [0] * (len(expected_tokens[1]) - 1))) ) # Test 4: # add_special_tokens=False expected_tokens = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(t, bos=False, eos=False) for t in text ] tokens = self.tokenizer(text, add_special_tokens=False, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [[1] * len(t) for t in expected_tokens]) self.assertEqual(tokens["special_tokens_mask"], [[0] * len(t) for t in expected_tokens]) # Test 5: # add_special_tokens=True and mode = finetuning expected_tokens = [self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(t, bos=True, eos=True) for t in text] with patch.object(self.tokenizer, "_mode", ValidationMode.finetuning): tokens = self.tokenizer(text, add_special_tokens=True, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual( tokens["special_tokens_mask"], [[1] + [0] * (len(expected_tokens[0]) - 2) + [1], [1] + [0] * (len(expected_tokens[1]) - 2) + [1]], ) # Test 6: # empty string in batch expected_tokens = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(t, bos=False, eos=False) for t in text ] expected_tokens.append([]) tokens = self.tokenizer(text + [""], add_special_tokens=False, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [[1] * len(t) for t in expected_tokens]) self.assertEqual(tokens["special_tokens_mask"], [[0] * len(t) for t in expected_tokens]) # Test 7: # empty batch tokens = self.tokenizer([""], add_special_tokens=False, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], [[]]) self.assertEqual(tokens["attention_mask"], [[]]) self.assertEqual(tokens["special_tokens_mask"], [[]]) def test_batch_call_with_truncation(self): # Test 1: # truncation=True text = ["Hello world!", "Hello world! Longer" * 10] expected_tokens = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(t, bos=True, eos=False) for t in text ] for truncation in [True, "longest_first", TruncationStrategy.LONGEST_FIRST]: tokens = self.tokenizer(text, truncation=True, max_length=10, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], [expected_tokens[0][:10], expected_tokens[1][:10]]) self.assertEqual(tokens["attention_mask"], [[1] * min(len(t), 10) for t in expected_tokens]) self.assertEqual( tokens["special_tokens_mask"], [[1 if id in self.ref_special_ids else 0 for id in ids[:10]] for ids in expected_tokens], ) # Test 2: # truncation=False for truncation in [False, "do_not_truncate", TruncationStrategy.DO_NOT_TRUNCATE]: tokens = self.tokenizer(text, truncation=truncation, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [[1] * len(t) for t in expected_tokens]) self.assertEqual( tokens["special_tokens_mask"], [[1] + [0] * (len(t) - 1) for t in expected_tokens], ) # Test 3: # truncation=True or "longest_first" or TruncationStrategy.LONGEST_FIRST with return_overflowing_tokens=True and stride for truncation in [True, "longest_first", TruncationStrategy.LONGEST_FIRST]: for stride in [0, 2]: tokens = self.tokenizer( text, truncation=truncation, max_length=10, return_overflowing_tokens=True, return_special_tokens_mask=True, stride=stride, ) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], [expected_tokens[0][:10], expected_tokens[1][:10]]) self.assertEqual(tokens["attention_mask"], [[1] * min(len(t), 10) for t in expected_tokens]) self.assertEqual( tokens["overflowing_tokens"], [[0], expected_tokens[1][10 - stride :]], ) self.assertEqual(tokens["num_truncated_tokens"], [[0], len(expected_tokens[1]) - 10]) self.assertEqual( tokens["special_tokens_mask"], [[1 if id in self.ref_special_ids else 0 for id in ids[:10]] for ids in expected_tokens], ) def test_batch_call_with_padding(self): # Test 1: # padding=False or padding=True or "do_not_pad" or PaddingStrategy.DO_NOT_PAD or padding="longest" or PaddingStrategy.LONGEST text = ["Hello world!", "Hello world! Longer"] expected_tokens = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(t, bos=True, eos=False) for t in text ] for padding in [False, "do_not_pad", PaddingStrategy.DO_NOT_PAD]: tokens = self.tokenizer(text, padding=padding, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) self.assertEqual(tokens["input_ids"], expected_tokens) self.assertEqual(tokens["attention_mask"], [[1] * len(t) for t in expected_tokens]) self.assertEqual( tokens["special_tokens_mask"], [[1] + [0] * (len(t) - 1) for t in expected_tokens], ) # Test 2: # padding="max_length" or PaddingStrategy.MAX_LENGTH for padding in ["max_length", PaddingStrategy.MAX_LENGTH]: tokens = self.tokenizer(text, padding=padding, max_length=20, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) num_padding = [20 - len(t) for t in expected_tokens] self.assertEqual( tokens["input_ids"], [ num_padding[0] * [self.tokenizer.pad_token_id] + expected_tokens[0], num_padding[1] * [self.tokenizer.pad_token_id] + expected_tokens[1], ], ) self.assertEqual( tokens["attention_mask"], [ num_padding[0] * [0] + [1] * len(expected_tokens[0]), num_padding[1] * [0] + [1] * len(expected_tokens[1]), ], ) self.assertEqual( tokens["special_tokens_mask"], [ num_padding[0] * [1] + [1] + [0] * (len(expected_tokens[0]) - 1), num_padding[1] * [1] + [1] + [0] * (len(expected_tokens[1]) - 1), ], ) # Test 3: # padding=True or "longest" or PaddingStrategy.LONGEST for padding in [True, "longest", PaddingStrategy.LONGEST]: tokens = self.tokenizer(text, padding=padding, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) num_padding = [len(expected_tokens[1]) - len(t) for t in expected_tokens] self.assertEqual( tokens["input_ids"], [ num_padding[0] * [self.tokenizer.pad_token_id] + expected_tokens[0], num_padding[1] * [self.tokenizer.pad_token_id] + expected_tokens[1], ], ) self.assertEqual( tokens["attention_mask"], [ num_padding[0] * [0] + [1] * len(expected_tokens[0]), num_padding[1] * [0] + [1] * len(expected_tokens[1]), ], ) self.assertEqual( tokens["special_tokens_mask"], [ num_padding[0] * [1] + [1] + [0] * (len(expected_tokens[0]) - 1), num_padding[1] * [1] + [1] + [0] * (len(expected_tokens[1]) - 1), ], ) # Test 4: # pad_to_multiple_of tokens = self.tokenizer( text, padding=True, max_length=32, pad_to_multiple_of=16, return_special_tokens_mask=True ) self.assertIsInstance(tokens, BatchEncoding) num_padding = [16 - len(t) for t in expected_tokens] self.assertEqual( tokens["input_ids"], [ num_padding[0] * [self.tokenizer.pad_token_id] + expected_tokens[0], num_padding[1] * [self.tokenizer.pad_token_id] + expected_tokens[1], ], ) self.assertEqual( tokens["attention_mask"], [ num_padding[0] * [0] + [1] * len(expected_tokens[0]), num_padding[1] * [0] + [1] * len(expected_tokens[1]), ], ) self.assertEqual( tokens["special_tokens_mask"], [ num_padding[0] * [1] + [1] + [0] * (len(expected_tokens[0]) - 1), num_padding[1] * [1] + [1] + [0] * (len(expected_tokens[1]) - 1), ], ) # Test 5: # padding="max_length" or PaddingStrategy.MAX_LENGTH and padding_side="right" for padding in ["max_length", PaddingStrategy.MAX_LENGTH]: tokens = self.tokenizer( text, padding=padding, max_length=20, padding_side="right", return_special_tokens_mask=True ) self.assertIsInstance(tokens, BatchEncoding) num_padding = [20 - len(t) for t in expected_tokens] self.assertEqual( tokens["input_ids"], [ expected_tokens[0] + num_padding[0] * [self.tokenizer.pad_token_id], expected_tokens[1] + num_padding[1] * [self.tokenizer.pad_token_id], ], ) self.assertEqual( tokens["attention_mask"], [ [1] * len(expected_tokens[0]) + num_padding[0] * [0], [1] * len(expected_tokens[1]) + num_padding[1] * [0], ], ) self.assertEqual( tokens["special_tokens_mask"], [ [1] + [0] * (len(expected_tokens[0]) - 1) + num_padding[0] * [1], [1] + [0] * (len(expected_tokens[1]) - 1) + num_padding[1] * [1], ], ) def test_batch_call_with_padding_and_truncation(self): # Test 1: # padding=True or "longest" or PaddingStrategy.LONGEST or "max_length" or PaddingStragy.MAX_LENGTH # and truncation=True or "longest_first" or TruncationStrategy.LONGEST_FIRST # and max_length text = ["Hello world!", "Hello world! Longer" * 10] expected_tokens = [ self.ref_tokenizer.instruct_tokenizer.tokenizer.encode(t, bos=True, eos=False) for t in text ] for padding in [True, "longest", PaddingStrategy.LONGEST, "max_length", PaddingStrategy.MAX_LENGTH]: for truncation in [True, "longest_first", TruncationStrategy.LONGEST_FIRST]: tokens = self.tokenizer( text, padding=padding, truncation=truncation, max_length=10, return_special_tokens_mask=True ) num_padding = [max(0, 10 - len(t)) for t in expected_tokens] self.assertIsInstance(tokens, BatchEncoding) self.assertEqual( tokens["input_ids"], [num_padding[i] * [self.tokenizer.pad_token_id] + t[:10] for i, t in enumerate(expected_tokens)], ) self.assertEqual( tokens["attention_mask"], [num_padding[i] * [0] + [1] * min(len(t), 10) for i, t in enumerate(expected_tokens)], ) self.assertEqual( tokens["special_tokens_mask"], [ num_padding[i] * [1] + [1 if id in self.ref_special_ids else 0 for id in ids[:10]] for i, ids in enumerate(expected_tokens) ], ) # Test 2: # padding=True or "longest" or PaddingStrategy.LONGEST and truncation=True or "longest_first" or TruncationStrategy.LONGEST_FIRST # and no max_length for padding in ["longest", PaddingStrategy.LONGEST]: for truncation in [True, "longest_first", TruncationStrategy.LONGEST_FIRST]: tokens = self.tokenizer(text, padding=padding, truncation=truncation, return_special_tokens_mask=True) self.assertIsInstance(tokens, BatchEncoding) num_padding = [max(len(t) for t in expected_tokens) - len(t) for t in expected_tokens] self.assertEqual( tokens["input_ids"], [num_padding[i] * [self.tokenizer.pad_token_id] + t for i, t in enumerate(expected_tokens)], ) self.assertEqual( tokens["attention_mask"], [num_padding[i] * [0] + [1] * len(t) for i, t in enumerate(expected_tokens)], ) self.assertEqual( tokens["special_tokens_mask"], [ num_padding[i] * [1] + [1 if id in self.ref_special_ids else 0 for id in ids] for i, ids in enumerate(expected_tokens) ], ) def test_get_vocab(self): vocab = self.tokenizer.get_vocab() # loss of some tokens due to conversion self.assertNotEqual(len(vocab), len(self.tokenizer)) for token, id_token in vocab.items(): # Issue during conversion if id_token == 0 and token != "<unk>": continue self.assertEqual(self.tokenizer.convert_tokens_to_ids(token), id_token) self.assertEqual( self.ref_tokenizer.decode([id_token], special_token_policy=SpecialTokenPolicy.KEEP), token ) def test_get_validation_mode(self): for mode, expected in [ ("test", ValidationMode.test), (ValidationMode.test, ValidationMode.test), ("finetuning", ValidationMode.finetuning), (ValidationMode.finetuning, ValidationMode.finetuning), ("serving", ValidationMode.serving), (ValidationMode.serving, ValidationMode.serving), ]: self.assertEqual(MistralCommonBackend._get_validation_mode(mode), expected) for invalid_mode in ["invalid", 1]: with self.assertRaises(ValueError): MistralCommonBackend._get_validation_mode(invalid_mode) def test_all_special_ids(self): with patch.object(self.tokenizer, "_all_special_ids", {1, 0}): self.assertEqual(self.tokenizer.all_special_ids, [0, 1]) spm_tokenizer = self._get_spm_tokenizer() with patch.object(spm_tokenizer, "_all_special_ids", {1, 0}): self.assertEqual(spm_tokenizer.all_special_ids, [0, 1]) def test_all_special_tokens(self): with patch.object(self.tokenizer, "_all_special_tokens", ["<unk>", "<s>"]): self.assertEqual(self.tokenizer.all_special_tokens, ["<unk>", "<s>"]) spm_tokenizer = self._get_spm_tokenizer() with patch.object(spm_tokenizer, "_all_special_tokens", ["<unk>", "<s>", "spm"]): self.assertEqual(spm_tokenizer.all_special_tokens, ["<unk>", "<s>", "spm"]) def test_mode(self): # Test 1: # mode property should return ValidationMode self.assertIsInstance(self.tokenizer.mode, ValidationMode) self.assertEqual(self.tokenizer.mode, ValidationMode.test) # Test 2: # mode should be settable via instantiation spm_finetuning_tokenizer = self._get_spm_tokenizer(mode="finetuning") self.assertEqual(spm_finetuning_tokenizer.mode, ValidationMode.finetuning) def test_build_inputs_with_special_tokens(self): # Test 1: # test mode with bos only token_ids = [100, 200, 300] expected = [self.tokenizer.bos_token_id] + token_ids result = self.tokenizer.build_inputs_with_special_tokens(token_ids) self.assertEqual(result, expected) # Test 2: # finetuning mode with bos and eos with patch.object(self.tokenizer, "_mode", ValidationMode.finetuning): expected = [self.tokenizer.bos_token_id] + token_ids + [self.tokenizer.eos_token_id] result = self.tokenizer.build_inputs_with_special_tokens(token_ids) self.assertEqual(result, expected) # Test 3: # token_ids_1 should raise ValueError with self.assertRaises(ValueError): self.tokenizer.build_inputs_with_special_tokens(token_ids, [400, 500]) def test_create_token_type_ids_from_sequences(self): # Test 1: # create token type ids for single sequence token_ids = [100, 200, 300] expected_length = len(token_ids) + 1 # +1 for bos token result = self.tokenizer.create_token_type_ids_from_sequences(token_ids) self.assertEqual(result, [0] * expected_length) # Test 2: # token_ids_1 should raise ValueError with self.assertRaises(ValueError): self.tokenizer.create_token_type_ids_from_sequences(token_ids, [400, 500]) def test_num_special_tokens_to_add(self): # Test 1: # test mode should add 1 token (bos) result = self.tokenizer.num_special_tokens_to_add() self.assertEqual(result, 1) # Test 2: # finetuning mode should add 2 tokens (bos and eos) with patch.object(self.tokenizer, "_mode", ValidationMode.finetuning): result = self.tokenizer.num_special_tokens_to_add() self.assertEqual(result, 2) # Test 3: # pair=True should raise ValueError with self.assertRaises(ValueError): self.tokenizer.num_special_tokens_to_add(pair=True) def test_prepare_for_model(self): # Test 1: # basic prepare_for_model with add_special_tokens=True token_ids = [100, 200, 300] result = self.tokenizer.prepare_for_model(token_ids, add_special_tokens=True) expected_ids = [self.tokenizer.bos_token_id] + token_ids self.assertEqual(result["input_ids"], expected_ids) self.assertEqual(result["attention_mask"], [1] * len(expected_ids)) # Test 2: # prepare_for_model with add_special_tokens=False result = self.tokenizer.prepare_for_model(token_ids, add_special_tokens=False) self.assertEqual(result["input_ids"], token_ids) self.assertEqual(result["attention_mask"], [1] * len(token_ids)) # Test 3: # prepare_for_model with padding result = self.tokenizer.prepare_for_model( token_ids, add_special_tokens=False, padding="max_length", max_length=10 ) expected_ids = [self.tokenizer.pad_token_id] * (10 - len(token_ids)) + token_ids expected_attention_mask = [0] * (10 - len(token_ids)) + [1] * len(token_ids) self.assertEqual(result["input_ids"], expected_ids) self.assertEqual(result["attention_mask"], expected_attention_mask) # Test 4: # prepare_for_model with truncation long_token_ids = [100, 200, 300, 400, 500, 600] result = self.tokenizer.prepare_for_model( long_token_ids, add_special_tokens=False, truncation=True, max_length=4 ) expected_ids = long_token_ids[:4] self.assertEqual(result["input_ids"], expected_ids) self.assertEqual(result["attention_mask"], [1] * len(expected_ids)) # Test 5: # prepare_for_model with return_tensors result = self.tokenizer.prepare_for_model(token_ids, add_special_tokens=False, return_tensors="pt") self.assertIsInstance(result["input_ids"], torch.Tensor) self.assertEqual(result["input_ids"].tolist(), token_ids) # Test 6: # pair_ids should raise ValueError with self.assertRaises(ValueError): self.tokenizer.prepare_for_model(token_ids, pair_ids=[400, 500]) # Test 7: # unsupported kwargs should raise ValueError with self.assertRaises(ValueError): self.tokenizer.prepare_for_model(token_ids, add_special_tokens=False, unsupported_arg="") class TestMistralCommonImport(unittest.TestCase): def test_import_does_not_raise_regardless_of_mistral_common_availability(self) -> None: # Regression test: importing `transformers.tokenization_mistral_common` must not raise, whether or not # `mistral_common` is installed. # The module body is executed under a throwaway module name via `importlib.util` so that the real # `sys.modules["transformers.tokenization_mistral_common"]` entry is never swapped out. module_origin = importlib.util.find_spec("transformers.tokenization_mistral_common").origin for available in (False, True): with self.subTest(mistral_common_available=available): with patch("transformers.utils.import_utils.is_mistral_common_available", return_value=available): throwaway_name = f"_regression_tokenization_mistral_common_available_{available}" spec = importlib.util.spec_from_file_location(throwaway_name, module_origin) module = importlib.util.module_from_spec(spec) sys.modules[throwaway_name] = module try: # Executing the module body must not raise (this is what the regression guards against). spec.loader.exec_module(module) # `_MAP_SPECIAL_TOKENS` is only bound when `mistral_common` is available. self.assertEqual(hasattr(module, "_MAP_SPECIAL_TOKENS"), available) finally: sys.modules.pop(throwaway_name, None)