Skip to content

vllm.models.deepseek_v4.amd.model

Classes:

DeepseekV4ForCausalLM

Bases: Module, SupportsPP, SupportsEagle3

Methods:

Source code in vllm/models/deepseek_v4/amd/model.py 
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
class DeepseekV4ForCausalLM(nn.Module, SupportsPP, SupportsEagle3):
    model_cls = DeepseekV4Model

    # Default mapper assumes the original FP4-expert checkpoint layout.
    # Overridden per-instance in __init__ when expert_dtype != "fp4".
    hf_to_vllm_mapper = _make_deepseek_v4_weights_mapper("fp4")

    def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
        super().__init__()

        config = vllm_config.model_config.hf_config
        self.config = config
        expert_dtype = getattr(config, "expert_dtype", "fp4")
        fuse_shared_experts = _fuse_shared_experts_enabled(config)
        if expert_dtype != "fp4" or fuse_shared_experts:
            self.hf_to_vllm_mapper = _make_deepseek_v4_weights_mapper(
                expert_dtype, fuse_shared_experts=fuse_shared_experts
            )

        self.model = self.model_cls(
            vllm_config=vllm_config, prefix=maybe_prefix(prefix, "model")
        )
        if get_pp_group().is_last_rank:
            self.lm_head = ParallelLMHead(
                config.vocab_size,
                config.hidden_size,
                prefix=maybe_prefix(prefix, "lm_head"),
            )
        else:
            self.lm_head = PPMissingLayer()
        self.logits_processor = LogitsProcessor(config.vocab_size)
        self.make_empty_intermediate_tensors = (  # type: ignore[method-assign]
            self.model.make_empty_intermediate_tensors
        )

    def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
        return self.model.embed_input_ids(input_ids)

    def compute_logits(
        self,
        hidden_states: torch.Tensor,
    ) -> torch.Tensor | None:
        logits = self.logits_processor(self.lm_head, hidden_states)
        return logits

    def forward(
        self,
        input_ids: torch.Tensor,
        positions: torch.Tensor,
        intermediate_tensors: IntermediateTensors | None = None,
        inputs_embeds: torch.Tensor | None = None,
    ) -> torch.Tensor | IntermediateTensors:
        hidden_states = self.model(
            input_ids, positions, intermediate_tensors, inputs_embeds
        )
        return hidden_states

    def get_mtp_target_hidden_states(self) -> torch.Tensor | None:
        """Pre-hc_head residual stream buffer (max_num_batched_tokens,
        hc_mult * hidden_size) for the MTP draft model. Populated by
        forward(); valid after each target step."""
        return getattr(self.model, "_mtp_hidden_buffer", None)

    def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
        loader = AutoWeightsLoader(self, skip_substrs=["mtp."])
        loaded_params = loader.load_weights(weights, mapper=self.hf_to_vllm_mapper)
        return loaded_params

    def get_expert_mapping(self) -> list[tuple[str, str, int, str]]:
        return self.model.get_expert_mapping()

get_mtp_target_hidden_states()

Pre-hc_head residual stream buffer (max_num_batched_tokens, hc_mult * hidden_size) for the MTP draft model. Populated by forward(); valid after each target step.

Source code in vllm/models/deepseek_v4/amd/model.py 
958
959
960
961
962
def get_mtp_target_hidden_states(self) -> torch.Tensor | None:
    """Pre-hc_head residual stream buffer (max_num_batched_tokens,
    hc_mult * hidden_size) for the MTP draft model. Populated by
    forward(); valid after each target step."""
    return getattr(self.model, "_mtp_hidden_buffer", None)

_fuse_shared_experts_enabled(config, prefix='')

Whether to fuse the shared expert into the routed MXFP4 grouped GEMM.

Fusion fuses the shared expert into the routed experts' MXFP4 grouped GEMM, so it only applies where the shared expert is the same precision as the routed experts. Some layers may carry a shared expert in a different quantization than the routed experts; when so, it runs as its own linear and must not be fused.

Source code in vllm/models/deepseek_v4/amd/model.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def _fuse_shared_experts_enabled(config, prefix: str = "") -> bool:
    """Whether to fuse the shared expert into the routed MXFP4 grouped GEMM.

    Fusion fuses the shared expert into the routed experts' MXFP4 grouped GEMM,
    so it only applies where the shared expert is the same precision as the
    routed experts. Some layers may carry a shared expert in a different quantization
    than the routed experts; when so, it runs as its own linear and must not be fused.
    """
    if not (
        current_platform.is_rocm()
        and getattr(config, "n_shared_experts", None)
        and envs.VLLM_ROCM_USE_AITER_FUSION_SHARED_EXPERTS
        and not get_current_vllm_config().parallel_config.enable_expert_parallel
    ):
        return False
    return _shared_experts_are_fp4(
        config, extract_layer_index(prefix) if prefix else None
    )

_shared_experts_are_fp4(config, layer_idx=None)

Whether the shared experts are MXFP4 and thus fusable.

layer_idx=None resolves the model-wide default (global scheme), used by the main-model weight loader / mapper callers that operate per-model.

Source code in vllm/models/deepseek_v4/amd/model.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def _shared_experts_are_fp4(config, layer_idx: int | None = None) -> bool:
    """Whether the shared experts are MXFP4 and thus fusable.

    ``layer_idx=None`` resolves the model-wide default (global scheme), used by
    the main-model weight loader / mapper callers that operate per-model.
    """
    quant_cfg = getattr(config, "quantization_config", None)
    if quant_cfg is None:
        return False
    if layer_idx is None:
        base = None
    elif layer_idx >= config.num_hidden_layers:
        base = f"mtp.{layer_idx - config.num_hidden_layers}.ffn.shared_experts"
    else:
        base = f"layers.{layer_idx}.ffn.shared_experts"
    if base and any(e.startswith(base) for e in (quant_cfg.get("exclude") or [])):
        return False
    entry = (
        (quant_cfg.get("layer_quant_config") or {}).get(f"{base}.w1") if base else None
    )
    if entry is None:
        entry = quant_cfg.get("global_quant_config")
    return ((entry or {}).get("weight") or {}).get("dtype") == "fp4"