# Copyright (c) 2023 PaddlePaddle Authors. All Rights Reserved. # Copyright 2020 The HuggingFace 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. from __future__ import annotations import copy import inspect import os import random import shutil import subprocess import tempfile import time import unittest from typing import Optional, Tuple, Type import numpy as np import paddle from paddle.distributed.utils.launch_utils import ( TrainerProc, find_free_ports, get_cluster, watch_local_trainers, ) from paddlenlp.taskflow.utils import static_mode_guard from paddlenlp.transformers import AutoModelForCausalLM, AutoTokenizer from paddlenlp.transformers.configuration_utils import PretrainedConfig from paddlenlp.transformers.model_utils import PretrainedModel from paddlenlp.utils.env import ( # MODEL_HOME, CONFIG_NAME, LEGACY_CONFIG_NAME, PADDLE_INFERENCE_MODEL_SUFFIX, PADDLE_INFERENCE_WEIGHTS_SUFFIX, ) from ..testing_utils import slow def _config_zero_init(config): configs_no_init = copy.deepcopy(config) for key in configs_no_init.__dict__.keys(): if "_range" in key or "_std" in key or "initializer_factor" in key or "layer_scale" in key: setattr(configs_no_init, key, 1e-10) return configs_no_init def get_cluster_from_args(selected_gpus): cluster_node_ips = "127.0.0.1" node_ip = "127.0.0.1" node_ips = [x.strip() for x in cluster_node_ips.split(",")] node_ips.index(node_ip) free_ports = None free_ports = find_free_ports(len(selected_gpus)) if free_ports is not None: free_ports = list(free_ports) trainer_endpoints = [] for ip in node_ips: trainer_endpoints.append(["%s:%d" % (ip, port) for port in free_ports]) return get_cluster(node_ips, node_ip, trainer_endpoints, selected_gpus) def get_gpus(selected_gpus): selected_gpus = [x.strip() for x in selected_gpus.split(",")] return selected_gpus def start_local_trainers_cpu(trainer_endpoints, training_script, training_script_args, log_dir=None): current_env = copy.copy(os.environ.copy()) current_env.pop("http_proxy", None) current_env.pop("https_proxy", None) procs = [] n_rank = len(trainer_endpoints) print(trainer_endpoints) for rank_id, endpoint in enumerate(trainer_endpoints): proc_env = { "PADDLE_DISTRI_BACKEND": "gloo", "PADDLE_TRAINER_ID": "%d" % rank_id, "PADDLE_CURRENT_ENDPOINT": "%s" % endpoint, "PADDLE_TRAINERS_NUM": "%d" % n_rank, "PADDLE_TRAINER_ENDPOINTS": ",".join(trainer_endpoints), } current_env.update(proc_env) print("trainer proc env:{}".format(current_env)) assert os.getenv("WITH_COVERAGE", "OFF") == "OFF", "Gloo don't support WITH_COVERAGE." cmd = "python -u " + training_script print("start trainer proc:{} env:{}".format(cmd, proc_env)) fn = None proc = subprocess.Popen(cmd.split(" "), env=current_env) tp = TrainerProc() tp.proc = proc tp.rank = rank_id tp.log_fn = fn tp.cmd = cmd procs.append(tp) return procs def start_local_trainers( cluster, pod, training_script, training_script_args="", eager_mode=True, allocator_strategy="auto_growth", log_dir=None, without_http_proxy=True, ): current_env = copy.copy(os.environ.copy()) # paddle broadcast ncclUniqueId use socket, and # proxy maybe make trainers unreachable, so delete them. # if we set them to "", grpc will log error message "bad uri" # so just delete them. # current_env.pop("http_proxy", None) # current_env.pop("https_proxy", None) # parse args if isinstance(training_script_args, dict): training_script_args = [f"--{k} {v}" for k, v in training_script_args.items()] if isinstance(training_script_args, list): training_script_args = " ".join(training_script_args) procs = [] for t in pod.trainers: proc_env = { "FLAGS_selected_gpus": "%s" % ",".join([str(g) for g in t.gpus]), "PADDLE_TRAINER_ID": "%d" % t.rank, "PADDLE_CURRENT_ENDPOINT": "%s" % t.endpoint, "PADDLE_TRAINERS_NUM": "%d" % cluster.trainers_nranks(), "PADDLE_TRAINER_ENDPOINTS": ",".join(cluster.trainers_endpoints()), } proc_env["FLAGS_allocator_strategy"] = allocator_strategy if allocator_strategy == "auto_growth": proc_env["FLAGS_fraction_of_gpu_memory_to_use"] = "0.1" current_env.update(proc_env) if os.getenv("WITH_COVERAGE", "OFF") == "ON": cmd = "python -m coverage run --branch -p " + training_script else: cmd = f"python -u {training_script} {training_script_args}" print("start trainer proc:{} env:{}".format(cmd, proc_env)) fn = None proc = subprocess.Popen(cmd.split(" "), env=current_env) tp = TrainerProc() tp.proc = proc tp.rank = t.rank tp.log_fn = fn tp.cmd = cmd procs.append(tp) return procs def ids_tensor(shape, vocab_size, dtype="int32"): # Creates a random int32 tensor of the shape within the vocab size return paddle.randint(low=1, high=vocab_size, dtype=dtype, shape=shape) def random_attention_mask(shape, dtype="int32"): attn_mask = ids_tensor(shape, vocab_size=2, dtype=dtype) # make sure that at least one token is attended to for each batch attn_mask[:, -1] = 1 return attn_mask def floats_tensor(shape, scale=1.0): """Creates a random float32 tensor""" return scale * paddle.randn(shape, dtype="float32") def check_two_model_parameter(first_model: PretrainedModel, second_model: PretrainedModel): assert len(set(first_model.state_dict().keys()) - set(second_model.state_dict().keys())) == 0 # random choice the keys to compare key = random.choice(list(first_model.state_dict().keys())) diff = first_model.state_dict()[key] - second_model.state_dict()[key] assert diff.sum().item() == 0 class ModelTesterMixin: model_tester = None base_model_class: Optional[Type[PretrainedModel]] = None all_model_classes: Tuple[Type[PretrainedModel]] = () all_generative_model_classes = () test_resize_embeddings = True test_resize_position_embeddings = False test_mismatched_shapes = True test_missing_keys = True test_model_compatibility_keys = True test_tie_weights = False use_test_inputs_embeds = False use_test_model_name_list = True is_encoder_decoder = False has_attentions = True model_split_percents = [0.5, 0.7, 0.9] def _prepare_for_class(self, inputs_dict, model_class): inputs_dict = copy.deepcopy(inputs_dict) if model_class.__name__.endswith("ForMultipleChoice"): inputs_dict = { k: v.unsqueeze(1).expand(shape=[-1, self.model_tester.num_choices, -1]) if isinstance(v, paddle.Tensor) and v.ndim > 1 else v for k, v in inputs_dict.items() } return inputs_dict def _make_model_instance(self, config, model_class): if isinstance(config, PretrainedConfig): return model_class(config) if model_class == self.base_model_class: return model_class(**config) return model_class(self.base_model_class(**config)) def test_save_load(self): config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common() def check_save_load(out1, out2): # make sure we don't have nans out_2 = out2.numpy() out_2[np.isnan(out_2)] = 0 out_1 = out1.numpy() out_1[np.isnan(out_1)] = 0 max_diff = np.amax(np.abs(out_1 - out_2)) self.assertLessEqual(max_diff, 1e-5) for model_class in self.all_model_classes: model = self._make_model_instance(config, model_class) model.eval() with paddle.no_grad(): first = model(**self._prepare_for_class(inputs_dict, model_class))[0] with tempfile.TemporaryDirectory() as tmpdirname: model.save_pretrained(tmpdirname) model = model_class.from_pretrained(tmpdirname) model.eval() with paddle.no_grad(): second = model(**self._prepare_for_class(inputs_dict, model_class))[0] # support tuple of tensor if isinstance(first, tuple) and isinstance(second, tuple): for tensor1, tensor2 in zip(first, second): check_save_load(tensor1, tensor2) else: check_save_load(first, second) def test_determinism(self): config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common() def check_determinism(first, second): out_1 = first.numpy() out_2 = second.numpy() out_1 = out_1[~np.isnan(out_1)] out_2 = out_2[~np.isnan(out_2)] max_diff = np.amax(np.abs(out_1 - out_2)) self.assertLessEqual(max_diff, 1e-5) for model_class in self.all_model_classes: model = self._make_model_instance(config, model_class) model.eval() with paddle.no_grad(): first = model(**self._prepare_for_class(inputs_dict, model_class))[0] second = model(**self._prepare_for_class(inputs_dict, model_class))[0] if isinstance(first, tuple) and isinstance(second, tuple): for tensor1, tensor2 in zip(first, second): check_determinism(tensor1, tensor2) else: check_determinism(first, second) def test_forward_signature(self): config, _ = self.model_tester.prepare_config_and_inputs_for_common() for model_class in self.all_model_classes: model = self._make_model_instance(config, model_class) signature = inspect.signature(model.forward) # signature.parameters is an OrderedDict => so arg_names order is deterministic arg_names = [*signature.parameters.keys()] expected_arg_names = ["input_ids"] self.assertListEqual(arg_names[:1], expected_arg_names) @unittest.skip("Not implemented yet") def test_training(self): # TODO(guosheng): add more tests for training if loss is implemented pass @unittest.skip("Not implemented yet") def test_training_gradient_checkpointing(self): pass def test_attention_outputs(self): if not self.has_attentions: return config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common() seq_len = getattr(self.model_tester, "seq_length", None) decoder_seq_length = getattr(self.model_tester, "decoder_seq_length", seq_len) encoder_seq_length = getattr(self.model_tester, "encoder_seq_length", seq_len) decoder_key_length = getattr(self.model_tester, "decoder_key_length", decoder_seq_length) encoder_key_length = getattr(self.model_tester, "key_length", encoder_seq_length) chunk_length = getattr(self.model_tester, "chunk_length", None) if chunk_length is not None and hasattr(self.model_tester, "num_hashes"): encoder_seq_length = encoder_seq_length * self.model_tester.num_hashes for model_class in self.all_model_classes: signature = inspect.signature(model_class.forward) # signature.parameters is an OrderedDict => so arg_names order is deterministic arg_names = [*signature.parameters.keys()] if not all(name in arg_names for name in ["output_attentions", "output_hidden_states", "return_dict"]): continue inputs_dict["output_attentions"] = True inputs_dict["output_hidden_states"] = False inputs_dict["return_dict"] = True model = self._make_model_instance(config, model_class) model.eval() with paddle.no_grad(): outputs = model(**self._prepare_for_class(inputs_dict, model_class)) attentions = outputs.encoder_attentions if self.is_encoder_decoder else outputs.attentions self.assertEqual(len(attentions), self.model_tester.num_hidden_layers) # TODO(guosheng): check that output_attentions also work using config if chunk_length is not None: self.assertListEqual( list(attentions[0].shape[-4:]), [self.model_tester.num_attention_heads, encoder_seq_length, chunk_length, encoder_key_length], ) else: self.assertListEqual( list(attentions[0].shape[-3:]), [self.model_tester.num_attention_heads, encoder_seq_length, encoder_key_length], ) out_len = len(outputs) if self.is_encoder_decoder: correct_outlen = 5 # loss is at first position if "labels" in inputs_dict: correct_outlen += 1 # loss is added to beginning # Question Answering model returns start_logits and end_logits if model_class.__name__.endswith("ForQuestionAnswering"): correct_outlen += 1 # start_logits and end_logits instead of only 1 output if "past_key_values" in outputs: correct_outlen += 1 # past_key_values have been returned self.assertEqual(out_len, correct_outlen) # decoder attentions decoder_attentions = outputs.decoder_attentions self.assertIsInstance(decoder_attentions, (list, tuple)) self.assertEqual(len(decoder_attentions), self.model_tester.num_hidden_layers) self.assertListEqual( list(decoder_attentions[0].shape[-3:]), [self.model_tester.num_attention_heads, decoder_seq_length, decoder_key_length], ) # cross attentions cross_attentions = outputs.cross_attentions self.assertIsInstance(cross_attentions, (list, tuple)) self.assertEqual(len(cross_attentions), self.model_tester.num_hidden_layers) self.assertListEqual( list(cross_attentions[0].shape[-3:]), [ self.model_tester.num_attention_heads, decoder_seq_length, encoder_key_length, ], ) # Check attention is always last and order is fine inputs_dict["output_attentions"] = True inputs_dict["output_hidden_states"] = True model = self._make_model_instance(config, model_class) model.eval() with paddle.no_grad(): outputs = model(**self._prepare_for_class(inputs_dict, model_class)) if hasattr(self.model_tester, "num_hidden_states_types"): added_hidden_states = self.model_tester.num_hidden_states_types elif self.is_encoder_decoder: added_hidden_states = 2 else: added_hidden_states = 1 self.assertEqual(out_len + added_hidden_states, len(outputs)) self_attentions = outputs.encoder_attentions if self.is_encoder_decoder else outputs.attentions self.assertEqual(len(self_attentions), self.model_tester.num_hidden_layers) if chunk_length is not None: self.assertListEqual( list(self_attentions[0].shape[-4:]), [self.model_tester.num_attention_heads, encoder_seq_length, chunk_length, encoder_key_length], ) else: self.assertListEqual( list(self_attentions[0].shape[-3:]), [self.model_tester.num_attention_heads, encoder_seq_length, encoder_key_length], ) def test_hidden_states_output(self): def check_hidden_states_output(inputs_dict, config, model_class): model = self._make_model_instance(config, model_class) model.eval() with paddle.no_grad(): outputs = model(**self._prepare_for_class(inputs_dict, model_class)) hidden_states = outputs.encoder_hidden_states if self.is_encoder_decoder else outputs.hidden_states expected_num_layers = getattr( self.model_tester, "expected_num_hidden_layers", self.model_tester.num_hidden_layers + 1 ) self.assertEqual(len(hidden_states), expected_num_layers) if hasattr(self.model_tester, "encoder_seq_length"): seq_length = self.model_tester.encoder_seq_length if hasattr(self.model_tester, "chunk_length") and self.model_tester.chunk_length > 1: seq_length = seq_length * self.model_tester.chunk_length else: seq_length = self.model_tester.seq_length self.assertListEqual( list(hidden_states[0].shape[-2:]), [seq_length, self.model_tester.hidden_size], ) if self.is_encoder_decoder: hidden_states = outputs.decoder_hidden_states self.assertIsInstance(hidden_states, (list, tuple)) self.assertEqual(len(hidden_states), expected_num_layers) seq_len = getattr(self.model_tester, "seq_length", None) decoder_seq_length = getattr(self.model_tester, "decoder_seq_length", seq_len) self.assertListEqual( list(hidden_states[0].shape[-2:]), [decoder_seq_length, self.model_tester.hidden_size], ) config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common() inputs_dict["return_dict"] = True for model_class in self.all_model_classes: signature = inspect.signature(model_class.forward) # signature.parameters is an OrderedDict => so arg_names order is deterministic arg_names = [*signature.parameters.keys()] if not all(name in arg_names for name in ["output_attentions", "output_hidden_states", "return_dict"]): continue inputs_dict["output_hidden_states"] = True check_hidden_states_output(inputs_dict, config, model_class) # TODO(guosheng): check that output_hidden_states also work using config @unittest.skip("Not implemented") def test_retain_grad_hidden_states_attentions(self): pass def test_resize_position_vector_embeddings(self): if not self.test_resize_position_embeddings: return ( original_config, inputs_dict, ) = self.model_tester.prepare_config_and_inputs_for_common() for model_class in self.all_model_classes: config = copy.deepcopy(original_config) model = self._make_model_instance(config, model_class) if self.model_tester.is_training is False: model.eval() max_position_embeddings = config.max_position_embeddings # Retrieve the embeddings and clone theme if self.is_encoder_decoder: encoder_model_embed, decoder_model_embed = model.get_position_embeddings() encoder_cloned_embeddings = encoder_model_embed.weight.clone() decoder_cloned_embeddings = decoder_model_embed.weight.clone() else: model_embed = model.get_position_embeddings() cloned_embeddings = model_embed.weight.clone() # Check that resizing the position embeddings with a larger max_position_embeddings increases # the model's position embeddings size model.resize_position_embeddings(max_position_embeddings + 10) self.assertEqual(model.config.max_position_embeddings, max_position_embeddings + 10) # Check that it actually resizes the embeddings matrix if model.config.is_encoder_decoder: encoder_model_embed, decoder_model_embed = model.get_position_embeddings() self.assertEqual(encoder_model_embed.weight.shape[0], encoder_cloned_embeddings.shape[0] + 10) self.assertEqual(decoder_model_embed.weight.shape[0], decoder_cloned_embeddings.shape[0] + 10) else: model_embed = model.get_position_embeddings() self.assertEqual(model_embed.weight.shape[0], cloned_embeddings.shape[0] + 10) # Check that the model can still do a forward pass successfully (every parameter should be resized) model(**self._prepare_for_class(inputs_dict, model_class)) # Check that resizing the position embeddings with a smaller max_position_embeddings decreases # the model's max_position_embeddings model.resize_position_embeddings(max_position_embeddings - 5) self.assertEqual(model.base_model.config["max_position_embeddings"], max_position_embeddings - 5) # Check that it actually resizes the embeddings matrix if self.is_encoder_decoder: encoder_model_embed, decoder_model_embed = model.get_position_embeddings() self.assertEqual(encoder_model_embed.weight.shape[0], encoder_cloned_embeddings.shape[0] - 5) self.assertEqual(decoder_model_embed.weight.shape[0], decoder_cloned_embeddings.shape[0] - 5) else: model_embed = model.get_position_embeddings() self.assertEqual(model_embed.weight.shape[0], cloned_embeddings.shape[0] - 5) # Check that the model can still do a forward pass successfully (every parameter should be resized) model(**self._prepare_for_class(inputs_dict, model_class)) # Check that adding and removing tokens has not modified the first part of the embedding matrix. models_equal = True if model.config.is_encoder_decoder: for p1, p2 in zip(encoder_cloned_embeddings, encoder_model_embed.weight): if p1.data.ne(p2.data).sum() > 0: models_equal = False for p1, p2 in zip(decoder_cloned_embeddings, decoder_model_embed.weight): if p1.data.ne(p2.data).sum() > 0: models_equal = False else: for p1, p2 in zip(cloned_embeddings, model_embed.weight): if p1.data.ne(p2.data).sum() > 0: models_equal = False self.assertTrue(models_equal) def test_resize_tokens_embeddings(self): ( original_config, inputs_dict, ) = self.model_tester.prepare_config_and_inputs_for_common() if not self.test_resize_embeddings: return for model_class in self.all_model_classes: config = copy.deepcopy(original_config) model = self._make_model_instance(config, model_class) if self.model_tester.is_training is False: model.eval() model_vocab_size = config.vocab_size # Retrieve the embeddings and clone theme model_embed = model.resize_token_embeddings(model_vocab_size) cloned_embeddings = model_embed.weight.clone() # Check that resizing the token embeddings with a larger vocab size increases the model's vocab size model_embed = model.resize_token_embeddings(model_vocab_size + 10) self.assertEqual(model.base_model.config.vocab_size, model_vocab_size + 10) # Check that it actually resizes the embeddings matrix self.assertEqual(model_embed.weight.shape[0], cloned_embeddings.shape[0] + 10) # Check that the model can still do a forward pass successfully (every parameter should be resized) model(**self._prepare_for_class(inputs_dict, model_class)) # Check that resizing the token embeddings with a smaller vocab size decreases the model's vocab size model_embed = model.resize_token_embeddings(model_vocab_size - 15) self.assertEqual(model.base_model.config.vocab_size, model_vocab_size - 15) # Check that it actually resizes the embeddings matrix self.assertEqual(model_embed.weight.shape[0], cloned_embeddings.shape[0] - 15) # Check that the model can still do a forward pass successfully (every parameter should be resized) # Input ids should be clamped to the maximum size of the vocabulary inputs_dict["input_ids"] = paddle.clip(inputs_dict["input_ids"], max=model_vocab_size - 15 - 1) # make sure that decoder_input_ids are resized as well if "decoder_input_ids" in inputs_dict: inputs_dict["decoder_input_ids"] = paddle.clip( inputs_dict["decoder_input_ids"], max=model_vocab_size - 15 - 1 ) model(**self._prepare_for_class(inputs_dict, model_class)) # Check that adding and removing tokens has not modified the first part of the embedding matrix. models_equal = True for p1, p2 in zip(cloned_embeddings, model_embed.weight): if not paddle.equal_all(p1, p2).item(): models_equal = False break self.assertTrue(models_equal) def _compare_tensor(self, tensor1, tensor2, rtol=1e-04, atol=1e-04): if tensor1.dtype != tensor2.dtype: return False if tensor1.dtype in [paddle.float32, paddle.float64]: return paddle.allclose(tensor1, tensor2, rtol=rtol, atol=atol) else: return paddle.equal_all(tensor1, tensor2) def test_inputs_embeds(self): # pass the test if don't need to test inputs embeddings if not self.use_test_inputs_embeds: return # get config for model and inputs_dict for model forward config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common() # test all model classes for model_class in self.all_model_classes: model = self._make_model_instance(config, model_class) model.eval() inputs = copy.deepcopy(self._prepare_for_class(inputs_dict, model_class)) with paddle.no_grad(): ids_output = model(**inputs) if not self.is_encoder_decoder: input_ids = inputs["input_ids"] del inputs["input_ids"] else: encoder_input_ids = inputs["input_ids"] decoder_input_ids = inputs.get("decoder_input_ids", encoder_input_ids) del inputs["input_ids"] inputs.pop("decoder_input_ids", None) wte = model.get_input_embeddings() if not self.is_encoder_decoder: inputs["inputs_embeds"] = wte(input_ids) else: inputs["inputs_embeds"] = wte(encoder_input_ids) inputs["decoder_inputs_embeds"] = wte(decoder_input_ids) with paddle.no_grad(): embeds_output = model(**inputs) if isinstance(embeds_output, paddle.Tensor): self.assertTrue(self._compare_tensor(ids_output, embeds_output)) else: for ids_item, embeds_item in zip(ids_output, embeds_output): self.assertTrue(self._compare_tensor(ids_item, embeds_item)) def test_model_name_list(self): if not self.use_test_model_name_list: return config = self.model_tester.get_config() if isinstance(config, PretrainedConfig): model = self.base_model_class(config) else: model = self.base_model_class(**config) self.assertTrue(len(model.model_name_list) != 0) def test_pretrained_config_save_load(self): if self.base_model_class is None or not self.base_model_class.constructed_from_pretrained_config(): return config_class = self.base_model_class.config_class with tempfile.TemporaryDirectory() as tempdir: config = config_class() config.save_pretrained(tempdir) # check the file exist self.assertFalse(os.path.exists(os.path.join(tempdir, LEGACY_CONFIG_NAME))) self.assertTrue(os.path.exists(os.path.join(tempdir, CONFIG_NAME))) # rename the CONFIG_NAME shutil.move(os.path.join(tempdir, CONFIG_NAME), os.path.join(tempdir, LEGACY_CONFIG_NAME)) loaded_config = config.__class__.from_pretrained(tempdir) for key in config.__dict__.keys(): if key == "paddlenlp_version" and config.paddlenlp_version is None: continue self.assertEqual(getattr(config, key), getattr(loaded_config, key)) def random_choice_pretrained_config_field(self) -> Optional[str]: if self.base_model_class is None or not self.base_model_class.constructed_from_pretrained_config(): return None config = self.base_model_class.config_class() fields = [key for key, value in config.to_dict() if value] return random.choice(fields) def test_for_missed_attribute(self): if not self.test_model_compatibility_keys: self.skipTest(f"Do not test model_compatibility_keys on {self.base_model_class}") return config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common() for model_class in self.all_model_classes: if not model_class.constructed_from_pretrained_config(): continue model = self._make_model_instance(config, model_class) all_maps: dict = copy.deepcopy(model_class.config_class.attribute_map) for old_attribute, new_attribute in all_maps.items(): old_value = getattr(model.config, old_attribute) new_value = getattr(model.config, new_attribute) # eg: dropout can be an instance of nn.Dropout, so we should check it attribute if type(new_value) != type(old_value): continue self.assertEqual(old_value, new_value) def test_tie_weight(self): # test whether id of input_embeding equal id of output_embeding ? if not self.test_tie_weights: return config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common() for model_class in self.all_model_classes: if "CausalLM" not in model_class.__name__ and "MaskedLM" not in model_class.__name__: continue model = self._make_model_instance(config, model_class) if not model.config.tie_word_embeddings: continue if hasattr(model, "get_input_embeddings") and hasattr(model, "get_output_embeddings"): try: input_embeddings = model.get_input_embeddings() except NotImplementedError: continue try: output_embeddings = model.get_output_embeddings() except NotImplementedError: continue if input_embeddings is not None and output_embeddings is not None: if hasattr(output_embeddings, "weight"): output_embeddings_weight = output_embeddings.weight else: output_embeddings_weight = output_embeddings if hasattr(input_embeddings, "weight"): input_embeddings_weight = input_embeddings.weight else: input_embeddings_weight = input_embeddings print( input_embeddings_weight, output_embeddings_weight, ) print( "model name :{},id is{},{}".format( model_class, id(output_embeddings_weight), id(input_embeddings_weight) ) ) self.assertEqual(id(output_embeddings_weight), id(input_embeddings_weight)) class ModelTesterPretrainedMixin: base_model_class: PretrainedModel = None hf_remote_test_model_path: str = None paddlehub_remote_test_model_path: str = None # Download from HF doesn't work in CI yet @slow def test_model_from_pretrained_hf_hub(self): if self.hf_remote_test_model_path is None or self.base_model_class is None: return model = self.base_model_class.from_pretrained(self.hf_remote_test_model_path, from_hf_hub=True) self.assertIsNotNone(model) def test_model_from_pretrained_paddle_hub(self): if self.paddlehub_remote_test_model_path is None or self.base_model_class is None: return model = self.base_model_class.from_pretrained(self.paddlehub_remote_test_model_path) self.assertIsNotNone(model) def test_model_from_config_paddle_hub(self): if self.paddlehub_remote_test_model_path is None or self.base_model_class is None: return config = self.base_model_class.config_class.from_pretrained(self.paddlehub_remote_test_model_path) model = self.base_model_class.from_config(config) self.assertIsNotNone(model) @slow def test_model_from_pretrained_with_cache_dir(self): for model_name in list(self.base_model_class.pretrained_init_configuration)[:1]: with tempfile.TemporaryDirectory() as tempdir: tempdir = str(tempdir) model = self.base_model_class.from_pretrained(model_name, cache_dir=tempdir) self.assertIsNotNone(model) self.assertTrue( os.path.isfile( os.path.join(tempdir, model_name, self.base_model_class.resource_files_names["model_state"]) ) ) self.assertTrue( os.path.isfile(os.path.join(tempdir, model_name, self.base_model_class.model_config_file)) ) @slow def test_pretrained_save_and_load(self): """test the pretrained model save and load with two different ways: url-file-name & model_state name eg: `bert-base-uncased.pdparams` and `model_state.pdparams` """ for model_name in list(self.base_model_class.pretrained_init_configuration)[:1]: model = self.base_model_class.from_pretrained(model_name) self.assertIsNotNone(model) # 1. save and load with tempfile.TemporaryDirectory() as tempdir: tempdirname = str(tempdir) model.save_pretrained(tempdirname) loaded_model = self.base_model_class.from_pretrained(tempdirname) check_two_model_parameter(model, loaded_model) class DistributedTest(unittest.TestCase): def setUp(self) -> None: self.gpus = "0,1" def get_world_size(self): return len(self.gpus.split(",")) def run_on_gpu( self, training_script, training_script_args=None, gpus: str = "0,1", eager_mode=True, allocator_strategy="auto_growth", ): if not paddle.framework.core.is_compiled_with_cuda() or paddle.framework.core.get_cuda_device_count() == 0: return cluster = None pod = None cluster, pod = get_cluster_from_args(get_gpus(gpus)) procs = start_local_trainers( cluster, pod, eager_mode=eager_mode, allocator_strategy=allocator_strategy, training_script=training_script, training_script_args=training_script_args, ) while True: alive = watch_local_trainers(procs, cluster.trainers_endpoints()) if not alive: print("Local procs complete, POD info:{}".format(pod)) break time.sleep(3) class GenerationD2STestMixin: article = """Justin Timberlake and Jessica Biel, welcome to parenthood.""" internal_testing_model = None TokenizerClass = AutoTokenizer CausalLMClass = AutoModelForCausalLM max_new_tokens = 20 def setUp(self): paddle.disable_static() super().setUp() def tearDown(self): paddle.disable_static() super().setUp() @unittest.skip("Paddle enable PIR API in Python") def test_to_static_use_top_k(self): tokenizer = self.TokenizerClass.from_pretrained(self.internal_testing_model) if tokenizer.__class__.__name__ == "LlamaTokenizer": tokenizer.pad_token = tokenizer.eos_token if tokenizer.eos_token else "" model = self.CausalLMClass.from_pretrained(self.internal_testing_model) model_kwargs = tokenizer( self.article, max_length=self.max_new_tokens, truncation=True, truncation_side="left", return_tensors="pd", padding=True, add_special_tokens=True, ) model.is_encoder_decoder = False model.eval() model_kwargs["use_cache"] = True model_kwargs["max_length"] = self.max_new_tokens + model_kwargs["input_ids"].shape[-1] decoded_ids = model.greedy_search( logits_processors=None, bos_token_id=model.config.bos_token_id, pad_token_id=model.config.pad_token_id, eos_token_id=model.config.eos_token_id, **model_kwargs, )[0] dygraph_decoded_ids = decoded_ids.tolist() with static_mode_guard(): with tempfile.TemporaryDirectory() as tempdir: path = os.path.join(tempdir, "model") model.to_static( path, config=dict( use_top_p=False, ), ) model_path = os.path.join(tempdir, f"model{PADDLE_INFERENCE_MODEL_SUFFIX}") params_path = os.path.join(tempdir, f"model{PADDLE_INFERENCE_WEIGHTS_SUFFIX}") config = paddle.inference.Config(model_path, params_path) config.disable_gpu() config.disable_glog_info() predictor = paddle.inference.create_predictor(config) model_kwargs["top_k"] = 1 model_kwargs["max_new_tokens"] = self.max_new_tokens # create input for key in model_kwargs.keys(): if paddle.is_tensor(model_kwargs[key]): model_kwargs[key] = model_kwargs[key].numpy() elif isinstance(model_kwargs[key], float): model_kwargs[key] = np.array(model_kwargs[key], dtype="float32") else: model_kwargs[key] = np.array(model_kwargs[key], dtype="int64") input_handles = {} for name in predictor.get_input_names(): input_handles[name] = predictor.get_input_handle(name) input_handles[name].copy_from_cpu(model_kwargs[name]) predictor.run() output_names = predictor.get_output_names() output_handle = predictor.get_output_handle(output_names[0]) results = output_handle.copy_to_cpu() static_decoded_ids = results.tolist() self.assertEqual(len(dygraph_decoded_ids[0]), self.max_new_tokens) self.assertEqual(len(static_decoded_ids[0]), self.max_new_tokens) self.assertEqual(dygraph_decoded_ids, static_decoded_ids) @unittest.skip("Paddle enable PIR API in Python") def test_to_static_use_top_p(self): tokenizer = self.TokenizerClass.from_pretrained(self.internal_testing_model) if tokenizer.__class__.__name__ == "LlamaTokenizer": tokenizer.pad_token = tokenizer.eos_token if tokenizer.eos_token else "" model = self.CausalLMClass.from_pretrained(self.internal_testing_model) model_kwargs = tokenizer( self.article, max_length=self.max_new_tokens, truncation=True, truncation_side="left", return_tensors="pd", padding=True, add_special_tokens=True, ) model.eval() model_kwargs["use_cache"] = True model_kwargs["max_new_tokens"] = self.max_new_tokens with static_mode_guard(): with tempfile.TemporaryDirectory() as tempdir: path = os.path.join(tempdir, "model") model.to_static( path, config=dict( use_top_p=False, ), ) model_path = os.path.join(tempdir, f"model{PADDLE_INFERENCE_MODEL_SUFFIX}") params_path = os.path.join(tempdir, f"model{PADDLE_INFERENCE_WEIGHTS_SUFFIX}") config = paddle.inference.Config(model_path, params_path) config.disable_gpu() config.disable_glog_info() predictor = paddle.inference.create_predictor(config) model_kwargs["top_k"] = 1 model_kwargs["max_new_tokens"] = self.max_new_tokens # create input for key in model_kwargs.keys(): if paddle.is_tensor(model_kwargs[key]): model_kwargs[key] = model_kwargs[key].numpy() else: model_kwargs[key] = np.array(model_kwargs[key]) input_handles = {} for name in predictor.get_input_names(): input_handles[name] = predictor.get_input_handle(name) input_handles[name].copy_from_cpu(model_kwargs[name]) predictor.run() output_names = predictor.get_output_names() output_handle = predictor.get_output_handle(output_names[0]) results = output_handle.copy_to_cpu() self.assertEqual(len(results.tolist()[0]), self.max_new_tokens) self.assertIsNotNone(results)