| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | from importlib import import_module |
| 5 | from types import SimpleNamespace |
| 6 | from unittest.mock import patch |
| 7 | |
| 8 | import pytest |
| 9 | import torch |
| 10 | from transformers import Qwen4ExpConfig, Qwen4ExpTextConfig |
| 11 | |
| 12 | from vllm.config.speculative import SpeculativeConfig |
| 13 | from vllm.model_executor.models.config import ( |
| 14 | Qwen3_5ForConditionalGenerationConfig, |
| 15 | Qwen4ExpForConditionalGenerationConfig, |
| 16 | ) |
| 17 | from vllm.models.qwen4_exp.nvidia.model_state import Qwen4ExpModelState |
| 18 | from vllm.v1.worker.gpu.model_states.mamba_hybrid import MambaHybridModelState |
| 19 | |
| 20 | from ...utils import spawn_new_process_for_each_test |
| 21 | |
| 22 | |
| 23 | def _text_config(**kwargs) -> Qwen4ExpTextConfig: |
| 24 | values = { |
| 25 | "vocab_size": 64, |
| 26 | "hidden_size": 16, |
| 27 | "intermediate_size": 32, |
| 28 | "num_hidden_layers": 2, |
| 29 | "num_attention_heads": 2, |
| 30 | "num_key_value_heads": 1, |
| 31 | "head_dim": 8, |
| 32 | "layer_types": ["linear_attention", "full_attention"], |
| 33 | "linear_num_key_heads": 2, |
| 34 | "linear_num_value_heads": 2, |
| 35 | "linear_key_head_dim": 8, |
| 36 | "linear_value_head_dim": 8, |
| 37 | "num_experts": 4, |
| 38 | "num_experts_per_tok": 2, |
| 39 | # `Qwen4ExpTextConfig` requires an EOS token whenever PLE is enabled. |
| 40 | "eos_token_id": 1, |
| 41 | "hc_count": 2, |
| 42 | "hc_lowrank": 4, |
| 43 | "ple_layer_ids": [1], |
| 44 | "mtp_num_hidden_layers": 1, |
| 45 | "mtp": {"hybrid": True}, |
| 46 | } |
| 47 | values.update(kwargs) |
| 48 | return Qwen4ExpTextConfig(**values) |
| 49 | |
| 50 | |
| 51 | def test_qwen4_exp_mtp_returns_sample_and_multi_streams() -> None: |
| 52 | from vllm.models.qwen4_exp.nvidia.mtp import ( |
| 53 | Qwen4ExpMultiTokenPredictor, |
| 54 | ) |
| 55 | |
| 56 | model = object.__new__(Qwen4ExpMultiTokenPredictor) |
| 57 | torch.nn.Module.__init__(model) |
| 58 | model.hc_count = 2 |
| 59 | model.hidden_size = 4 |
| 60 | model.num_mtp_layers = 1 |
| 61 | model.layers = [ |
| 62 | lambda **kwargs: ( |
| 63 | kwargs["hidden_states"], |
| 64 | kwargs["hidden_states"], |
| 65 | torch.zeros(kwargs["hidden_states"].shape[0], 2), |
| 66 | ), |
| 67 | ] |
| 68 | model.hyper_connection_mixer = SimpleNamespace( |
| 69 | combine_and_mix=lambda hidden_states, block_output, injection: ( |
| 70 | hidden_states, |
| 71 | hidden_states.unflatten(-1, (2, 4)).mean(-2), |
| 72 | None, |
| 73 | ), |
| 74 | ) |
| 75 | multi_hidden = torch.arange(16, dtype=torch.float32).reshape(2, 8) |
| 76 | pp_group = SimpleNamespace(is_first_rank=False, is_last_rank=True) |
| 77 | |
| 78 | with patch( |
| 79 | "vllm.models.qwen4_exp.nvidia.mtp.get_pp_group", |
| 80 | return_value=pp_group, |
| 81 | ): |
| 82 | sample_hidden, returned_multi_hidden = model.forward( |
| 83 | input_ids=None, |
| 84 | positions=torch.arange(2), |
| 85 | intermediate_tensors={"hidden_states": multi_hidden}, |
| 86 | ) |
| 87 | |
| 88 | torch.testing.assert_close( |
| 89 | sample_hidden, |
| 90 | multi_hidden.unflatten(-1, (2, 4)).mean(dim=-2), |
| 91 | ) |
| 92 | assert returned_multi_hidden is multi_hidden |
| 93 | |
| 94 | |
| 95 | @spawn_new_process_for_each_test |
| 96 | @pytest.mark.parametrize("backend", ["amd", "nvidia"]) |
| 97 | def test_qwen4_exp_mtp_remaps_mixed_precision_layer_indices(backend: str) -> None: |
| 98 | mtp_module = import_module(f"vllm.models.qwen4_exp.{backend}.mtp") |
| 99 | |
| 100 | quantized_layers = { |
| 101 | "model.language_model.layers.0.mlp.experts": {"quant_algo": "NVFP4"}, |
| 102 | "mtp.layers.0.mlp.experts": { |
| 103 | "quant_algo": "FP8_BLOCK_SCALES", |
| 104 | "group_size": 128, |
| 105 | }, |
| 106 | } |
| 107 | |
| 108 | assert mtp_module._remap_quantized_layers(quantized_layers, 48) == { |
| 109 | "model.language_model.layers.0.mlp.experts": {"quant_algo": "NVFP4"}, |
| 110 | "mtp.layers.48.mlp.experts": { |
| 111 | "quant_algo": "FP8_BLOCK_SCALES", |
| 112 | "group_size": 128, |
| 113 | }, |
| 114 | } |
| 115 | |
| 116 | |
| 117 | @pytest.mark.parametrize("wrapped_config", [False, True]) |
| 118 | def test_qwen4_exp_mtp_override_sets_draft_config( |
| 119 | wrapped_config: bool, |
| 120 | ) -> None: |
| 121 | text_config = _text_config( |
| 122 | architectures=["Qwen4ExpForCausalLM"], |
| 123 | index_share_for_mtp_iteration=True, |
| 124 | ) |
| 125 | config = ( |
| 126 | Qwen4ExpConfig( |
| 127 | architectures=["Qwen4ExpForConditionalGeneration"], |
| 128 | text_config=text_config, |
| 129 | ) |
| 130 | if wrapped_config |
| 131 | else text_config |
| 132 | ) |
| 133 | |
| 134 | draft_config = SpeculativeConfig.hf_config_override(config) |
| 135 | |
| 136 | assert draft_config.index_share_for_mtp_iteration is True |
| 137 | assert draft_config.model_type == "qwen4_exp_mtp" |
| 138 | assert draft_config.architectures == ["Qwen4ExpMTP"] |
| 139 | assert draft_config.hc_mult == 2 |
| 140 | assert draft_config.n_predict == 1 |
| 141 | |
| 142 | |
| 143 | @pytest.mark.parametrize("ple_layer_ids", [[1], []]) |
| 144 | def test_qwen4_exp_rejects_pipeline_parallel_only_with_ple(ple_layer_ids) -> None: |
| 145 | """PLE needs raw input_ids, which non-first pipeline ranks never see. The |
| 146 | rest of the architecture is PP-capable, so the refusal must be conditional |
| 147 | -- and must land before the engine spends time loading weights.""" |
| 148 | vllm_config = SimpleNamespace( |
| 149 | model_config=SimpleNamespace( |
| 150 | hf_text_config=_text_config(ple_layer_ids=ple_layer_ids), |
| 151 | multimodal_config=None, |
| 152 | ), |
| 153 | parallel_config=SimpleNamespace( |
| 154 | pipeline_parallel_size=2, enable_dbo=False, ubatch_size=1 |
| 155 | ), |
| 156 | speculative_config=None, |
| 157 | ) |
| 158 | with patch.object( |
| 159 | Qwen3_5ForConditionalGenerationConfig, "verify_and_update_config" |
| 160 | ): |
| 161 | if ple_layer_ids: |
| 162 | with pytest.raises(NotImplementedError, match="pipeline_parallel_size=1"): |
| 163 | Qwen4ExpForConditionalGenerationConfig.verify_and_update_config( |
| 164 | vllm_config |
| 165 | ) |
| 166 | else: |
| 167 | Qwen4ExpForConditionalGenerationConfig.verify_and_update_config(vllm_config) |
| 168 | |
| 169 | |
| 170 | def test_qwen4_exp_model_state_prepares_ngram_context() -> None: |
| 171 | model_state = object.__new__(Qwen4ExpModelState) |
| 172 | model_state.uses_ngram_embedding = True |
| 173 | model_state.ngram_context_len = 3 |
| 174 | model_state.ngram_eos_token_id = 99 |
| 175 | model_state.ngram_context = torch.empty((8, 3), dtype=torch.int32) |
| 176 | model_state.ngram_context_offsets = torch.arange(-3, 0, dtype=torch.int64) |
| 177 | model_state.ple_query_start_loc = torch.empty(9, dtype=torch.int32) |
| 178 | |
| 179 | input_batch = SimpleNamespace( |
| 180 | num_reqs=2, |
| 181 | num_reqs_after_padding=3, |
| 182 | idx_mapping=torch.tensor([1, 0]), |
| 183 | query_start_loc=torch.tensor([0, 2, 3, 3], dtype=torch.int32), |
| 184 | ) |
| 185 | req_states = SimpleNamespace( |
| 186 | num_computed_tokens=SimpleNamespace(gpu=torch.tensor([3, 1])), |
| 187 | all_token_ids=SimpleNamespace( |
| 188 | gpu=torch.tensor([[1, 2, 3, 4], [20, 21, 22, 23]], dtype=torch.int32) |
| 189 | ), |
| 190 | ) |
| 191 | |
| 192 | with patch.object(MambaHybridModelState, "prepare_inputs", return_value={}): |
| 193 | model_inputs = model_state.prepare_inputs(input_batch, req_states) |
| 194 | |
| 195 | expected_query_start_loc = torch.full((9,), 3, dtype=torch.int32) |
| 196 | expected_query_start_loc[0] = 0 |
| 197 | expected_query_start_loc[1] = 2 |
| 198 | torch.testing.assert_close( |
| 199 | model_inputs["query_start_loc"], expected_query_start_loc |
| 200 | ) |
| 201 | expected_context = torch.full((8, 3), 99, dtype=torch.int32) |
| 202 | expected_context[:2] = torch.tensor([[99, 99, 20], [1, 2, 3]]) |
| 203 | torch.testing.assert_close(model_inputs["ngram_context"], expected_context) |
| 204 | |
| 205 | # Retain the views to detect reallocations as the request layout changes. |
| 206 | query_start_loc = model_inputs["query_start_loc"] |
| 207 | ngram_context = model_inputs["ngram_context"] |
| 208 | input_batch.num_reqs = 1 |
| 209 | input_batch.num_reqs_after_padding = 1 |
| 210 | input_batch.idx_mapping = torch.tensor([0]) |
| 211 | input_batch.query_start_loc = torch.tensor([0, 3], dtype=torch.int32) |
| 212 | with patch.object(MambaHybridModelState, "prepare_inputs", return_value={}): |
| 213 | model_inputs = model_state.prepare_inputs(input_batch, req_states) |
| 214 | |
| 215 | expected_query_start_loc.fill_(3) |
| 216 | expected_query_start_loc[0] = 0 |
| 217 | torch.testing.assert_close( |
| 218 | model_inputs["query_start_loc"], expected_query_start_loc |
| 219 | ) |
| 220 | expected_context.fill_(99) |
| 221 | expected_context[0] = torch.tensor([1, 2, 3]) |
| 222 | torch.testing.assert_close(model_inputs["ngram_context"], expected_context) |
| 223 | assert model_inputs["query_start_loc"].data_ptr() == query_start_loc.data_ptr() |
| 224 | assert model_inputs["ngram_context"].data_ptr() == ngram_context.data_ptr() |
| 225 | |
| 226 | |
| 227 | def test_qwen4_exp_model_state_prepares_stable_dummy_ngram_inputs() -> None: |
| 228 | model_state = object.__new__(Qwen4ExpModelState) |
| 229 | model_state.uses_ngram_embedding = True |
| 230 | model_state.ngram_eos_token_id = 99 |
| 231 | model_state.ngram_context = torch.empty((8, 3), dtype=torch.int32) |
| 232 | model_state.ple_query_start_loc = torch.empty(9, dtype=torch.int32) |
| 233 | |
| 234 | with patch.object(MambaHybridModelState, "prepare_dummy_inputs", return_value={}): |
| 235 | first = model_state.prepare_dummy_inputs(num_reqs=3, num_tokens=4) |
| 236 | # Dummy runs establish the addresses used during CUDA graph capture. |
| 237 | query_start_loc_ptr = first["query_start_loc"].data_ptr() |
| 238 | ngram_context_ptr = first["ngram_context"].data_ptr() |
| 239 | second = model_state.prepare_dummy_inputs(num_reqs=3, num_tokens=4) |
| 240 | |
| 241 | expected_query_start_loc = torch.full((9,), 4, dtype=torch.int32) |
| 242 | expected_query_start_loc[:4] = torch.tensor([0, 1, 2, 4], dtype=torch.int32) |
| 243 | torch.testing.assert_close(second["query_start_loc"], expected_query_start_loc) |
| 244 | torch.testing.assert_close( |
| 245 | second["ngram_context"], torch.full((8, 3), 99, dtype=torch.int32) |
| 246 | ) |
| 247 | assert second["query_start_loc"].data_ptr() == query_start_loc_ptr |
| 248 | assert second["ngram_context"].data_ptr() == ngram_context_ptr |