tests/causal_lm_tester.py

# Copyright 2025 HuggingFace Inc.
#
# 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 tempfile
from inspect import signature

import pytest
from parameterized import parameterized

from transformers import AutoModelForCausalLM, PreTrainedConfig, set_seed
from transformers.models.auto.auto_factory import getattribute_from_module
from transformers.testing_utils import (
    _COMMON_MODEL_NAMES_MAP,
    is_flaky,
    require_flash_attn,
    require_torch_accelerator,
    slow,
)

from .test_configuration_common import ConfigTester
from .test_modeling_common import (
    GenerationTesterMixin,
    ModelTesterMixin,
    ids_tensor,
    is_torch_available,
    require_torch,
    torch_device,
)
from .test_pipeline_mixin import PipelineTesterMixin
from .test_training_mixin import TrainingTesterMixin


if is_torch_available():
    import torch


class CausalLMModelTester:
    # If the model follows the standard naming conventions, only `base_model_class` needs to be set (the others are
    # inferred from available public classes).
    base_model_class = None
    # ⚠️ Don't set these unless the model does NOT follow the standard naming conventions ⚠️
    config_class = None
    causal_lm_class = None
    question_answering_class = None
    sequence_classification_class = None
    token_classification_class = None
    # These attributes are required after the initialization phase of the tester.
    _required_attributes = ("base_model_class", "config_class", "causal_lm_class")

    # Arguments that should be passed to the config class even if not in its signature
    forced_config_args = ["pad_token_id"]

    @classmethod
    def _verify_and_infer_model_attributes(cls):
        """
        Verifies that the required tester attributes are set correctly, and infers unset tester attributes.
        Intentionally nitpicks the tester class attributes, to prevent human errors.
        """
        # `base_model_class` is mandatory, and it must be a valid model class.
        base_model_class = getattr(cls, "base_model_class")
        if base_model_class is None or "PreTrainedModel" not in str(base_model_class.__mro__):
            raise ValueError(
                f"You have inherited from `CausalLMModelTester` but did not set the `base_model_class` "
                f"attribute to a valid model class. (It's set to `{base_model_class}`)"
            )

        # Infers other model classes from the base class name and available public classes, if the corresponding
        # attributes are not set explicitly. If they are set, they must be set to a valid class (config or model).
        model_name = base_model_class.__name__.replace("Model", "")
        base_class_module = ".".join(base_model_class.__module__.split(".")[:-1])
        for tester_attribute_name, model_class_termination in _COMMON_MODEL_NAMES_MAP.items():
            if getattr(cls, tester_attribute_name) is None:
                try:
                    model_class = getattribute_from_module(base_class_module, model_name + model_class_termination)
                    setattr(cls, tester_attribute_name, model_class)
                except ValueError:
                    pass
            else:
                if tester_attribute_name == "config_class":
                    if "PreTrainedConfig" not in str(getattr(cls, tester_attribute_name).__mro__):
                        raise ValueError(
                            f"You have inherited from `CausalLMModelTester` but did not set the "
                            f"`{tester_attribute_name}` attribute to a valid config class. (It's set to "
                            f"`{getattr(cls, tester_attribute_name)}`). If the config class follows a standard "
                            f"naming convention, you should unset `{tester_attribute_name}`."
                        )
                else:
                    if "PreTrainedModel" not in str(getattr(cls, tester_attribute_name).__mro__):
                        raise ValueError(
                            f"You have inherited from `CausalLMModelTester` but did not set the "
                            f"`{tester_attribute_name}` attribute to a valid model class. (It's set to "
                            f"`{getattr(cls, tester_attribute_name)}`). If the model class follows a standard "
                            f"naming convention, you should unset `{tester_attribute_name}`."
                        )

        # After inferring, if we don't have the basic classes set, we raise an error.
        for required_attribute in cls._required_attributes:
            if getattr(cls, required_attribute) is None:
                raise ValueError(
                    f"You have inherited from `CausalLMModelTester` but did not set the `{required_attribute}` "
                    "attribute. It can't be automatically inferred either -- this means it is not following a "
                    "standard naming convention. If this is intentional, please set the attribute explicitly."
                )

        # To prevent issues with typos, no other attributes can be set to a model class
        for instance_attribute_name, instance_attribute in cls.__dict__.items():
            if (
                (
                    instance_attribute_name not in _COMMON_MODEL_NAMES_MAP
                    and instance_attribute_name != "base_model_class"
                )
                and isinstance(instance_attribute, type)
                and "PreTrainedModel" in str(instance_attribute.__mro__)
            ):
                raise ValueError(
                    f"You have inherited from `CausalLMModelTester` but set an unexpected attribute to a model class "
                    f"(`{instance_attribute_name}` is set to `{instance_attribute}`). "
                    f"Only the following attributes can be set to model classes: {_COMMON_MODEL_NAMES_MAP.keys()}."
                )

    @property
    def all_model_classes(self):
        # Models that set `all_model_classes` in their `XXXModelTest` class must have a new class that doesn't fit
        # any of the common classes.
        return [
            model_class
            for model_class in (
                self.base_model_class,
                self.causal_lm_class,
                self.question_answering_class,
                self.sequence_classification_class,
                self.token_classification_class,
            )
            if model_class is not None
        ]

    @property
    def pipeline_model_mapping(self):
        # This is the default pipeline mapping.
        mapping = {
            "feature-extraction": self.base_model_class,
            "text-generation": self.causal_lm_class,
        }
        if self.question_answering_class is not None:
            mapping["question-answering"] = self.question_answering_class
        if self.sequence_classification_class is not None:
            mapping["text-classification"] = self.sequence_classification_class
        if self.token_classification_class is not None:
            mapping["token-classification"] = self.token_classification_class
        if self.sequence_classification_class is not None:
            mapping["zero-shot"] = self.sequence_classification_class
        return mapping

    def __init__(
        self,
        parent,
        batch_size=13,
        seq_length=7,
        is_training=True,
        use_input_mask=True,
        use_token_type_ids=False,
        use_labels=True,
        vocab_size=99,
        hidden_size=32,
        num_hidden_layers=2,
        num_attention_heads=2,
        num_key_value_heads=2,
        intermediate_size=32,
        hidden_act="gelu",
        hidden_dropout_prob=0.1,
        attention_probs_dropout_prob=0.1,
        max_position_embeddings=512,
        type_vocab_size=16,
        type_sequence_label_size=2,
        initializer_range=0.02,
        num_labels=3,
        num_choices=4,
        pad_token_id=0,
        bos_token_id=1,
        eos_token_id=2,
        is_decoder=False,
        scope=None,
        expert_interval=1,
        moe_layer_start_index=0,
        moe_intermediate_size=16,
        shared_expert_intermediate_size=36,
        shared_expert_gate=True,
        moe_num_shared_experts=2,
        num_experts_per_tok=2,
        num_experts=8,
        mamba_n_groups=1,
        mamba_n_heads=16,
        mamba_d_state=16,
        mamba_d_conv=4,
        mamba_expand=2,
        mamba_chunk_size=16,
    ):
        self._verify_and_infer_model_attributes()
        self.parent = parent
        self.batch_size = batch_size
        self.seq_length = seq_length
        self.is_training = is_training
        self.use_input_mask = use_input_mask
        self.use_token_type_ids = use_token_type_ids
        self.use_labels = use_labels
        self.vocab_size = vocab_size
        self.hidden_size = hidden_size
        self.num_hidden_layers = num_hidden_layers
        self.num_attention_heads = num_attention_heads
        self.num_key_value_heads = num_key_value_heads
        self.intermediate_size = intermediate_size
        self.hidden_act = hidden_act
        self.hidden_dropout_prob = hidden_dropout_prob
        self.attention_probs_dropout_prob = attention_probs_dropout_prob
        self.max_position_embeddings = max_position_embeddings
        self.type_vocab_size = type_vocab_size
        self.type_sequence_label_size = type_sequence_label_size
        self.initializer_range = initializer_range
        self.num_labels = num_labels
        self.num_choices = num_choices
        self.pad_token_id = pad_token_id
        self.bos_token_id = bos_token_id
        self.eos_token_id = eos_token_id
        self.scope = scope
        self.head_dim = self.hidden_size // self.num_attention_heads
        self.is_decoder = is_decoder
        self.expert_interval = expert_interval
        self.moe_layer_start_index = moe_layer_start_index
        self.moe_intermediate_size = moe_intermediate_size
        self.shared_expert_intermediate_size = shared_expert_intermediate_size
        self.shared_expert_gate = shared_expert_gate
        self.moe_num_shared_experts = moe_num_shared_experts
        self.num_experts_per_tok = num_experts_per_tok
        self.num_experts = num_experts
        self.mamba_n_groups = mamba_n_groups
        self.mamba_n_heads = mamba_n_heads
        self.mamba_d_state = mamba_d_state
        self.mamba_d_conv = mamba_d_conv
        self.mamba_expand = mamba_expand
        self.mamba_chunk_size = mamba_chunk_size
        self.tie_word_embeddings = False

    def prepare_config_and_inputs(self):
        input_ids = ids_tensor([self.batch_size, self.seq_length], self.vocab_size)

        input_mask = None
        if self.use_input_mask:
            input_mask = torch.tril(torch.ones_like(input_ids).to(torch_device))

        token_type_ids = None
        if self.use_token_type_ids:
            token_type_ids = ids_tensor([self.batch_size, self.seq_length], self.type_vocab_size)

        sequence_labels = None
        token_labels = None
        choice_labels = None
        if self.use_labels:
            sequence_labels = ids_tensor([self.batch_size], self.type_sequence_label_size)
            token_labels = ids_tensor([self.batch_size, self.seq_length], self.num_labels)
            choice_labels = ids_tensor([self.batch_size], self.num_choices)

        config = self.get_config()

        return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels

    @property
    def config_args(self):
        return list(signature(self.config_class.__init__).parameters.keys())

    def get_config(self):
        kwargs = {}
        model_name_to_common_name = {v: k for k, v in self.config_class.attribute_map.items()}
        for k in self.config_args + self.forced_config_args:
            if hasattr(self, k) and k != "self":
                kwargs[k] = getattr(self, k)
            elif k in model_name_to_common_name and hasattr(self, model_name_to_common_name[k]):
                kwargs[k] = getattr(self, model_name_to_common_name[k])
        return self.config_class(**kwargs)

    def create_and_check_model(
        self, config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels
    ):
        model = self.base_model_class(config=config)
        model.to(torch_device)
        model.eval()
        result = model(input_ids, attention_mask=input_mask)
        result = model(input_ids)
        self.parent.assertEqual(result.last_hidden_state.shape, (self.batch_size, self.seq_length, self.hidden_size))

    def prepare_config_and_inputs_for_common(self):
        config, input_ids, _, input_mask, _, _, _ = self.prepare_config_and_inputs()
        inputs_dict = {"input_ids": input_ids, "attention_mask": input_mask}
        return config, inputs_dict


