wehub-resource-sync
204 lines
8.6 KiB
Python
204 lines
8.6 KiB
Python
# Copyright (c) 2022 PaddlePaddle Authors. 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 os
|
|
import unittest
|
|
from tempfile import TemporaryDirectory
|
|
|
|
import paddle
|
|
from parameterized import parameterized
|
|
|
|
from paddlenlp.prompt import (
|
|
AutoTemplate,
|
|
PromptModelForSequenceClassification,
|
|
SoftVerbalizer,
|
|
)
|
|
from paddlenlp.taskflow.text_classification import TextClassificationTask
|
|
from paddlenlp.transformers import (
|
|
AutoModelForMaskedLM,
|
|
AutoModelForSequenceClassification,
|
|
AutoTokenizer,
|
|
)
|
|
|
|
|
|
class TestTextClassificationTask(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.temp_dir = TemporaryDirectory()
|
|
tokenizer = AutoTokenizer.from_pretrained("__internal_testing__/ernie")
|
|
|
|
# finetune (dynamic)
|
|
cls.finetune_dygraph_model_path = os.path.join(cls.temp_dir.name, "finetune_dygraph")
|
|
finetune_dygraph_model = AutoModelForSequenceClassification.from_pretrained(
|
|
"__internal_testing__/ernie", num_classes=2
|
|
)
|
|
finetune_dygraph_model.save_pretrained(cls.finetune_dygraph_model_path)
|
|
tokenizer.save_pretrained(cls.finetune_dygraph_model_path)
|
|
|
|
# finetune (static)
|
|
cls.finetune_static_model_path = os.path.join(cls.temp_dir.name, "finetune_static")
|
|
input_spec = [
|
|
paddle.static.InputSpec(shape=[None, None], dtype="int64", name="input_ids"),
|
|
paddle.static.InputSpec(shape=[None, None], dtype="int64", name="token_type_ids"),
|
|
]
|
|
finetune_static_model = paddle.jit.to_static(finetune_dygraph_model, input_spec=input_spec)
|
|
paddle.jit.save(finetune_static_model, os.path.join(cls.finetune_static_model_path, "model"))
|
|
tokenizer.save_pretrained(cls.finetune_static_model_path)
|
|
|
|
# prompt (dynamic)
|
|
cls.prompt_dygraph_model_path = os.path.join(cls.temp_dir.name, "prompt_dygraph")
|
|
prompt_plm_model = AutoModelForMaskedLM.from_pretrained("__internal_testing__/ernie", num_classes=2)
|
|
template = AutoTemplate.create_from("测试：", tokenizer, 16, model=prompt_plm_model)
|
|
label_words = {"negative": ["负面"], "positive": ["正面"]}
|
|
verbalizer = SoftVerbalizer(label_words, tokenizer, prompt_plm_model)
|
|
prompt_dygraph_model = PromptModelForSequenceClassification(prompt_plm_model, template, verbalizer)
|
|
|
|
template.save(cls.prompt_dygraph_model_path)
|
|
verbalizer.save(cls.prompt_dygraph_model_path)
|
|
tokenizer.save_pretrained(cls.prompt_dygraph_model_path)
|
|
prompt_plm_model.save_pretrained(os.path.join(cls.prompt_dygraph_model_path, "plm"))
|
|
state_dict = prompt_dygraph_model.state_dict()
|
|
paddle.save(state_dict, os.path.join(cls.prompt_dygraph_model_path, "model_state.pdparams"))
|
|
|
|
# prompt (static)
|
|
cls.prompt_static_model_path = os.path.join(cls.temp_dir.name, "prompt_static")
|
|
input_spec = prompt_dygraph_model.get_input_spec()
|
|
prompt_static_model = paddle.jit.to_static(prompt_dygraph_model, input_spec=input_spec)
|
|
|
|
template.save(cls.prompt_static_model_path)
|
|
verbalizer.save(cls.prompt_static_model_path)
|
|
tokenizer.save_pretrained(cls.prompt_static_model_path)
|
|
prompt_plm_model.save_pretrained(os.path.join(cls.prompt_static_model_path, "plm"))
|
|
paddle.jit.save(prompt_static_model, os.path.join(cls.prompt_static_model_path, "model"))
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.temp_dir.cleanup()
|
|
|
|
@parameterized.expand(
|
|
[
|
|
(1, "multi_class", "finetune"),
|
|
# (1, "multi_class", "prompt"), # TODO (paddle 2.5.1 breaks this test)
|
|
(1, "multi_label", "finetune"),
|
|
# (1, "multi_label", "prompt"), # TODO (paddle 2.5.1 breaks this test)
|
|
]
|
|
)
|
|
def test_classification_task(self, batch_size, problem_type, model):
|
|
# input_text is a tuple to simulate the args passed from Taskflow to TextClassificationTask
|
|
input_text = (["百度", "深度学习框架", "飞桨", "PaddleNLP"],)
|
|
id2label = {
|
|
0: "negative",
|
|
1: "positive",
|
|
}
|
|
if model == "finetune":
|
|
dygraph_model_path = self.finetune_dygraph_model_path
|
|
static_model_path = self.finetune_static_model_path
|
|
else:
|
|
dygraph_model_path = self.prompt_dygraph_model_path
|
|
static_model_path = self.prompt_static_model_path
|
|
|
|
dygraph_taskflow = TextClassificationTask(
|
|
model=model,
|
|
task="text_classification",
|
|
task_path=dygraph_model_path,
|
|
id2label=id2label,
|
|
batch_size=batch_size,
|
|
device_id=0,
|
|
problem_type=problem_type,
|
|
)
|
|
|
|
dygraph_results = dygraph_taskflow(input_text)
|
|
|
|
self.assertEqual(len(dygraph_results), len(input_text[0]))
|
|
|
|
static_taskflow = TextClassificationTask(
|
|
model=model,
|
|
task="text_classification",
|
|
is_static_model=True,
|
|
task_path=static_model_path,
|
|
id2label=id2label,
|
|
batch_size=batch_size,
|
|
device_id=0,
|
|
problem_type=problem_type,
|
|
)
|
|
|
|
static_results = static_taskflow(input_text)
|
|
self.assertEqual(len(static_results), len(input_text[0]))
|
|
|
|
for dygraph_result, static_result in zip(dygraph_results, static_results):
|
|
for dygraph_pred, static_pred in zip(dygraph_result["predictions"], static_result["predictions"]):
|
|
self.assertEqual(dygraph_pred["label"], static_pred["label"])
|
|
self.assertAlmostEqual(dygraph_pred["score"], static_pred["score"], delta=1e-6)
|
|
# if multi_label, all predictions should be greater than the threshold
|
|
if model == "multi_label":
|
|
self.assertGreater(dygraph_pred["score"], dygraph_taskflow.multilabel_threshold)
|
|
|
|
# @unittest.skip("numerical error")
|
|
# @parameterized.expand(
|
|
# [
|
|
# (1, "multi_class", "finetune"),
|
|
# (1, "multi_class", "prompt"),
|
|
# (1, "multi_label", "finetune"),
|
|
# (1, "multi_label", "prompt"),
|
|
# ]
|
|
# )
|
|
# def test_taskflow_task(self, batch_size, problem_type, mode):
|
|
# input_text = ["百度", "深度学习框架", "飞桨", "PaddleNLP"]
|
|
# id2label = {
|
|
# 0: "negative",
|
|
# 1: "positive",
|
|
# }
|
|
# if mode == "finetune":
|
|
# dygraph_model_path = self.finetune_dygraph_model_path
|
|
# static_model_path = self.finetune_static_model_path
|
|
# else:
|
|
# dygraph_model_path = self.prompt_dygraph_model_path
|
|
# static_model_path = self.prompt_static_model_path
|
|
|
|
# dygraph_taskflow = Taskflow(
|
|
# mode=mode,
|
|
# task="text_classification",
|
|
# task_path=dygraph_model_path,
|
|
# id2label=id2label,
|
|
# batch_size=batch_size,
|
|
# device_id=0,
|
|
# problem_type=problem_type,
|
|
# )
|
|
|
|
# dygraph_results = dygraph_taskflow(input_text)
|
|
|
|
# self.assertEqual(len(dygraph_results), len(input_text))
|
|
|
|
# static_taskflow = Taskflow(
|
|
# mode=mode,
|
|
# task="text_classification",
|
|
# is_static_model=True,
|
|
# task_path=static_model_path,
|
|
# id2label=id2label,
|
|
# batch_size=batch_size,
|
|
# device_id=0,
|
|
# problem_type=problem_type,
|
|
# )
|
|
|
|
# static_results = static_taskflow(input_text)
|
|
# self.assertEqual(len(static_results), len(input_text))
|
|
|
|
# for dygraph_result, static_result in zip(dygraph_results, static_results):
|
|
# for dygraph_pred, static_pred in zip(dygraph_result["predictions"], static_result["predictions"]):
|
|
# self.assertEqual(dygraph_pred["label"], static_pred["label"])
|
|
# self.assertAlmostEqual(dygraph_pred["score"], static_pred["score"], delta=1e-6)
|
|
# # if multi_label, all predictions should be greater than the threshold
|
|
# if mode == "multi_label":
|
|
# self.assertGreater(dygraph_pred["score"], dygraph_taskflow.task_instance.multilabel_threshold)
|