[VLM] refactor: refactor load_mm_data to improve performance (#14644)

Co-authored-by: luoyuan.luo <luoyuan.luo@antgroup.com>

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
 

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)