From 086813ae8addfa354b718b49a6c557dd9d1089a4 Mon Sep 17 00:00:00 2001 From: Yuan Luo Date: Fri, 26 Dec 2025 13:17:53 +0800 Subject: [PATCH] [VLM] refactor: refactor load_mm_data to improve performance (#14644) Co-authored-by: luoyuan.luo --- .../multimodal/processors/base_processor.py | 164 ++++++++++++++++++ .../srt/multimodal/processors/minicpm.py | 1 + 2 files changed, 165 insertions(+) diff --git a/python/sglang/srt/multimodal/processors/base_processor.py b/python/sglang/srt/multimodal/processors/base_processor.py index ca0dc1f57..29aef6b14 100644 --- a/python/sglang/srt/multimodal/processors/base_processor.py +++ b/python/sglang/srt/multimodal/processors/base_processor.py @@ -415,6 +415,49 @@ class BaseMultimodalProcessor(ABC): except Exception as e: raise RuntimeError(f"Error while loading data {data}: {e}") + def _submit_mm_data_loading_tasks_simple( + self, + data_list: Optional[list], + modality: Modality, + audio_sample_rate: Optional[int], + discard_alpha_channel: bool, + ) -> List[Tuple[Modality, int, concurrent.futures.Future]]: + """ + Simple version: For one modal data submit IO load task. + + Return: + List[(modality, index_in_that_modality, future)] + """ + futures: List[Tuple[Modality, int, concurrent.futures.Future]] = [] + + if not data_list: + logger.debug( + "[_submit_mm_data_loading_tasks_simple] no data for modality=%s", + modality.name, + ) + return futures + + for idx, data in enumerate(data_list): + logger.debug( + "[_submit_mm_data_loading_tasks_simple] submit load task: " + "modality=%s, index=%d, data_type=%s", + modality.name, + idx, + type(data), + ) + future = self.io_executor.submit( + BaseMultimodalProcessor._load_single_item, + data, + modality, + None, # frame_count_limit: no consider for fast path + audio_sample_rate, + discard_alpha_channel, + ) + futures.append((modality, idx, future)) + + return futures + + # TODO(yuan-luo): To be obsoleted. def submit_data_loading_tasks( self, text_parts: List[str], @@ -569,6 +612,127 @@ class BaseMultimodalProcessor(ABC): discard_alpha_channel: bool = True, audio_sample_rate: Optional[int] = None, ) -> BaseMultiModalProcessorOutput: + + # For MiniCPMO and MiniCPMV + if getattr(self, "support_dynamic_frame_expansion", False): + return self.legacy_load_mm_data( + prompt=prompt, + multimodal_tokens=multimodal_tokens, + image_data=image_data, + video_data=video_data, + audio_data=audio_data, + return_text=return_text, + discard_alpha_channel=discard_alpha_channel, + audio_sample_rate=audio_sample_rate, + ) + # For models other than MiniCPMO and MiniCPMV + else: + return self.fast_load_mm_data( + prompt=prompt, + multimodal_tokens=multimodal_tokens, + image_data=image_data, + video_data=video_data, + audio_data=audio_data, + return_text=return_text, + discard_alpha_channel=discard_alpha_channel, + audio_sample_rate=audio_sample_rate, + ) + + def fast_load_mm_data( + self, + prompt: str, + multimodal_tokens: MultimodalSpecialTokens, + image_data: Optional[list] = None, + video_data: Optional[list] = None, + audio_data: Optional[list] = None, + return_text: Optional[bool] = True, + discard_alpha_channel: bool = True, + audio_sample_rate: Optional[int] = None, + ) -> BaseMultiModalProcessorOutput: + """ + A fast version of `load_mm_data` that loads multimodal data directly. + This version does not scan the prompt to recognize tokens. It assumes + that the caller has already aligned the tokens and data in a 1:1 manner. + The behavior is as follows: + 1. It runs `_load_single_item` for all input data concurrently. + 2. It returns the loaded images, videos, and audios in their original order. + 3. It returns the input prompt as a string. + """ + + # Convert prompt into str + if isinstance(prompt, list) and return_text: + assert len(prompt) and isinstance(prompt[0], int) + prompt_str = self._processor.tokenizer.decode(prompt) + else: + assert isinstance(prompt, str) + prompt_str = prompt + + futures: List[Tuple[Modality, int, concurrent.futures.Future]] = [] + + modalities_data = [ + (image_data, Modality.IMAGE), + (video_data, Modality.VIDEO), + (audio_data, Modality.AUDIO), + ] + + for data_list, modality in modalities_data: + futures.extend( + self._submit_mm_data_loading_tasks_simple( + data_list, modality, audio_sample_rate, discard_alpha_channel + ) + ) + + logger.debug("[load_mm_data(simple)] total futures submitted: %d", len(futures)) + + images: List[Any] = [None] * len(image_data) if image_data else [] + videos: List[Any] = [None] * len(video_data) if video_data else [] + audios: List[Any] = [None] * len(audio_data) if audio_data else [] + + for modality, idx, future in futures: + try: + result = future.result() + except Exception as e: + logger.exception( + "[load_mm_data(simple)] error loading %s data at index=%d", + modality.name, + idx, + ) + raise RuntimeError( + f"An exception occurred while loading {modality.name} data at index {idx}: {e}" + ) + + if modality == Modality.IMAGE: + images[idx] = result + elif modality == Modality.VIDEO: + videos[idx] = result + elif modality == Modality.AUDIO: + audios[idx] = result + + logger.debug( + "[load_mm_data(simple)] loaded counts: images=%d, videos=%d, audios=%d", + len(images), + len(videos), + len(audios), + ) + + return BaseMultiModalProcessorOutput( + images=images, + audios=audios, + videos=videos, + input_text=prompt_str, + ) + + def legacy_load_mm_data( + self, + prompt: str, + multimodal_tokens: MultimodalSpecialTokens, + image_data: Optional[list] = None, + video_data: Optional[list] = None, + audio_data: Optional[list] = None, + return_text: Optional[bool] = True, + discard_alpha_channel: bool = True, + audio_sample_rate: Optional[int] = None, + ) -> BaseMultiModalProcessorOutput: """ Each frame of video/image will be replaced by a single image token diff --git a/python/sglang/srt/multimodal/processors/minicpm.py b/python/sglang/srt/multimodal/processors/minicpm.py index 9ddbf4fb6..defc047aa 100644 --- a/python/sglang/srt/multimodal/processors/minicpm.py +++ b/python/sglang/srt/multimodal/processors/minicpm.py @@ -14,6 +14,7 @@ from sglang.srt.multimodal.processors.base_processor import ( # Compatible with both 'O' and 'V' class MiniCPMMultimodalProcessor(BaseMultimodalProcessor): models = [MiniCPMV, MiniCPMO] + support_dynamic_frame_expansion = True def __init__(self, hf_config, server_args, _processor, *args, **kwargs): super().__init__(hf_config, server_args, _processor, *args, **kwargs)