diff --git a/python/sglang/srt/managers/mm_utils.py b/python/sglang/srt/managers/mm_utils.py index 914c96cda..98b9371ab 100644 --- a/python/sglang/srt/managers/mm_utils.py +++ b/python/sglang/srt/managers/mm_utils.py @@ -660,43 +660,46 @@ def general_mm_embed_routine( """ assert hasattr(language_model, "get_input_embeddings") embed_tokens = language_model.get_input_embeddings() - if ( - not forward_batch.forward_mode.is_decode() - and not forward_batch.forward_mode.is_target_verify() - and forward_batch.contains_mm_inputs() - ): - mm_inputs_list = [ - mm_input for mm_input in forward_batch.mm_inputs if mm_input is not None - ] - extend_prefix_lens = [ - prefix_len - for i, prefix_len in enumerate(forward_batch.extend_prefix_lens_cpu) - if forward_batch.mm_inputs[i] is not None - ] - extend_seq_lens = [ - seq_len - for i, seq_len in enumerate(forward_batch.extend_seq_lens_cpu) - if forward_batch.mm_inputs[i] is not None - ] - inputs_embeds, other_info = embed_mm_inputs( - mm_inputs_list=mm_inputs_list, - extend_prefix_lens=extend_prefix_lens, - extend_seq_lens=extend_seq_lens, - input_ids=input_ids, - multimodal_model=multimodal_model, - input_embedding=embed_tokens, - data_embedding_func_mapping=data_embedding_funcs, - placeholder_tokens=placeholder_tokens, - use_deepstack=use_deepstack, - ) - # add for qwen3_vl deepstack - if use_deepstack: - kwargs["input_deepstack_embeds"] = other_info["input_deepstack_embeds"] - # once used, mm_inputs is useless, considering chunked-prefill is disabled for multimodal models - # just being defensive here - forward_batch.mm_inputs = None + if not hasattr(language_model, "pp_group") or language_model.pp_group.is_first_rank: + if ( + not forward_batch.forward_mode.is_decode() + and not forward_batch.forward_mode.is_target_verify() + and forward_batch.contains_mm_inputs() + ): + mm_inputs_list = [ + mm_input for mm_input in forward_batch.mm_inputs if mm_input is not None + ] + extend_prefix_lens = [ + prefix_len + for i, prefix_len in enumerate(forward_batch.extend_prefix_lens_cpu) + if forward_batch.mm_inputs[i] is not None + ] + extend_seq_lens = [ + seq_len + for i, seq_len in enumerate(forward_batch.extend_seq_lens_cpu) + if forward_batch.mm_inputs[i] is not None + ] + inputs_embeds, other_info = embed_mm_inputs( + mm_inputs_list=mm_inputs_list, + extend_prefix_lens=extend_prefix_lens, + extend_seq_lens=extend_seq_lens, + input_ids=input_ids, + multimodal_model=multimodal_model, + input_embedding=embed_tokens, + data_embedding_func_mapping=data_embedding_funcs, + placeholder_tokens=placeholder_tokens, + use_deepstack=use_deepstack, + ) + # add for qwen3_vl deepstack + if use_deepstack: + kwargs["input_deepstack_embeds"] = other_info["input_deepstack_embeds"] + # once used, mm_inputs is useless, considering chunked-prefill is disabled for multimodal models + # just being defensive here + forward_batch.mm_inputs = None + else: + inputs_embeds = embed_tokens(input_ids) else: - inputs_embeds = embed_tokens(input_ids) + inputs_embeds = None hidden_states = language_model( input_ids=None, diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index 9fb7003ef..dc257dcdf 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -382,10 +382,12 @@ class Scheduler( # avoiding any coupling with CUDA streams/devices. if self.server_args.enable_dp_attention: self.cpu_group = self.attn_tp_cpu_group + self.entry_rank = self.attn_tp_group.first_rank self.is_entry_rank = self.attn_tp_rank == 0 else: self.cpu_group = self.tp_cpu_group - self.is_entry_rank = self.tp_group.rank == 0 + self.entry_rank = self.tp_group.first_rank + self.is_entry_rank = self.tp_group.rank_in_group == 0 self.pad_input_ids_func = self.tp_worker.get_pad_input_ids_func() set_random_seed(self.random_seed) @@ -1221,7 +1223,7 @@ class Scheduler( if group_world_size > 1: obj_list = [image_inputs] torch.distributed.broadcast_object_list( - obj_list, src=0, group=self.cpu_group + obj_list, src=self.entry_rank, group=self.cpu_group ) image_inputs = obj_list[0] else: @@ -1229,7 +1231,7 @@ class Scheduler( if group_world_size > 1: obj_list = [None] torch.distributed.broadcast_object_list( - obj_list, src=0, group=self.cpu_group + obj_list, src=self.entry_rank, group=self.cpu_group ) image_inputs = obj_list[0] else: diff --git a/python/sglang/srt/models/qwen2_5_vl.py b/python/sglang/srt/models/qwen2_5_vl.py index b6abf722b..75660a1fe 100644 --- a/python/sglang/srt/models/qwen2_5_vl.py +++ b/python/sglang/srt/models/qwen2_5_vl.py @@ -40,6 +40,7 @@ from transformers.models.qwen2_5_vl.modeling_qwen2_5_vl import ( Qwen2_5_VisionRotaryEmbedding, ) +from sglang.srt.distributed.parallel_state import get_pp_group from sglang.srt.layers.attention.vision import VisionAttention from sglang.srt.layers.layernorm import RMSNorm from sglang.srt.layers.linear import ( @@ -50,13 +51,14 @@ from sglang.srt.layers.linear import ( from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.pooler import Pooler, PoolingType from sglang.srt.layers.quantization.base_config import QuantizationConfig +from sglang.srt.layers.utils import PPMissingLayer, get_layer_id from sglang.srt.layers.vocab_parallel_embedding import ParallelLMHead from sglang.srt.managers.mm_utils import ( MultiModalityDataPaddingPatternMultimodalTokens, general_mm_embed_routine, ) from sglang.srt.managers.schedule_batch import MultimodalDataItem, MultimodalInputs -from sglang.srt.model_executor.forward_batch_info import ForwardBatch +from sglang.srt.model_executor.forward_batch_info import ForwardBatch, PPProxyTensors from sglang.srt.model_loader.weight_utils import default_weight_loader from sglang.srt.models.qwen2 import Qwen2Model from sglang.srt.models.utils import permute_inv @@ -482,6 +484,7 @@ class Qwen2_5_VLForConditionalGeneration(nn.Module): ) -> None: super().__init__() + self.pp_group = get_pp_group() self.config = config self.visual = Qwen2_5_VisionTransformer( config.vision_config, @@ -498,15 +501,20 @@ class Qwen2_5_VLForConditionalGeneration(nn.Module): prefix=add_prefix("model", prefix), ) - if config.tie_word_embeddings: - self.lm_head = self.model.embed_tokens + if self.pp_group.is_last_rank: + if self.pp_group.world_size == 1 and self.config.tie_word_embeddings: + self.lm_head = self.model.embed_tokens + else: + self.lm_head = ParallelLMHead( + self.config.vocab_size, + self.config.hidden_size, + quant_config=quant_config, + prefix=add_prefix("lm_head", prefix), + ) else: - self.lm_head = ParallelLMHead( - config.vocab_size, - config.hidden_size, - quant_config=quant_config, - prefix=add_prefix("lm_head", prefix), - ) + # ranks other than the last rank will have a placeholder layer + self.lm_head = PPMissingLayer() + self.is_mrope_enabled = "mrope_section" in self.config.rope_scaling self.logits_processor = LogitsProcessor(config) @@ -551,6 +559,7 @@ class Qwen2_5_VLForConditionalGeneration(nn.Module): positions: torch.Tensor, forward_batch: ForwardBatch, get_embedding: bool = False, + pp_proxy_tensors: Optional[PPProxyTensors] = None, ): """Run forward pass for Qwen2_5-VL. @@ -583,18 +592,25 @@ class Qwen2_5_VLForConditionalGeneration(nn.Module): language_model=self.model, multimodal_model=self, positions=positions, + pp_proxy_tensors=pp_proxy_tensors, ) aux_hidden_states = None if self.capture_aux_hidden_states: hidden_states, aux_hidden_states = hidden_states - if not get_embedding: - return self.logits_processor( - input_ids, hidden_states, self.lm_head, forward_batch, aux_hidden_states - ) + if self.pp_group.is_last_rank: + if not get_embedding: + return self.logits_processor( + input_ids, + hidden_states, + self.lm_head, + forward_batch, + ) + else: + return self.pooler(hidden_states, forward_batch) else: - return self.pooler(hidden_states, forward_batch) + return hidden_states def load_weights(self, weights: Iterable[Tuple[str, torch.Tensor]]): stacked_params_mapping = [ @@ -620,6 +636,16 @@ class Qwen2_5_VLForConditionalGeneration(nn.Module): ): continue name = name.replace(weight_name, param_name) + layer_id = get_layer_id(name) + if ( + layer_id is not None + and hasattr(self.model, "start_layer") + and ( + layer_id < self.model.start_layer + or layer_id >= self.model.end_layer + ) + ): + continue # Skip loading extra bias for GPTQ models. if name.endswith(".bias") and name not in params_dict: @@ -637,7 +663,10 @@ class Qwen2_5_VLForConditionalGeneration(nn.Module): # Skip loading extra bias for GPTQ models. if name.endswith(".bias") and name not in params_dict: continue - param = params_dict[name] + if name in params_dict.keys(): + param = params_dict[name] + else: + continue except KeyError: print(params_dict.keys()) raise