"""Patched MPT model for mt_dao evaluation."""
# Adapted from https://github.com/vllm-project/vllm/blob/53ea6ad8304f5d16bab97527ea1f4a19cfb25588/vllm/model_executor/models/mpt.py
# with mutually exclusive ALiBi or RoPE support and no kv_caches/attn_metadata.

import math
from collections.abc import Iterable
from dataclasses import asdict
from typing import Any

import torch
import torch.nn.functional as F  # noqa: N812
from torch import nn
from vllm.attention import Attention
from vllm.compilation.decorators import support_torch_compile
from vllm.config import CacheConfig, VllmConfig
from vllm.distributed import (
    get_pp_group,
    get_tensor_model_parallel_rank,
    get_tensor_model_parallel_world_size,
)
from vllm.model_executor.layers.activation import get_act_fn
from vllm.model_executor.layers.linear import (
    ColumnParallelLinear,
    QKVParallelLinear,
    ReplicatedLinear,
    RowParallelLinear,
)
from vllm.model_executor.layers.logits_processor import LogitsProcessor
from vllm.model_executor.layers.quantization import QuantizationConfig
from vllm.model_executor.layers.rotary_embedding import get_rope
from vllm.model_executor.layers.vocab_parallel_embedding import (
    VocabParallelEmbedding,
)
from vllm.model_executor.model_loader.weight_utils import default_weight_loader
from vllm.model_executor.models.interfaces import SupportsPP
from vllm.model_executor.models.utils import (
    AutoWeightsLoader,
    is_pp_missing_parameter,
    make_empty_intermediate_tensors_factory,
    make_layers,
    maybe_prefix,
)
from vllm.model_executor.sampling_metadata import SamplingMetadata
from vllm.sequence import IntermediateTensors
from vllm.transformers_utils.configs.mpt import MPTConfig

from mt_dao.models.deptoe.mup_config import MuPConfig


def _get_alibi_slopes(
    total_num_heads: int,
    alibi_bias_max: int,
) -> torch.Tensor:
    next_power_of_2 = 2 ** math.ceil(math.log2(total_num_heads))
    m = torch.arange(1, next_power_of_2 + 1, dtype=torch.float32)
    m = m.mul(alibi_bias_max / next_power_of_2)
    slopes = 1.0 / torch.pow(2, m)
    if next_power_of_2 != total_num_heads:
        slopes = torch.concat([slopes[1::2], slopes[::2]])[:total_num_heads]
    return slopes


class MPTMuPConfig(MPTConfig):
    """Configuration for MPT with MuP (Maximal Update Parametrisation)."""

    # pylint: disable=dangerous-default-value

    def __init__(
        self,
        mup_config: dict | None = None,
        *,
        use_peri_norm: bool = False,
        use_embedding_norm: bool = False,
        **kwargs: Any,  # noqa: ANN401
    ) -> None:
        """Initialize MPTMuPConfig with MuP configuration and other parameters.

        Parameters
        ----------
        mup_config : dict | None
            Configuration for MuP. If None, defaults to an empty MuPConfig.
        use_peri_norm : bool
            Whether to use periodic normalization in the model.
        use_embedding_norm : bool
            Whether to use normalization on the input embeddings.
        **kwargs : Any
            Additional keyword arguments for MPTConfig.

        """
        self.use_peri_norm = use_peri_norm
        self.use_embedding_norm = use_embedding_norm
        obj_mup_config = (
            MuPConfig(**mup_config) if mup_config is not None else MuPConfig()
        )

        self.mup_config = asdict(
            obj_mup_config,
        )
        super().__init__(**kwargs)
        self._validate_config()