@require_torch
class CausalLMModelTest(ModelTesterMixin, GenerationTesterMixin, PipelineTesterMixin, TrainingTesterMixin):
    model_tester_class = None
    all_model_classes = None
    pipeline_model_mapping = None

    def setUp(self):
        if self.model_tester_class is None:
            raise ValueError(
                "You have inherited from CausalLMModelTest but did not set the model_tester_class attribute."
            )
        self.model_tester = self.model_tester_class(self)
        self.config_tester = ConfigTester(self, config_class=self.model_tester.config_class)

        if self.pipeline_model_mapping is None:
            # If `all_model_classes` is not the default, maybe there are more pipeline mappings to be set.
            if self.all_model_classes is not None:
                raise ValueError(
                    "Testes that inherit from `CausalLMModelTest` and set `all_model_classes` must manually set "
                    "`pipeline_model_mapping`."
                )
            # Otherwise, we know the pipeline mapping is the default.
            else:
                self.pipeline_model_mapping = self.model_tester.pipeline_model_mapping

        if self.all_model_classes is None:
            self.all_model_classes = self.model_tester.all_model_classes

    def test_config(self):
        self.config_tester.run_common_tests()

    def test_model(self):
        config_and_inputs = self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*config_and_inputs)

    def test_sequence_classification_model(self):
        if self.model_tester.sequence_classification_class is None:
            self.skipTest("Model does not support sequence classification")
        config, input_dict = self.model_tester.prepare_config_and_inputs_for_common()
        config.num_labels = 3
        input_ids = input_dict["input_ids"]
        attention_mask = input_ids.ne(1).to(torch_device)
        sequence_labels = ids_tensor([self.model_tester.batch_size], self.model_tester.type_sequence_label_size)
        model = self.model_tester.sequence_classification_class(config)
        model.to(torch_device)
        model.eval()
        result = model(input_ids, attention_mask=attention_mask, labels=sequence_labels)
        self.assertEqual(result.logits.shape, (self.model_tester.batch_size, self.model_tester.num_labels))

    def test_sequence_classification_model_for_single_label(self):
        if self.model_tester.sequence_classification_class is None:
            self.skipTest("Model does not support sequence classification")
        config, input_dict = self.model_tester.prepare_config_and_inputs_for_common()
        config.num_labels = 3
        config.problem_type = "single_label_classification"
        input_ids = input_dict["input_ids"]
        attention_mask = input_ids.ne(1).to(torch_device)
        sequence_labels = ids_tensor([self.model_tester.batch_size], self.model_tester.type_sequence_label_size)
        model = self.model_tester.sequence_classification_class(config)
        model.to(torch_device)
        model.eval()
        result = model(input_ids, attention_mask=attention_mask, labels=sequence_labels)
        self.assertEqual(result.logits.shape, (self.model_tester.batch_size, self.model_tester.num_labels))

    def test_sequence_classification_model_for_multi_label(self):
        if self.model_tester.sequence_classification_class is None:
            self.skipTest("Model does not support sequence classification")
        config, input_dict = self.model_tester.prepare_config_and_inputs_for_common()
        config.num_labels = 3
        config.problem_type = "multi_label_classification"
        input_ids = input_dict["input_ids"]
        attention_mask = input_ids.ne(1).to(torch_device)
        sequence_labels = ids_tensor(
            [self.model_tester.batch_size, config.num_labels], self.model_tester.type_sequence_label_size
        ).to(torch.float)
        model = self.model_tester.sequence_classification_class(config)
        model.to(torch_device)
        model.eval()
        result = model(input_ids, attention_mask=attention_mask, labels=sequence_labels)
        self.assertEqual(result.logits.shape, (self.model_tester.batch_size, self.model_tester.num_labels))

    def test_token_classification_model(self):
        if self.model_tester.token_classification_class is None:
            self.skipTest("Model does not support token classification")
        config, input_dict = self.model_tester.prepare_config_and_inputs_for_common()
        config.num_labels = 3
        input_ids = input_dict["input_ids"]
        attention_mask = input_ids.ne(1).to(torch_device)
        token_labels = ids_tensor([self.model_tester.batch_size, self.model_tester.seq_length], config.num_labels)
        model = self.model_tester.token_classification_class(config=config)
        model.to(torch_device)
        model.eval()
        result = model(input_ids, attention_mask=attention_mask, labels=token_labels)
        self.assertEqual(
            result.logits.shape,
            (self.model_tester.batch_size, self.model_tester.seq_length, self.model_tester.num_labels),
        )

    def test_question_answering_model(self):
        if self.model_tester.question_answering_class is None:
            self.skipTest("Model does not support question answering")
        config, input_dict = self.model_tester.prepare_config_and_inputs_for_common()
        config.num_labels = 3

        input_ids = input_dict["input_ids"]
        attention_mask = input_ids.ne(1).to(torch_device)
        model = self.model_tester.question_answering_class(config=config)
        model.to(torch_device)
        model.eval()
        result = model(input_ids, attention_mask=attention_mask)
        self.assertEqual(
            result.start_logits.shape,
            (self.model_tester.batch_size, self.model_tester.seq_length),
        )
        self.assertEqual(
            result.end_logits.shape,
            (self.model_tester.batch_size, self.model_tester.seq_length),
        )

    @parameterized.expand([("linear",), ("dynamic",), ("yarn",)])
    def test_model_rope_scaling_from_config(self, scaling_type):
        """
        Tests that we can initialize a model with RoPE scaling in the config, that it can run a forward pass, and
        that a few basic model output properties are honored.
        """
        config, _ = self.model_tester.prepare_config_and_inputs_for_common()

        if not _config_supports_rope_scaling(config):
            self.skipTest("This model does not support RoPE scaling")

        partial_rotary_factor = config.rope_parameters.get("partial_rotary_factor", 1.0)
        short_input = ids_tensor([1, 10], config.vocab_size)
        long_input = ids_tensor([1, int(config.max_position_embeddings * 1.5)], config.vocab_size)

        set_seed(42)  # Fixed seed at init time so the two models get the same random weights
        _set_config_rope_params(
            config,
            {
                "rope_type": "default",
                "rope_theta": 10_000.0,
                "partial_rotary_factor": partial_rotary_factor,
                "original_max_position_embeddings": 16384,
            },
        )
        original_model = self.model_tester_class.base_model_class(config)
        original_model.to(torch_device)
        original_model.eval()
        original_short_output = original_model(short_input).last_hidden_state
        original_long_output = original_model(long_input).last_hidden_state

        set_seed(42)  # Fixed seed at init time so the two models get the same random weights
        _set_config_rope_params(
            config,
            {
                "rope_type": scaling_type,
                "factor": 10.0,
                "rope_theta": 10_000.0,
                "partial_rotary_factor": partial_rotary_factor,
            },
        )
        scaled_model = self.model_tester_class.base_model_class(config)
        scaled_model.to(torch_device)
        scaled_model.eval()
        scaled_short_output = scaled_model(short_input).last_hidden_state
        scaled_long_output = scaled_model(long_input).last_hidden_state

        # Dynamic scaling does not change the RoPE embeddings until it receives an input longer than the original
        # maximum sequence length, so the outputs for the short input should match.
        if scaling_type == "dynamic":
            torch.testing.assert_close(original_short_output, scaled_short_output, rtol=1e-5, atol=1e-5)
        else:
            self.assertFalse(torch.allclose(original_short_output, scaled_short_output, atol=1e-5))

        # The output should be different for long inputs
        self.assertFalse(torch.allclose(original_long_output, scaled_long_output, atol=1e-5))

    def test_model_rope_scaling_frequencies(self):
        """Tests the frequency properties of the different RoPE scaling types on the model RoPE layer."""
        config, _ = self.model_tester.prepare_config_and_inputs_for_common()

        if not _config_supports_rope_scaling(config):
            self.skipTest("This model does not support RoPE scaling")

        # Retrieves the RoPE layer class from the base model class. Uses `.named_modules()` to avoid hardcoding the
        # named location of the RoPE layer class.
        base_model = self.model_tester.base_model_class(config)
        possible_rope_attributes = [
            "pos_emb",
            "rotary_emb",  # most common case
            "global_rotary_emb",
            "local_rotary_emb",
        ]
        for name, module in base_model.named_modules():
            if any(potential_name in name for potential_name in possible_rope_attributes):
                rope_class = type(module)
                break

        scaling_factor = 10
        short_input_length = 10
        partial_rotary_factor = config.rope_parameters.get("partial_rotary_factor", 1.0)
        long_input_length = int(config.max_position_embeddings * 1.5)

        # Inputs
        x = torch.randn(
            1, dtype=torch.float32, device=torch_device
        )  # used exclusively to get the dtype and the device
        position_ids_short = torch.arange(short_input_length, dtype=torch.long, device=torch_device)
        position_ids_short = position_ids_short.unsqueeze(0)
        position_ids_long = torch.arange(long_input_length, dtype=torch.long, device=torch_device)
        position_ids_long = position_ids_long.unsqueeze(0)

        # Sanity check original RoPE
        _set_config_rope_params(
            config, {"rope_type": "default", "rope_theta": 10_000.0, "partial_rotary_factor": partial_rotary_factor}
        )
        original_rope = rope_class(config=config).to(torch_device)
        original_cos_short, original_sin_short = original_rope(x, position_ids_short)
        original_cos_long, original_sin_long = original_rope(x, position_ids_long)
        torch.testing.assert_close(original_cos_short, original_cos_long[:, :short_input_length, :])
        torch.testing.assert_close(original_sin_short, original_sin_long[:, :short_input_length, :])

        # Sanity check linear RoPE scaling
        # New position "x" should match original position with index "x/scaling_factor"
        _set_config_rope_params(
            config,
            {
                "rope_type": "linear",
                "factor": scaling_factor,
                "rope_theta": 10_000.0,
                "partial_rotary_factor": partial_rotary_factor,
            },
        )
        linear_scaling_rope = rope_class(config=config).to(torch_device)
        linear_cos_short, linear_sin_short = linear_scaling_rope(x, position_ids_short)
        linear_cos_long, linear_sin_long = linear_scaling_rope(x, position_ids_long)
        torch.testing.assert_close(linear_cos_short, linear_cos_long[:, :short_input_length, :])
        torch.testing.assert_close(linear_sin_short, linear_sin_long[:, :short_input_length, :])
        for new_position in range(0, long_input_length, scaling_factor):
            original_position = int(new_position // scaling_factor)
            torch.testing.assert_close(linear_cos_long[:, new_position, :], original_cos_long[:, original_position, :])
            torch.testing.assert_close(linear_sin_long[:, new_position, :], original_sin_long[:, original_position, :])

        # Sanity check Dynamic NTK RoPE scaling
        # Scaling should only be observed after a long input is fed. We can observe that the frequencies increase
        # with scaling_factor (or that `inv_freq` decreases)
        _set_config_rope_params(
            config,
            {
                "rope_type": "dynamic",
                "factor": scaling_factor,
                "rope_theta": 10_000.0,
                "partial_rotary_factor": partial_rotary_factor,
            },
        )
        ntk_scaling_rope = rope_class(config=config).to(torch_device)
        ntk_cos_short, ntk_sin_short = ntk_scaling_rope(x, position_ids_short)
        ntk_cos_long, ntk_sin_long = ntk_scaling_rope(x, position_ids_long)
        torch.testing.assert_close(ntk_cos_short, original_cos_short)
        torch.testing.assert_close(ntk_sin_short, original_sin_short)
        with self.assertRaises(AssertionError):
            torch.testing.assert_close(ntk_cos_long, original_cos_long)
        with self.assertRaises(AssertionError):
            torch.testing.assert_close(ntk_sin_long, original_sin_long)
        self.assertTrue((ntk_scaling_rope.inv_freq <= original_rope.inv_freq).all())

        # Sanity check Yarn RoPE scaling
        # Scaling should be over the entire input
        _set_config_rope_params(
            config,
            {
                "rope_type": "yarn",
                "factor": scaling_factor,
                "rope_theta": 10_000.0,
                "partial_rotary_factor": partial_rotary_factor,
            },
        )
        yarn_scaling_rope = rope_class(config=config).to(torch_device)
        yarn_cos_short, yarn_sin_short = yarn_scaling_rope(x, position_ids_short)
        yarn_cos_long, yarn_sin_long = yarn_scaling_rope(x, position_ids_long)
        torch.testing.assert_close(yarn_cos_short, yarn_cos_long[:, :short_input_length, :])
        torch.testing.assert_close(yarn_sin_short, yarn_sin_long[:, :short_input_length, :])
        with self.assertRaises(AssertionError):
            torch.testing.assert_close(yarn_cos_short, original_cos_short)
        with self.assertRaises(AssertionError):
            torch.testing.assert_close(yarn_sin_short, original_sin_short)
        with self.assertRaises(AssertionError):
            torch.testing.assert_close(yarn_cos_long, original_cos_long)
        with self.assertRaises(AssertionError):
            torch.testing.assert_close(yarn_sin_long, original_sin_long)

    @require_flash_attn
    @require_torch_accelerator
    @pytest.mark.flash_attn_test
    @is_flaky()
    @slow
    def test_flash_attn_2_equivalence(self):
        for model_class in self.all_model_classes:
            if not model_class._supports_flash_attn:
                self.skipTest(reason="Model does not support Flash Attention 2")

            config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common()
            model = model_class(config)

            with tempfile.TemporaryDirectory() as tmpdirname:
                model.save_pretrained(tmpdirname)
                model_fa = model_class.from_pretrained(
                    tmpdirname, dtype=torch.bfloat16, attn_implementation="flash_attention_2"
                )
                model_fa.to(torch_device)

                model = model_class.from_pretrained(tmpdirname, dtype=torch.bfloat16, attn_implementation="eager")
                model.to(torch_device)

                dummy_input = inputs_dict[model_class.main_input_name]
                dummy_input = dummy_input.to(torch_device)
                outputs = model(dummy_input, output_hidden_states=True)
                outputs_fa = model_fa(dummy_input, output_hidden_states=True)

                logits = outputs.hidden_states[-1]
                logits_fa = outputs_fa.hidden_states[-1]
                torch.testing.assert_close(logits_fa, logits, atol=3e-2, rtol=3e-2)

    def test_causal_lm_can_accept_training_kwargs(self):
        if not getattr(self.model_tester, "is_training", False):
            self.skipTest(reason="ModelTester is not configured to run training tests")

        config, inputs_dict = self.model_tester.prepare_config_and_inputs_for_common()

        with tempfile.TemporaryDirectory() as tmpdir:
            with torch.device(torch_device):
                model_eager = AutoModelForCausalLM.from_config(config, dtype=torch.float32)

            model_eager.save_pretrained(tmpdir)
            model = AutoModelForCausalLM.from_pretrained(tmpdir, dtype=torch.float32, device_map=torch_device)
            inputs_dict["num_items_in_batch"] = torch.tensor(inputs_dict["input_ids"].shape[0])
            inputs_dict["labels"] = inputs_dict["input_ids"]
            _ = model(**inputs_dict, return_dict=False)


def _config_supports_rope_scaling(config: PreTrainedConfig) -> bool:
    """Returns whether a certain model config supports RoPE scaling parameterization."""
    # Has rope_scaling -> model was designed with rope scaling in mind
    # Has rope_theta (and no rope_scaling) -> probably an older model, but should support rope scaling as well
    main_config_has_rope = hasattr(config, "rope_parameters")
    sub_config_has_rope = any(
        hasattr(getattr(config, sub_config), "rope_parameters") for sub_config in config.sub_configs.keys()
    )
    return main_config_has_rope or sub_config_has_rope


def _set_config_rope_params(config: PreTrainedConfig, rope_params: dict) -> bool:
    """Recursively sets RoPE parameters on configs and subconfigs, by duplicating the same RoPE values."""
    config.rope_parameters = getattr(config, "rope_parameters", {}) or {}
    config.rope_parameters.update(rope_params)

    if any(name in config.__class__.__name__.lower() for name in ["gemma3", "modernbert"]):
        config.rope_parameters = {layer_type: config.rope_parameters.copy() for layer_type in config.layer_types}

    for sub_config in config.sub_configs.keys():
        _set_config_rope_params(getattr(config, sub_config), rope_params)
    return config