chendl/compositional_test
1
1# Copyright 2020 The HuggingFace Team. All rights reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14 15import logging16import os17import sys18import tempfile19import unittest20from pathlib import Path21 22import datasets23import numpy as np24from huggingface_hub import HfFolder, Repository, create_repo, delete_repo25from requests.exceptions import HTTPError26 27from transformers import (28 AutoModelForSequenceClassification,29 AutoTokenizer,30 DistilBertForSequenceClassification,31 TextClassificationPipeline,32 TFAutoModelForSequenceClassification,33 pipeline,34)35from transformers.pipelines import PIPELINE_REGISTRY, get_task36from transformers.pipelines.base import Pipeline, _pad37from transformers.testing_utils import (38 TOKEN,39 USER,40 CaptureLogger,41 RequestCounter,42 is_pipeline_test,43 is_staging_test,44 nested_simplify,45 require_tensorflow_probability,46 require_tf,47 require_torch,48 require_torch_or_tf,49 slow,50)51from transformers.utils import direct_transformers_import, is_tf_available, is_torch_available52from transformers.utils import logging as transformers_logging53 54 55sys.path.append(str(Path(__file__).parent.parent.parent / "utils"))56 57from test_module.custom_pipeline import PairClassificationPipeline # noqa E40258 59 60logger = logging.getLogger(__name__)61 62 63PATH_TO_TRANSFORMERS = os.path.join(Path(__file__).parent.parent.parent, "src/transformers")64 65 66# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.67transformers_module = direct_transformers_import(PATH_TO_TRANSFORMERS)68 69 70class ANY:71 def __init__(self, *_types):72 self._types = _types73 74 def __eq__(self, other):75 return isinstance(other, self._types)76 77 def __repr__(self):78 return f"ANY({', '.join(_type.__name__ for _type in self._types)})"79 80 81@is_pipeline_test82class CommonPipelineTest(unittest.TestCase):83 @require_torch84 def test_pipeline_iteration(self):85 from torch.utils.data import Dataset86 87 class MyDataset(Dataset):88 data = [89 "This is a test",90 "This restaurant is great",91 "This restaurant is awful",92 ]93 94 def __len__(self):95 return 396 97 def __getitem__(self, i):98 return self.data[i]99 100 text_classifier = pipeline(101 task="text-classification", model="hf-internal-testing/tiny-random-distilbert", framework="pt"102 )103 dataset = MyDataset()104 for output in text_classifier(dataset):105 self.assertEqual(output, {"label": ANY(str), "score": ANY(float)})106 107 @require_torch108 def test_check_task_auto_inference(self):109 pipe = pipeline(model="hf-internal-testing/tiny-random-distilbert")110 111 self.assertIsInstance(pipe, TextClassificationPipeline)112 113 @require_torch114 def test_pipeline_batch_size_global(self):115 pipe = pipeline(model="hf-internal-testing/tiny-random-distilbert")116 self.assertEqual(pipe._batch_size, None)117 self.assertEqual(pipe._num_workers, None)118 119 pipe = pipeline(model="hf-internal-testing/tiny-random-distilbert", batch_size=2, num_workers=1)120 self.assertEqual(pipe._batch_size, 2)121 self.assertEqual(pipe._num_workers, 1)122 123 @require_torch124 def test_pipeline_pathlike(self):125 pipe = pipeline(model="hf-internal-testing/tiny-random-distilbert")126 with tempfile.TemporaryDirectory() as d:127 pipe.save_pretrained(d)128 path = Path(d)129 newpipe = pipeline(task="text-classification", model=path)130 self.assertIsInstance(newpipe, TextClassificationPipeline)131 132 @require_torch133 def test_pipeline_override(self):134 class MyPipeline(TextClassificationPipeline):135 pass136 137 text_classifier = pipeline(model="hf-internal-testing/tiny-random-distilbert", pipeline_class=MyPipeline)138 139 self.assertIsInstance(text_classifier, MyPipeline)140 141 def test_check_task(self):142 task = get_task("gpt2")143 self.assertEqual(task, "text-generation")144 145 with self.assertRaises(RuntimeError):146 # Wrong framework147 get_task("espnet/siddhana_slurp_entity_asr_train_asr_conformer_raw_en_word_valid.acc.ave_10best")148 149 @require_torch150 def test_iterator_data(self):151 def data(n: int):152 for _ in range(n):153 yield "This is a test"154 155 pipe = pipeline(model="hf-internal-testing/tiny-random-distilbert")156 157 results = []158 for out in pipe(data(10)):159 self.assertEqual(nested_simplify(out), {"label": "LABEL_0", "score": 0.504})160 results.append(out)161 self.assertEqual(len(results), 10)162 163 # When using multiple workers on streamable data it should still work164 # This will force using `num_workers=1` with a warning for now.165 results = []166 for out in pipe(data(10), num_workers=2):167 self.assertEqual(nested_simplify(out), {"label": "LABEL_0", "score": 0.504})168 results.append(out)169 self.assertEqual(len(results), 10)170 171 @require_tf172 def test_iterator_data_tf(self):173 def data(n: int):174 for _ in range(n):175 yield "This is a test"176 177 pipe = pipeline(model="hf-internal-testing/tiny-random-distilbert", framework="tf")178 out = pipe("This is a test")179 results = []180 for out in pipe(data(10)):181 self.assertEqual(nested_simplify(out), {"label": "LABEL_0", "score": 0.504})182 results.append(out)183 self.assertEqual(len(results), 10)184 185 @require_torch186 def test_unbatch_attentions_hidden_states(self):187 model = DistilBertForSequenceClassification.from_pretrained(188 "hf-internal-testing/tiny-random-distilbert", output_hidden_states=True, output_attentions=True189 )190 tokenizer = AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-distilbert")191 text_classifier = TextClassificationPipeline(model=model, tokenizer=tokenizer)192 193 # Used to throw an error because `hidden_states` are a tuple of tensors194 # instead of the expected tensor.195 outputs = text_classifier(["This is great !"] * 20, batch_size=32)196 self.assertEqual(len(outputs), 20)197 198 199@is_pipeline_test200class PipelineScikitCompatTest(unittest.TestCase):201 @require_torch202 def test_pipeline_predict_pt(self):203 data = ["This is a test"]204 205 text_classifier = pipeline(206 task="text-classification", model="hf-internal-testing/tiny-random-distilbert", framework="pt"207 )208 209 expected_output = [{"label": ANY(str), "score": ANY(float)}]210 actual_output = text_classifier.predict(data)211 self.assertEqual(expected_output, actual_output)212 213 @require_tf214 def test_pipeline_predict_tf(self):215 data = ["This is a test"]216 217 text_classifier = pipeline(218 task="text-classification", model="hf-internal-testing/tiny-random-distilbert", framework="tf"219 )220 221 expected_output = [{"label": ANY(str), "score": ANY(float)}]222 actual_output = text_classifier.predict(data)223 self.assertEqual(expected_output, actual_output)224 225 @require_torch226 def test_pipeline_transform_pt(self):227 data = ["This is a test"]228 229 text_classifier = pipeline(230 task="text-classification", model="hf-internal-testing/tiny-random-distilbert", framework="pt"231 )232 233 expected_output = [{"label": ANY(str), "score": ANY(float)}]234 actual_output = text_classifier.transform(data)235 self.assertEqual(expected_output, actual_output)236 237 @require_tf238 def test_pipeline_transform_tf(self):239 data = ["This is a test"]240 241 text_classifier = pipeline(242 task="text-classification", model="hf-internal-testing/tiny-random-distilbert", framework="tf"243 )244 245 expected_output = [{"label": ANY(str), "score": ANY(float)}]246 actual_output = text_classifier.transform(data)247 self.assertEqual(expected_output, actual_output)248 249 250@is_pipeline_test251class PipelinePadTest(unittest.TestCase):252 @require_torch253 def test_pipeline_padding(self):254 import torch255 256 items = [257 {258 "label": "label1",259 "input_ids": torch.LongTensor([[1, 23, 24, 2]]),260 "attention_mask": torch.LongTensor([[0, 1, 1, 0]]),261 },262 {263 "label": "label2",264 "input_ids": torch.LongTensor([[1, 23, 24, 43, 44, 2]]),265 "attention_mask": torch.LongTensor([[0, 1, 1, 1, 1, 0]]),266 },267 ]268 269 self.assertEqual(_pad(items, "label", 0, "right"), ["label1", "label2"])270 self.assertTrue(271 torch.allclose(272 _pad(items, "input_ids", 10, "right"),273 torch.LongTensor([[1, 23, 24, 2, 10, 10], [1, 23, 24, 43, 44, 2]]),274 )275 )276 self.assertTrue(277 torch.allclose(278 _pad(items, "input_ids", 10, "left"),279 torch.LongTensor([[10, 10, 1, 23, 24, 2], [1, 23, 24, 43, 44, 2]]),280 )281 )282 self.assertTrue(283 torch.allclose(284 _pad(items, "attention_mask", 0, "right"), torch.LongTensor([[0, 1, 1, 0, 0, 0], [0, 1, 1, 1, 1, 0]])285 )286 )287 288 @require_torch289 def test_pipeline_image_padding(self):290 import torch291 292 items = [293 {294 "label": "label1",295 "pixel_values": torch.zeros((1, 3, 10, 10)),296 },297 {298 "label": "label2",299 "pixel_values": torch.zeros((1, 3, 10, 10)),300 },301 ]302 303 self.assertEqual(_pad(items, "label", 0, "right"), ["label1", "label2"])304 self.assertTrue(305 torch.allclose(306 _pad(items, "pixel_values", 10, "right"),307 torch.zeros((2, 3, 10, 10)),308 )309 )310 311 @require_torch312 def test_pipeline_offset_mapping(self):313 import torch314 315 items = [316 {317 "offset_mappings": torch.zeros([1, 11, 2], dtype=torch.long),318 },319 {320 "offset_mappings": torch.zeros([1, 4, 2], dtype=torch.long),321 },322 ]323 324 self.assertTrue(325 torch.allclose(326 _pad(items, "offset_mappings", 0, "right"),327 torch.zeros((2, 11, 2), dtype=torch.long),328 ),329 )330 331 332@is_pipeline_test333class PipelineUtilsTest(unittest.TestCase):334 @require_torch335 def test_pipeline_dataset(self):336 from transformers.pipelines.pt_utils import PipelineDataset337 338 dummy_dataset = [0, 1, 2, 3]339 340 def add(number, extra=0):341 return number + extra342 343 dataset = PipelineDataset(dummy_dataset, add, {"extra": 2})344 self.assertEqual(len(dataset), 4)345 outputs = [dataset[i] for i in range(4)]346 self.assertEqual(outputs, [2, 3, 4, 5])347 348 @require_torch349 def test_pipeline_iterator(self):350 from transformers.pipelines.pt_utils import PipelineIterator351 352 dummy_dataset = [0, 1, 2, 3]353 354 def add(number, extra=0):355 return number + extra356 357 dataset = PipelineIterator(dummy_dataset, add, {"extra": 2})358 self.assertEqual(len(dataset), 4)359 360 outputs = list(dataset)361 self.assertEqual(outputs, [2, 3, 4, 5])362 363 @require_torch364 def test_pipeline_iterator_no_len(self):365 from transformers.pipelines.pt_utils import PipelineIterator366 367 def dummy_dataset():368 for i in range(4):369 yield i370 371 def add(number, extra=0):372 return number + extra373 374 dataset = PipelineIterator(dummy_dataset(), add, {"extra": 2})375 with self.assertRaises(TypeError):376 len(dataset)377 378 outputs = list(dataset)379 self.assertEqual(outputs, [2, 3, 4, 5])380 381 @require_torch382 def test_pipeline_batch_unbatch_iterator(self):383 from transformers.pipelines.pt_utils import PipelineIterator384 385 dummy_dataset = [{"id": [0, 1, 2]}, {"id": [3]}]386 387 def add(number, extra=0):388 return {"id": [i + extra for i in number["id"]]}389 390 dataset = PipelineIterator(dummy_dataset, add, {"extra": 2}, loader_batch_size=3)391 392 outputs = list(dataset)393 self.assertEqual(outputs, [{"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}])394 395 @require_torch396 def test_pipeline_batch_unbatch_iterator_tensors(self):397 import torch398 399 from transformers.pipelines.pt_utils import PipelineIterator400 401 dummy_dataset = [{"id": torch.LongTensor([[10, 20], [0, 1], [0, 2]])}, {"id": torch.LongTensor([[3]])}]402 403 def add(number, extra=0):404 return {"id": number["id"] + extra}405 406 dataset = PipelineIterator(dummy_dataset, add, {"extra": 2}, loader_batch_size=3)407 408 outputs = list(dataset)409 self.assertEqual(410 nested_simplify(outputs), [{"id": [[12, 22]]}, {"id": [[2, 3]]}, {"id": [[2, 4]]}, {"id": [[5]]}]411 )412 413 @require_torch414 def test_pipeline_chunk_iterator(self):415 from transformers.pipelines.pt_utils import PipelineChunkIterator416 417 def preprocess_chunk(n: int):418 for i in range(n):419 yield i420 421 dataset = [2, 3]422 423 dataset = PipelineChunkIterator(dataset, preprocess_chunk, {}, loader_batch_size=3)424 425 outputs = list(dataset)426 427 self.assertEqual(outputs, [0, 1, 0, 1, 2])428 429 @require_torch430 def test_pipeline_pack_iterator(self):431 from transformers.pipelines.pt_utils import PipelinePackIterator432 433 def pack(item):434 return {"id": item["id"] + 1, "is_last": item["is_last"]}435 436 dataset = [437 {"id": 0, "is_last": False},438 {"id": 1, "is_last": True},439 {"id": 0, "is_last": False},440 {"id": 1, "is_last": False},441 {"id": 2, "is_last": True},442 ]443 444 dataset = PipelinePackIterator(dataset, pack, {})445 446 outputs = list(dataset)447 self.assertEqual(448 outputs,449 [450 [451 {"id": 1},452 {"id": 2},453 ],454 [455 {"id": 1},456 {"id": 2},457 {"id": 3},458 ],459 ],460 )461 462 @require_torch463 def test_pipeline_pack_unbatch_iterator(self):464 from transformers.pipelines.pt_utils import PipelinePackIterator465 466 dummy_dataset = [{"id": [0, 1, 2], "is_last": [False, True, False]}, {"id": [3], "is_last": [True]}]467 468 def add(number, extra=0):469 return {"id": [i + extra for i in number["id"]], "is_last": number["is_last"]}470 471 dataset = PipelinePackIterator(dummy_dataset, add, {"extra": 2}, loader_batch_size=3)472 473 outputs = list(dataset)474 self.assertEqual(outputs, [[{"id": 2}, {"id": 3}], [{"id": 4}, {"id": 5}]])475 476 # is_false Across batch477 dummy_dataset = [{"id": [0, 1, 2], "is_last": [False, False, False]}, {"id": [3], "is_last": [True]}]478 479 def add(number, extra=0):480 return {"id": [i + extra for i in number["id"]], "is_last": number["is_last"]}481 482 dataset = PipelinePackIterator(dummy_dataset, add, {"extra": 2}, loader_batch_size=3)483 484 outputs = list(dataset)485 self.assertEqual(outputs, [[{"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}]])486 487 def test_pipeline_negative_device(self):488 # To avoid regressing, pipeline used to accept device=-1489 classifier = pipeline("text-generation", "hf-internal-testing/tiny-random-bert", device=-1)490 491 expected_output = [{"generated_text": ANY(str)}]492 actual_output = classifier("Test input.")493 self.assertEqual(expected_output, actual_output)494 495 @slow496 @require_torch497 def test_load_default_pipelines_pt(self):498 import torch499 500 from transformers.pipelines import SUPPORTED_TASKS501 502 set_seed_fn = lambda: torch.manual_seed(0) # noqa: E731503 for task in SUPPORTED_TASKS.keys():504 if task == "table-question-answering":505 # test table in seperate test due to more dependencies506 continue507 508 self.check_default_pipeline(task, "pt", set_seed_fn, self.check_models_equal_pt)509 510 @slow511 @require_tf512 def test_load_default_pipelines_tf(self):513 import tensorflow as tf514 515 from transformers.pipelines import SUPPORTED_TASKS516 517 set_seed_fn = lambda: tf.random.set_seed(0) # noqa: E731518 for task in SUPPORTED_TASKS.keys():519 if task == "table-question-answering":520 # test table in seperate test due to more dependencies521 continue522 523 self.check_default_pipeline(task, "tf", set_seed_fn, self.check_models_equal_tf)524 525 @slow526 @require_torch527 def test_load_default_pipelines_pt_table_qa(self):528 import torch529 530 set_seed_fn = lambda: torch.manual_seed(0) # noqa: E731531 self.check_default_pipeline("table-question-answering", "pt", set_seed_fn, self.check_models_equal_pt)532 533 @slow534 @require_tf535 @require_tensorflow_probability536 def test_load_default_pipelines_tf_table_qa(self):537 import tensorflow as tf538 539 set_seed_fn = lambda: tf.random.set_seed(0) # noqa: E731540 self.check_default_pipeline("table-question-answering", "tf", set_seed_fn, self.check_models_equal_tf)541 542 def check_default_pipeline(self, task, framework, set_seed_fn, check_models_equal_fn):543 from transformers.pipelines import SUPPORTED_TASKS, pipeline544 545 task_dict = SUPPORTED_TASKS[task]546 # test to compare pipeline to manually loading the respective model547 model = None548 relevant_auto_classes = task_dict[framework]549 550 if len(relevant_auto_classes) == 0:551 # task has no default552 logger.debug(f"{task} in {framework} has no default")553 return554 555 # by default use first class556 auto_model_cls = relevant_auto_classes[0]557 558 # retrieve correct model ids559 if task == "translation":560 # special case for translation pipeline which has multiple languages561 model_ids = []562 revisions = []563 tasks = []564 for translation_pair in task_dict["default"].keys():565 model_id, revision = task_dict["default"][translation_pair]["model"][framework]566 567 model_ids.append(model_id)568 revisions.append(revision)569 tasks.append(task + f"_{'_to_'.join(translation_pair)}")570 else:571 # normal case - non-translation pipeline572 model_id, revision = task_dict["default"]["model"][framework]573 574 model_ids = [model_id]575 revisions = [revision]576 tasks = [task]577 578 # check for equality579 for model_id, revision, task in zip(model_ids, revisions, tasks):580 # load default model581 try:582 set_seed_fn()583 model = auto_model_cls.from_pretrained(model_id, revision=revision)584 except ValueError:585 # first auto class is possible not compatible with model, go to next model class586 auto_model_cls = relevant_auto_classes[1]587 set_seed_fn()588 model = auto_model_cls.from_pretrained(model_id, revision=revision)589 590 # load default pipeline591 set_seed_fn()592 default_pipeline = pipeline(task, framework=framework)593 594 # compare pipeline model with default model595 models_are_equal = check_models_equal_fn(default_pipeline.model, model)596 self.assertTrue(models_are_equal, f"{task} model doesn't match pipeline.")597 598 logger.debug(f"{task} in {framework} succeeded with {model_id}.")599 600 def check_models_equal_pt(self, model1, model2):601 models_are_equal = True602 for model1_p, model2_p in zip(model1.parameters(), model2.parameters()):603 if model1_p.data.ne(model2_p.data).sum() > 0:604 models_are_equal = False605 606 return models_are_equal607 608 def check_models_equal_tf(self, model1, model2):609 models_are_equal = True610 for model1_p, model2_p in zip(model1.weights, model2.weights):611 if np.abs(model1_p.numpy() - model2_p.numpy()).sum() > 1e-5:612 models_are_equal = False613 614 return models_are_equal615 616 617class CustomPipeline(Pipeline):618 def _sanitize_parameters(self, **kwargs):619 preprocess_kwargs = {}620 if "maybe_arg" in kwargs:621 preprocess_kwargs["maybe_arg"] = kwargs["maybe_arg"]622 return preprocess_kwargs, {}, {}623 624 def preprocess(self, text, maybe_arg=2):625 input_ids = self.tokenizer(text, return_tensors="pt")626 return input_ids627 628 def _forward(self, model_inputs):629 outputs = self.model(**model_inputs)630 return outputs631 632 def postprocess(self, model_outputs):633 return model_outputs["logits"].softmax(-1).numpy()634 635 636@is_pipeline_test637class CustomPipelineTest(unittest.TestCase):638 def test_warning_logs(self):639 transformers_logging.set_verbosity_debug()640 logger_ = transformers_logging.get_logger("transformers.pipelines.base")641 642 alias = "text-classification"643 # Get the original task, so we can restore it at the end.644 # (otherwise the subsequential tests in `TextClassificationPipelineTests` will fail)645 _, original_task, _ = PIPELINE_REGISTRY.check_task(alias)646 647 try:648 with CaptureLogger(logger_) as cm:649 PIPELINE_REGISTRY.register_pipeline(alias, PairClassificationPipeline)650 self.assertIn(f"{alias} is already registered", cm.out)651 finally:652 # restore653 PIPELINE_REGISTRY.supported_tasks[alias] = original_task654 655 def test_register_pipeline(self):656 PIPELINE_REGISTRY.register_pipeline(657 "custom-text-classification",658 pipeline_class=PairClassificationPipeline,659 pt_model=AutoModelForSequenceClassification if is_torch_available() else None,660 tf_model=TFAutoModelForSequenceClassification if is_tf_available() else None,661 default={"pt": "hf-internal-testing/tiny-random-distilbert"},662 type="text",663 )664 assert "custom-text-classification" in PIPELINE_REGISTRY.get_supported_tasks()665 666 _, task_def, _ = PIPELINE_REGISTRY.check_task("custom-text-classification")667 self.assertEqual(task_def["pt"], (AutoModelForSequenceClassification,) if is_torch_available() else ())668 self.assertEqual(task_def["tf"], (TFAutoModelForSequenceClassification,) if is_tf_available() else ())669 self.assertEqual(task_def["type"], "text")670 self.assertEqual(task_def["impl"], PairClassificationPipeline)671 self.assertEqual(task_def["default"], {"model": {"pt": "hf-internal-testing/tiny-random-distilbert"}})672 673 # Clean registry for next tests.674 del PIPELINE_REGISTRY.supported_tasks["custom-text-classification"]675 676 @require_torch_or_tf677 def test_dynamic_pipeline(self):678 PIPELINE_REGISTRY.register_pipeline(679 "pair-classification",680 pipeline_class=PairClassificationPipeline,681 pt_model=AutoModelForSequenceClassification if is_torch_available() else None,682 tf_model=TFAutoModelForSequenceClassification if is_tf_available() else None,683 )684 685 classifier = pipeline("pair-classification", model="hf-internal-testing/tiny-random-bert")686 687 # Clean registry as we won't need the pipeline to be in it for the rest to work.688 del PIPELINE_REGISTRY.supported_tasks["pair-classification"]689 690 with tempfile.TemporaryDirectory() as tmp_dir:691 classifier.save_pretrained(tmp_dir)692 # checks693 self.assertDictEqual(694 classifier.model.config.custom_pipelines,695 {696 "pair-classification": {697 "impl": "custom_pipeline.PairClassificationPipeline",698 "pt": ("AutoModelForSequenceClassification",) if is_torch_available() else (),699 "tf": ("TFAutoModelForSequenceClassification",) if is_tf_available() else (),700 }701 },702 )703 # Fails if the user forget to pass along `trust_remote_code=True`704 with self.assertRaises(ValueError):705 _ = pipeline(model=tmp_dir)706 707 new_classifier = pipeline(model=tmp_dir, trust_remote_code=True)708 # Using trust_remote_code=False forces the traditional pipeline tag709 old_classifier = pipeline("text-classification", model=tmp_dir, trust_remote_code=False)710 # Can't make an isinstance check because the new_classifier is from the PairClassificationPipeline class of a711 # dynamic module712 self.assertEqual(new_classifier.__class__.__name__, "PairClassificationPipeline")713 self.assertEqual(new_classifier.task, "pair-classification")714 results = new_classifier("I hate you", second_text="I love you")715 self.assertDictEqual(716 nested_simplify(results),717 {"label": "LABEL_0", "score": 0.505, "logits": [-0.003, -0.024]},718 )719 720 self.assertEqual(old_classifier.__class__.__name__, "TextClassificationPipeline")721 self.assertEqual(old_classifier.task, "text-classification")722 results = old_classifier("I hate you", text_pair="I love you")723 self.assertListEqual(724 nested_simplify(results),725 [{"label": "LABEL_0", "score": 0.505}],726 )727 728 @require_torch_or_tf729 def test_cached_pipeline_has_minimum_calls_to_head(self):730 # Make sure we have cached the pipeline.731 _ = pipeline("text-classification", model="hf-internal-testing/tiny-random-bert")732 with RequestCounter() as counter:733 _ = pipeline("text-classification", model="hf-internal-testing/tiny-random-bert")734 self.assertEqual(counter.get_request_count, 0)735 self.assertEqual(counter.head_request_count, 1)736 self.assertEqual(counter.other_request_count, 0)737 738 @require_torch739 def test_chunk_pipeline_batching_single_file(self):740 # Make sure we have cached the pipeline.741 pipe = pipeline(model="hf-internal-testing/tiny-random-Wav2Vec2ForCTC")742 ds = datasets.load_dataset("hf-internal-testing/librispeech_asr_dummy", "clean", split="validation").sort("id")743 audio = ds[40]["audio"]["array"]744 745 pipe = pipeline(model="hf-internal-testing/tiny-random-Wav2Vec2ForCTC")746 # For some reason scoping doesn't work if not using `self.`747 self.COUNT = 0748 forward = pipe.model.forward749 750 def new_forward(*args, **kwargs):751 self.COUNT += 1752 return forward(*args, **kwargs)753 754 pipe.model.forward = new_forward755 756 for out in pipe(audio, return_timestamps="char", chunk_length_s=3, stride_length_s=[1, 1], batch_size=1024):757 pass758 759 self.assertEqual(self.COUNT, 1)760 761 762@require_torch763@is_staging_test764class DynamicPipelineTester(unittest.TestCase):765 vocab_tokens = ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "I", "love", "hate", "you"]766 767 @classmethod768 def setUpClass(cls):769 cls._token = TOKEN770 HfFolder.save_token(TOKEN)771 772 @classmethod773 def tearDownClass(cls):774 try:775 delete_repo(token=cls._token, repo_id="test-dynamic-pipeline")776 except HTTPError:777 pass778 779 def test_push_to_hub_dynamic_pipeline(self):780 from transformers import BertConfig, BertForSequenceClassification, BertTokenizer781 782 PIPELINE_REGISTRY.register_pipeline(783 "pair-classification",784 pipeline_class=PairClassificationPipeline,785 pt_model=AutoModelForSequenceClassification,786 )787 788 config = BertConfig(789 vocab_size=99, hidden_size=32, num_hidden_layers=5, num_attention_heads=4, intermediate_size=37790 )791 model = BertForSequenceClassification(config).eval()792 793 with tempfile.TemporaryDirectory() as tmp_dir:794 create_repo(f"{USER}/test-dynamic-pipeline", token=self._token)795 repo = Repository(tmp_dir, clone_from=f"{USER}/test-dynamic-pipeline", token=self._token)796 797 vocab_file = os.path.join(tmp_dir, "vocab.txt")798 with open(vocab_file, "w", encoding="utf-8") as vocab_writer:799 vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]))800 tokenizer = BertTokenizer(vocab_file)801 802 classifier = pipeline("pair-classification", model=model, tokenizer=tokenizer)803 804 # Clean registry as we won't need the pipeline to be in it for the rest to work.805 del PIPELINE_REGISTRY.supported_tasks["pair-classification"]806 807 classifier.save_pretrained(tmp_dir)808 # checks809 self.assertDictEqual(810 classifier.model.config.custom_pipelines,811 {812 "pair-classification": {813 "impl": "custom_pipeline.PairClassificationPipeline",814 "pt": ("AutoModelForSequenceClassification",),815 "tf": (),816 }817 },818 )819 820 repo.push_to_hub()821 822 # Fails if the user forget to pass along `trust_remote_code=True`823 with self.assertRaises(ValueError):824 _ = pipeline(model=f"{USER}/test-dynamic-pipeline")825 826 new_classifier = pipeline(model=f"{USER}/test-dynamic-pipeline", trust_remote_code=True)827 # Can't make an isinstance check because the new_classifier is from the PairClassificationPipeline class of a828 # dynamic module829 self.assertEqual(new_classifier.__class__.__name__, "PairClassificationPipeline")830 831 results = classifier("I hate you", second_text="I love you")832 new_results = new_classifier("I hate you", second_text="I love you")833 self.assertDictEqual(nested_simplify(results), nested_simplify(new_results))834 835 # Using trust_remote_code=False forces the traditional pipeline tag836 old_classifier = pipeline(837 "text-classification", model=f"{USER}/test-dynamic-pipeline", trust_remote_code=False838 )839 self.assertEqual(old_classifier.__class__.__name__, "TextClassificationPipeline")840 self.assertEqual(old_classifier.task, "text-classification")841 new_results = old_classifier("I hate you", text_pair="I love you")842 self.assertListEqual(843 nested_simplify([{"label": results["label"], "score": results["score"]}]), nested_simplify(new_results)844 )845