bugfix for mtp with multistream_moe (#3419)

### What this PR does / why we need it?
when infer deepseek mtp layer with multistream_moe, we should pass a
boolean to evaluate this feature and fix bugs when we are in mtp layer

- vLLM version: v0.11.0rc3
- vLLM main: https://github.com/vllm-project/vllm/commit/v0.11.0

---------

Signed-off-by: zouyida2052 <[email protected]>

Author: zouyida2052

tests/e2e/singlecard/spec_decode_v1/test_v1_mtp_torchair_correctness.py

@@ -41,6 +41,7 @@ def test_mtp_torchair_correctness(
                             "use_cached_graph": False,
                             "graph_batch_sizes": [1, 2, 4],
                         },
+                        "multistream_overlap_shared_expert": "True"
                     }) as ref_llm:
         ref_outputs = ref_llm.generate(example_prompts, sampling_config)
     with VllmRunner(model_name,
@@ -60,7 +61,8 @@ def test_mtp_torchair_correctness(
                             "enabled": True,
                             "use_cached_graph": False,
                             "graph_batch_sizes": [1, 2, 4],
-                        }
+                        },
+                        "multistream_overlap_shared_expert": "True"
                     }) as spec_llm:
         spec_outputs = spec_llm.generate(example_prompts, sampling_config)
 

tests/ut/torchair/models/test_torchair_deepseek_mtp.py

@@ -17,6 +17,9 @@ def setup_mtp_layer(self, mocker: MockerFixture):
         config = PretrainedConfig(vocab_size=1000,
                                   hidden_size=768,
                                   rms_norm_eps=1e-5)
+        mocker.patch(
+            'vllm_ascend.torchair.models.torchair_deepseek_mtp.get_tensor_model_parallel_world_size',
+            return_value=1)
         mocker.patch(
             "vllm.model_executor.layers.vocab_parallel_embedding.VocabParallelEmbedding.__init__",
             return_value=None)
@@ -56,6 +59,8 @@ def test_forward(self, mocker: MockerFixture, setup_mtp_layer):
         mocker.patch("torch.cat", return_value=torch.randn(2, 3, 768))
         mtp_layer.mtp_block.return_value = (torch.randn(2, 3, 768),
                                             torch.randn(2, 3, 768))
+        mtp_layer.enorm.return_value = torch.randn(2, 3, 768)
+        mtp_layer.hnorm.return_value = torch.randn(2, 3, 768)
 
         input_ids = torch.tensor([[1, 2, 3], [4, 5, 6]])
         positions = torch.tensor([[0, 1, 2], [0, 1, 2]])
@@ -65,7 +70,7 @@ def test_forward(self, mocker: MockerFixture, setup_mtp_layer):
 
         output = mtp_layer(input_ids, positions, kv_cache, None,
                            previous_hidden_states, inputs_embeds, 0)
-        assert output.shape == (2, 3, 768)
+        assert output.shape == (3, 768)
 
 
 class TestTorchairDeepSeekMultiTokenPredictor(PytestBase):

vllm_ascend/attention/utils.py

@@ -103,8 +103,6 @@ def split_decodes_and_prefills(
         return num_reqs, 0, num_tokens, 0
 
     first_prefill = is_prefill.int().argmax(dim=-1).item()
-    assert torch.all(query_lens[first_prefill:] > decode_threshold)
-    assert torch.all(query_lens[:first_prefill] <= decode_threshold)
     num_decodes = first_prefill
     num_prefills = num_reqs - num_decodes
     num_decode_tokens = query_start_loc[first_prefill].item()

vllm_ascend/torchair/models/torchair_deepseek_mtp.py

@@ -24,6 +24,7 @@
 from transformers import PretrainedConfig
 from vllm.attention.backends.abstract import AttentionMetadata
 from vllm.config import CacheConfig, ModelConfig, VllmConfig
+from vllm.distributed import get_tensor_model_parallel_world_size
 from vllm.model_executor.layers.layernorm import RMSNorm
 from vllm.model_executor.layers.logits_processor import LogitsProcessor
 from vllm.model_executor.layers.quantization import QuantizationConfig
@@ -66,6 +67,7 @@ def __init__(
     ) -> None:
         nn.Module.__init__(self)
 
+        self.tp_size = get_tensor_model_parallel_world_size()
         self.enorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
         self.hnorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
         self.eh_proj = nn.Linear(config.hidden_size * 2,
@@ -100,11 +102,15 @@ def forward(
         hidden_states = self.eh_proj(
             torch.cat([inputs_embeds, previous_hidden_states], dim=-1))
 
-        hidden_states, residual = self.mtp_block(positions=positions,
-                                                 hidden_states=hidden_states,
-                                                 kv_cache=kv_cache,
-                                                 attn_metadata=attn_metadata,
-                                                 residual=None)
+        replace_allreduce = hidden_states.shape[0] % self.tp_size == 0
+
+        hidden_states, residual = self.mtp_block(
+            positions=positions,
+            hidden_states=hidden_states,
+            residual=None,
+            kv_cache=kv_cache,
+            attn_metadata=attn_metadata,
+            replace_allreduce=replace_allreduce)
         hidden_states = residual + hidden_states
         return hidden_states
 

vllm_ascend/torchair/models/torchair_deepseek_v2.py

@@ -975,7 +975,7 @@ def forward(
                 # to save npu memory because they're no longer used.
                 dispose_tensor(previous_hidden_states)
                 dispose_tensor(previous_residual)
-        if mla_moe_communication and self.layer_idx > self.first_k_dense_replace:
+        if mla_moe_communication and self.layer_idx > self.first_k_dense_replace and self.layer_idx < self.layers:
             hidden_states = tensor_model_parallel_all_gather(hidden_states,
                                                              dim=0)
 
@@ -1034,7 +1034,7 @@ def forward(
             # The scaling of DeepseekV2MOE output would be done in the forward
             # of DeepseekV2MOE
             hidden_states *= 1. / self.routed_scaling_factor
-        if mla_moe_communication and self.layer_idx == self.layers - 1:
+        if mla_moe_communication and self.layer_idx >= self.layers - 1:
             hidden_states = tensor_model_parallel_all_gather(hidden_states,
                                                              dim=0)
             residual = tensor_model_parallel_all_gather(residual, dim=0)