class MPTAttention(nn.Module):
    """MPTAttention module implementing either ALiBi or RoPE attention mechanisms."""

    def __init__(  # noqa: PLR0915
        self,
        config: MPTMuPConfig,
        cache_config: CacheConfig | None = None,
        quant_config: QuantizationConfig | None = None,
        prefix: str = "",
    ) -> None:
        """Initialize MPTAttention with configuration and quantization settings.

        Parameters
        ----------
        config : MPTMuPConfig
            Configuration for MPT with MuP.
        cache_config : CacheConfig | None
            Configuration for caching, if applicable.
        quant_config : QuantizationConfig | None
            Configuration for quantization, if applicable.
        prefix : str
            Prefix for naming the attention module, useful for logging or debugging.

        Raises
        ------
        ValueError
            If both ALiBi and RoPE are enabled, or if neither is enabled.

        """
        super().__init__()
        self.d_model = config.d_model
        self.total_num_heads = config.n_heads
        self.head_dim = self.d_model // self.total_num_heads
        self.clip_qkv = config.attn_config["clip_qkv"]
        self.qk_ln = config.attn_config["qk_ln"]

        # ALiBi config
        self.alibi_bias_max = config.attn_config["alibi_bias_max"]

        # RoPE config
        rope_theta = config.attn_config.get("rope_theta", 10000.0)
        rope_scaling = config.attn_config.get("rope_scaling", None)
        partial_rotary_factor = config.attn_config.get("partial_rotary_factor", 1.0)

        if "kv_n_heads" in config.attn_config:
            self.total_num_kv_heads = config.attn_config["kv_n_heads"]
        else:
            self.total_num_kv_heads = self.total_num_heads

        assert not config.attn_config["prefix_lm"]

        # pylint: disable=invalid-name
        self.Wqkv = QKVParallelLinear(
            self.d_model,
            self.d_model // self.total_num_heads,
            self.total_num_heads,
            self.total_num_kv_heads,
            bias=not config.no_bias,
            quant_config=quant_config,
        )
        if self.qk_ln:
            self.q_ln = nn.LayerNorm(self.d_model)
            self.k_ln = nn.LayerNorm(self.d_model)
        self.out_proj = RowParallelLinear(
            self.d_model,
            self.d_model,
            bias=not config.no_bias,
            quant_config=quant_config,
        )

        tp_world_size = get_tensor_model_parallel_world_size()
        assert self.total_num_heads % tp_world_size == 0
        self.num_heads = self.total_num_heads // tp_world_size

        if self.total_num_kv_heads >= tp_world_size:
            # Number of KV heads is greater than TP size, so we partition
            # the KV heads across multiple tensor parallel GPUs.
            assert self.total_num_kv_heads % tp_world_size == 0
        else:
            # Number of KV heads is less than TP size, so we replicate
            # the KV heads across multiple tensor parallel GPUs.
            assert tp_world_size % self.total_num_kv_heads == 0
        self.num_kv_heads = max(1, self.total_num_kv_heads // tp_world_size)
        self.q_size = self.num_heads * self.head_dim
        self.kv_size = self.num_kv_heads * self.head_dim
        # Create the alibi slopes and slice them.
        tp_rank = get_tensor_model_parallel_rank()
        head_start = tp_rank * self.num_heads
        head_end = (tp_rank + 1) * self.num_heads

        # ALiBi or RoPE flags:
        self.alibi = config.attn_config.get("alibi", False)
        self.rope = config.attn_config.get("rope", False)
        if self.alibi and self.rope:
            msg = "Cannot enable both ALiBi and RoPE at once."
            raise ValueError(msg)
        if not self.alibi and not self.rope:
            msg = "Must enable either ALiBi or RoPE."
            raise ValueError(msg)

        # Possibly build ALiBi slopes
        alibi_slopes = None
        if self.alibi:
            full_slopes = _get_alibi_slopes(self.total_num_heads, self.alibi_bias_max)
            head_start = tp_rank * self.num_heads
            head_end = (tp_rank + 1) * self.num_heads
            alibi_slopes = full_slopes[head_start:head_end].tolist()

        # Possibly build RoPE
        self.rotary_emb = None
        if self.rope:
            rope_dim = int(self.head_dim * partial_rotary_factor)
            self.rotary_emb = get_rope(
                head_size=self.head_dim,
                rotary_dim=rope_dim,
                max_position=config.max_seq_len,
                base=rope_theta,
                rope_scaling=rope_scaling,
                # MPT uses "neox"-style ordering of heads
                is_neox_style=True,
            )

        self.head_dim = self.d_model // self.total_num_heads
        scaling = self.head_dim**-0.5

        self.attn = Attention(
            self.num_heads,
            self.head_dim,
            scaling,
            alibi_slopes=alibi_slopes,
            num_kv_heads=self.num_kv_heads,
            cache_config=cache_config,
            quant_config=quant_config,
            prefix=f"{prefix}.attn",
        )

    def forward(
        self,
        position_ids: torch.Tensor,
        hidden_states: torch.Tensor,
    ) -> torch.Tensor:
        """Forward pass with either ALiBi or RoPE.

        The snippet doesn't handle kv_caches,
        so we just pass (q, k, v) into Attention directly.

        Parameters
        ----------
        position_ids : torch.Tensor
            Position IDs for the input sequence, used for RoPE.
        hidden_states : torch.Tensor
            Input hidden states to the attention layer.

        Returns
        -------
        torch.Tensor
            Output hidden states after applying attention.

        """
        qkv, _ = self.Wqkv(hidden_states)

        if self.clip_qkv is not None:
            qkv.clamp_(min=-self.clip_qkv, max=self.clip_qkv)

        q, k, v = qkv.split([self.q_size, self.kv_size, self.kv_size], dim=-1)

        if self.qk_ln:
            q = self.q_ln(q)
            k = self.k_ln(k)

        # If RoPE is enabled, apply it to (q, k)
        if self.rotary_emb is not None:
            q, k = self.rotary_emb(position_ids, q, k)

        attn_output = self.attn(q, k, v)
        output, _ = self.out_proj(attn_output)
        return output


class MPTMLP(nn.Module):
    """MPTMLP module implementing the ffn with up and down projections."""

    def __init__(
        self,
        config: MPTMuPConfig,
        quant_config: QuantizationConfig | None = None,
    ) -> None:
        """Initialize MPTMLP with configuration and quantization settings.

        Parameters
        ----------
        config : MPTMuPConfig
            Configuration for MPT with MuP.
        quant_config : QuantizationConfig | None
            Configuration for quantization

        """
        super().__init__()
        hidden_size = config.d_model
        expansion_ratio = config.expansion_ratio
        intermediate_size = expansion_ratio * hidden_size
        self.up_proj = ColumnParallelLinear(
            hidden_size,
            intermediate_size,
            bias=not config.no_bias,
            quant_config=quant_config,
        )

        # NOTE: Not all of these are available in VLLm, silu is
        self.act = get_act_fn(
            config.ffn_config.get("ffn_act_fn", {"name": "silu"}).get("name", "silu"),
        )
        self.down_proj = RowParallelLinear(
            intermediate_size,
            hidden_size,
            bias=not config.no_bias,
            quant_config=quant_config,
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """Forward pass through the MPTMLP.

        Parameters
        ----------
        x : torch.Tensor
            Input tensor to the MPTMLP.

        Returns
        -------
        torch.Tensor
            Output tensor after applying the up projection, activation, and down proj.

        """
        x, _ = self.up_proj(x)
        x = self.act(x)
        x, _ = self.down_proj(x)
        return x


class SigmaMoe(nn.Module):
    """A VLLM-compatible implementation of the original SigmaMoE.

    (MLP architecture), replacing the FusedMoE-based implementation.
    """

    def __init__(
        self,
        config: MPTMuPConfig,
        quant_config: QuantizationConfig | None = None,
    ) -> None:
        """Initialize SigmaMoe with configuration and quantization settings.

        Parameters
        ----------
        config : MPTMuPConfig
            Configuration for MPT with MuP.
        quant_config : QuantizationConfig | None
            Configuration for quantization, if applicable.

        """
        super().__init__()

        self.d_model = config.d_model
        self.n_experts = config.ffn_config["ff_n_experts"]
        self.n_active_experts = config.ffn_config["ff_n_active_experts"]
        self.expert_size = config.ffn_config["ff_expert_size"]

        # --- VLLM Layer for Router ---
        # We use ReplicatedLinear to leverage VLLM's loading/quantization utils.
        self.expert_sel = ReplicatedLinear(
            self.d_model,
            self.n_experts,
            bias=False,
            quant_config=quant_config,
        )

        # Activation function
        self.activation = get_act_fn(
            config.ffn_config.get("ffn_act_fn", {"name": "silu"}).get("name", "silu"),
        )

        # --- Standard Parameters for Experts (MLP Architecture) ---
        # We use standard nn.Parameters as FusedMoE is incompatible.

        # up_proj shape: (n_experts, expert_size, d_model)
        self.up_proj = torch.nn.Parameter(
            torch.empty(self.n_experts, self.expert_size, self.d_model),
        )
        # down_proj shape: (n_experts, expert_size, d_model)
        self.down_proj = torch.nn.Parameter(
            torch.empty(self.n_experts, self.expert_size, self.d_model),
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """Forward pass through the SigmaMoe.

        Handles both the dense case (n_experts == 1) + the sparse case (n_experts > 1).

        Parameters
        ----------
        x : torch.Tensor
            Input tensor to the SigmaMoe.

        Returns
        -------
        torch.Tensor
            Output tensor after applying the expert selection and MLP operations.

        """
        if self.n_active_experts == 1 and self.n_experts == 1:
            return self._forward_simplified(x)

        output_shape = x.size()

        # 1. Router
        # router_logits shape: (batch_size, seq_length, n_experts)
        router_logits, _ = self.expert_sel(x)

        # 2. Sigmoid Gating and Top-K
        sel = F.sigmoid(router_logits)
        sel_val, sel_index = sel.topk(self.n_active_experts, dim=-1, sorted=False)

        # 3. Prepare for Vectorized Computation
        x_flat = x.view(-1, self.d_model)  # Shape: [N_tokens, d_model]
        n_tokens = x_flat.shape[0]

        # Create the full routing weight matrix [N_tokens, n_experts]
        # We use the flattened shape for the weights to match x_flat
        sel_weights = torch.zeros(
            n_tokens,
            self.n_experts,
            dtype=sel_val.dtype,
            device=sel_val.device,
        )

        # Flatten indices and values for scattering
        sel_index_flat = sel_index.view(n_tokens, self.n_active_experts)
        sel_val_flat = sel_val.view(n_tokens, self.n_active_experts)

        # Scatter the top-k values into the full weight matrix
        sel_weights.scatter_(1, sel_index_flat, sel_val_flat)

        # 4. Vectorized Expert Computation (MLP)

        # Step 4.1: Up-projection
        # x_flat: [N, d_model], up_proj: [n_experts, expert_size, d_model]
        # Output: [N, n_experts, expert_size]  # noqa: ERA001
        # Einsum: "nd,esd->nes"  # noqa: ERA001
        up_outputs = torch.einsum("nd,esd->nes", x_flat, self.up_proj)

        # Step 4.2: Activation
        up_outputs = self.activation(up_outputs)

        # Step 4.3: Down-projection
        # Input: [N, n_experts, expert_size], down_proj:
        #             [n_experts, expert_size, d_model]  # noqa: ERA001
        # Output: [N, n_experts, d_model]  # noqa: ERA001
        # Einsum: "nes,esv->nev" (v=d_model)  # noqa: ERA001
        expert_outputs = torch.einsum("nes,esv->nev", up_outputs, self.down_proj)

        # Step 4.4: Apply selection weights and Sum
        # expert_outputs: [N, n_experts, d_model], sel_weights: [N, n_experts]
        # Output: [N, d_model]  # noqa: ERA001
        # Einsum: "ned,ne->nd"  # noqa: ERA001
        y = torch.einsum("ned,ne->nd", expert_outputs, sel_weights)

        return y.view(output_shape)

    def _forward_simplified(self, x: torch.Tensor) -> torch.Tensor:
        # Handles the dense case (n_experts == 1)
        up_weight = self.up_proj.view(self.expert_size, self.d_model)
        # Transpose down_proj for F.linear: (d_model, expert_size)
        down_weight = self.down_proj.view(self.expert_size, self.d_model).t()

        x_up_proj = self.activation(F.linear(x, up_weight, None))
        x_down_proj = F.linear(x_up_proj, down_weight, None)
        return x_down_proj.view(*x.shape)


class MPTBlock(nn.Module):
    """MPTBlock implementing a single transformer block."""

    def __init__(
        self,
        config: MPTMuPConfig,
        cache_config: CacheConfig | None = None,
        quant_config: QuantizationConfig | None = None,
        prefix: str = "",
    ) -> None:
        """Initialize MPTBlock with configuration, cache, and quantization settings.

        Enables MuP/CompleteP.

        Parameters
        ----------
        config : MPTMuPConfig
            Configuration for MPT with MuP.
        cache_config : CacheConfig | None
            Configuration for caching, if applicable.
        quant_config : QuantizationConfig | None
            Configuration for quantization, if applicable.
        prefix : str
            Prefix for naming the block, useful for logging or debugging.

        """
        super().__init__()
        hidden_size = config.d_model
        self.norm_1 = nn.LayerNorm(hidden_size)
        self.attn = MPTAttention(
            config,
            cache_config,
            quant_config,
            prefix=f"{prefix}.attn",
        )
        self.norm_2 = nn.LayerNorm(hidden_size)
        self.ffn = (
            SigmaMoe(
                config,
                quant_config,
            )
            if config.ffn_config["ffn_type"] == "sigma_moe"
            else MPTMLP(
                config,
                quant_config,
            )
        )
        self.mup_config = MuPConfig(
            **config.mup_config,
        )
        self.residual_scaling = (
            1.0
            / (
                self.mup_config.completep_depth_multiplier
                ** self.mup_config.completep_depth_alpha_exp
            )
            if self.mup_config.completep_depth_alpha_enabled
            else 1.0
        )
        self.post_attn_norm = None
        self.post_ffn_norm = None
        if config.use_peri_norm:
            self.post_attn_norm = nn.LayerNorm(hidden_size)
            self.post_ffn_norm = nn.LayerNorm(hidden_size)

    def forward(
        self,
        position_ids: torch.Tensor,
        hidden_states: torch.Tensor,
    ) -> torch.Tensor:
        """Forward pass through the MPTBlock.

        Parameters
        ----------
        position_ids : torch.Tensor
            Position IDs for the input sequence, used for RoPE.
        hidden_states : torch.Tensor
            Input hidden states to the MPTBlock.

        Returns
        -------
        torch.Tensor
            Output hidden states after applying attention and feed-forward network.

        """
        x = self.norm_1(hidden_states)
        x = self.attn(
            position_ids=position_ids,
            hidden_states=x,
        )
        x = self.post_attn_norm(x) if self.post_attn_norm else x
        hidden_states = hidden_states + x * self.residual_scaling  # noqa: PLR6104
        x = self.norm_2(hidden_states)
        x = self.ffn(x)
        x = self.post_ffn_norm(x) if self.post_ffn_norm else x
        return hidden_states + x * self.residual_scaling


@support_torch_compile
class MPTModel(nn.Module):
    """MptModel implementing the MPT transformer architecture with MuP.

    Parameters
    ----------
    vllm_config : VllmConfig
        Configuration for the MPT model, including model, cache, and quant settings.
    prefix : str, optional
        Prefix for naming the model components, useful for logging or debugging.

    Attributes
    ----------
    use_embedding_norm : bool
        Whether to use normalization on the input embeddings.
    input_alpha : float
        Input scaling factor for the MuP model.

    """

    def __init__(self, *, vllm_config: VllmConfig, prefix: str = "") -> None:
        """Initialize MPTModel with configuration and prefix.

        Parameters
        ----------
        vllm_config : VllmConfig
            Configuration for the MPT model, including model, cache, and quant settings.
        prefix : str, optional
            Prefix for naming the model components, useful for logging or debugging.

        """
        super().__init__()

        config = MPTMuPConfig(**vllm_config.model_config.hf_config.__dict__)
        cache_config = vllm_config.cache_config
        quant_config = vllm_config.quant_config
        self.mup_config = MuPConfig(
            **config.mup_config,
        )

        assert config.embedding_fraction == 1.0
        assert config.norm_type == "low_precision_layernorm"

        self.use_embedding_norm: bool = config.use_embedding_norm
        self.input_alpha: float = self.mup_config.mup_input_alpha
        self.output_alpha: float = self.mup_config.mup_output_alpha

        self.wte = VocabParallelEmbedding(
            config.vocab_size,
            config.d_model,
        )

        self.embed_norm = None
        if self.use_embedding_norm:
            self.embed_norm = nn.LayerNorm(config.d_model)

        self.start_layer, self.end_layer, self.blocks = make_layers(
            config.n_layers,
            lambda prefix: MPTBlock(config, cache_config, quant_config, prefix=prefix),  # pyright: ignore[reportArgumentType]
            prefix=f"{prefix}.blocks",
        )
        self.norm_f = nn.LayerNorm(config.d_model)

        if config.no_bias:
            for module in self.modules():
                if hasattr(module, "bias") and isinstance(module.bias, nn.Parameter):
                    # Remove the bias term in Linear and LayerNorm.
                    module.register_parameter("bias", None)
        self.make_empty_intermediate_tensors = make_empty_intermediate_tensors_factory(
            ["hidden_states"],
            config.d_model,
        )

    def get_input_embeddings(self, input_ids: torch.Tensor) -> torch.Tensor:
        """Get input embeddings for the given input IDs.

        Parameters
        ----------
        input_ids : torch.Tensor
            Input tensor containing token IDs.

        Returns
        -------
        torch.Tensor
            Output tensor containing the input embeddings.

        """
        x = self.wte(input_ids)
        if self.embed_norm:
            x = self.embed_norm(x) * self.input_alpha
        return x

    def forward(
        self,
        input_ids: torch.Tensor,
        position_ids: torch.Tensor,
        intermediate_tensors: IntermediateTensors | None,
        inputs_embeds: torch.Tensor | None = None,
    ) -> torch.Tensor | IntermediateTensors:
        """Forward pass through the MPTModel.

        Parameters
        ----------
        input_ids : torch.Tensor
            Input tensor containing token IDs.
        position_ids : torch.Tensor
            Position IDs for the input sequence, used for RoPE.
        intermediate_tensors : IntermediateTensors | None
            Intermediate tensors for storing hidden states, if applicable.
        inputs_embeds : torch.Tensor | None
            Input embeddings.

        Returns
        -------
        torch.Tensor | IntermediateTensors
            Output hidden states after applying the transformer blocks.

        """
        if get_pp_group().is_first_rank:
            if inputs_embeds is not None:
                hidden_states = inputs_embeds
            else:
                hidden_states = self.get_input_embeddings(input_ids)
        else:
            assert intermediate_tensors is not None
            hidden_states = intermediate_tensors["hidden_states"]

        for block in self.blocks[self.start_layer : self.end_layer]:
            hidden_states = block(position_ids, hidden_states)
        if not get_pp_group().is_last_rank:
            return IntermediateTensors({"hidden_states": hidden_states})
        return self.norm_f(hidden_states) * self.output_alpha

    def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
        """Load weights into the MPTModel.

        Parameters
        ----------
        weights : Iterable[tuple[str, torch.Tensor]]
            Iterable of tuples containing parameter names.

        Returns
        -------
        set[str]
            Set of parameter names that were successfully loaded.

        """
        params_dict = dict(self.named_parameters(remove_duplicate=False))
        loaded_params: set[str] = set()
        for saved_name, loaded_weight in weights:
            # Skip loading extra bias for GPTQ models.
            name = saved_name
            if name.endswith(".bias") and name not in params_dict:
                continue
            if is_pp_missing_parameter(name, self):
                continue
            if "wte.norm" in name:
                name = name.replace("wte.norm", "embed_norm")
            # log which parameter we have loaded
            # Try to find the parameter in the model's state_dict.
            # Handle the case where the checkpoint name needs a ".weight" suffix.
            if name not in params_dict:
                # This handles layers like ReplicatedLinear, ColumnParallelLinear etc.,
                # where the checkpoint key might be "layer_name" but the
                # parameter name in the model is "layer_name.weight".
                name = f"{name}.weight"
            param = params_dict[name]
            weight_loader = getattr(param, "weight_loader", default_weight_loader)
            weight_loader(param, loaded_weight)
            loaded_params.add(name)
        return loaded_params


class MPTForCausalLM(nn.Module, SupportsPP):
    """MPTForCausalLM implementing the MPT model for causal language modeling."""

    def __init__(self, *, vllm_config: VllmConfig, prefix: str = "") -> None:
        """Initialize MPTForCausalLM with configuration and prefix.

        Parameters.
        ----------
        vllm_config : VllmConfig
            Configuration for the MPT model, including model, cache, and quant settings.
        prefix : str, optional
            Prefix for naming the model components, useful for logging or debugging.
        """
        super().__init__()
        config = vllm_config.model_config.hf_config
        quant_config = vllm_config.quant_config
        self.config = config
        assert config.tie_word_embeddings
        self.quant_config = quant_config

        self.transformer = MPTModel(
            vllm_config=vllm_config,
            prefix=maybe_prefix(prefix, "transformer"),
        )
        self.lm_head = self.transformer.wte
        self.logits_processor = LogitsProcessor(config.vocab_size)
        self.make_empty_intermediate_tensors = (  # type: ignore[method-assign]
            self.transformer.make_empty_intermediate_tensors
        )

    def get_input_embeddings(self, input_ids: torch.Tensor) -> torch.Tensor:
        """Get input embeddings for the given input IDs.

        Parameters
        ----------
        input_ids : torch.Tensor
            Input tensor containing token IDs.

        Returns
        -------
        torch.Tensor
            Output tensor containing the input embeddings.

        """
        return self.transformer.get_input_embeddings(input_ids)

    def forward(  # type: ignore[reportIncompatibleMethodOverride,override]
        self,
        input_ids: torch.Tensor,
        positions: torch.Tensor,
        intermediate_tensors: IntermediateTensors | None = None,
        inputs_embeds: torch.Tensor | None = None,
    ) -> torch.Tensor | IntermediateTensors:
        """Forward pass through the MPTForCausalLM.

        Parameters
        ----------
        input_ids : torch.Tensor
            Input tensor containing token IDs.
        positions : torch.Tensor
            Position IDs for the input sequence, used for RoPE.
        intermediate_tensors : IntermediateTensors | None
            Intermediate tensors for storing hidden states, if applicable.
        inputs_embeds : torch.Tensor | None
            Input embeddings.

        Returns
        -------
        torch.Tensor | IntermediateTensors
            Output hidden states after applying the transformer blocks.

        """
        return self.transformer(
            input_ids,
            positions,
            intermediate_tensors,
            inputs_embeds,
        )

    def compute_logits(
        self,
        hidden_states: torch.Tensor,
        sampling_metadata: SamplingMetadata,
    ) -> torch.Tensor | None:
        """Compute logits from the hidden states.

        Parameters
        ----------
        hidden_states : torch.Tensor
            Hidden states from the transformer model.
        sampling_metadata : SamplingMetadata
            Metadata for sampling, including temperature, top_k, etc.

        Returns
        -------
        torch.Tensor | None
            Computed logits for the next token prediction, or None if not applicable.

        """
        return self.logits_processor(self.lm_head, hidden_states, sampling_metadata)

    def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
        """Load weights into the MPTForCausalLM.

        Parameters
        ----------
        weights : Iterable[tuple[str, torch.Tensor]]
            Iterable of tuples containing parameter names+weights.

        Returns
        -------
        set[str]
            Set of parameter names that were successfully loaded.

        """
        loader = AutoWeightsLoader(self)
        return loader.load_weights(weights)
