feat(sglang): support --frontend-decoding for aggregated multimodal

components/src/dynamo/common/configuration/groups/frontend_decoding_args.py

@@ -0,0 +1,43 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared `--frontend-decoding` flag helper.
+
+Both `dynamo.vllm` and `dynamo.sglang` (and any future backend) expose the
+same `--frontend-decoding` knob — when set, the Rust frontend decodes
+multimodal images and ships pre-decoded pixels via NIXL RDMA instead of
+letting the engine fetch + decode. The flag definition is identical
+modulo the engine-specific env-var prefix (`DYN_VLLM_*` vs `DYN_SGL_*`),
+so the per-backend `backend_args.py` modules call this helper instead of
+duplicating the `add_negatable_bool_argument` block.
+
+Engine-specific validation (e.g. sglang rejects FD combined with the EPD
+topology) stays in each backend's `ConfigBase.validate()` — only the
+flag plumbing is shared.
+"""
+
+from dynamo.common.configuration.utils import add_negatable_bool_argument
+
+
+def add_frontend_decoding_arg(g, *, env_prefix: str) -> None:
+    """Add the shared `--frontend-decoding` flag to an argparse group.
+
+    Args:
+        g: an argparse argument group (from `parser.add_argument_group(...)`).
+        env_prefix: engine identifier inserted into the env var, e.g. `"VLLM"`
+            yields `DYN_VLLM_FRONTEND_DECODING`, `"SGL"` yields
+            `DYN_SGL_FRONTEND_DECODING`.
+    """
+    add_negatable_bool_argument(
+        g,
+        flag_name="--frontend-decoding",
+        env_var=f"DYN_{env_prefix}_FRONTEND_DECODING",
+        default=False,
+        help=(
+            "Enable frontend decoding of multimodal images. "
+            "Images are decoded in the Rust frontend and transferred to the "
+            "backend via NIXL RDMA, bypassing in-engine HTTP fetch + decode. "
+            "Engine-specific topology constraints may apply — see backend "
+            "validation."
+        ),
+    )

components/src/dynamo/sglang/backend_args.py

@@ -8,6 +8,9 @@
 
 from dynamo.common.configuration.arg_group import ArgGroup
 from dynamo.common.configuration.config_base import ConfigBase
+from dynamo.common.configuration.groups.frontend_decoding_args import (
+    add_frontend_decoding_arg,
+)
 from dynamo.common.configuration.utils import add_argument, add_negatable_bool_argument
 from dynamo.common.constants import EmbeddingTransferMode
 
@@ -109,6 +112,10 @@ def add_arguments(self, parser: argparse.ArgumentParser) -> None:
             help="Enable RL training support. Registers the call_tokenizer_manager engine route for generic tokenizer_manager passthrough.",
         )
 
+        # Topology constraint: rejecting --frontend-decoding combined with the
+        # EPD multimodal flags happens in DynamoSGLangConfig.validate() below.
+        add_frontend_decoding_arg(g, env_prefix="SGL")
+
 
 class DynamoSGLangConfig(ConfigBase):
     """Configuration for Dynamo SGLang wrapper (SGLang-specific only)."""
@@ -125,6 +132,7 @@ class DynamoSGLangConfig(ConfigBase):
 
     video_generation_worker: bool
     enable_rl: bool
+    frontend_decoding: bool = False
 
     def validate(self) -> None:
         if not isinstance(self.embedding_transfer_mode, EmbeddingTransferMode):
@@ -136,3 +144,14 @@ def validate(self) -> None:
             raise ValueError(
                 "Both 'disagg_config' and 'disagg_config_key' must be provided together."
             )
+
+        if self.frontend_decoding and (
+            self.multimodal_encode_worker or self.multimodal_worker
+        ):
+            raise ValueError(
+                "--frontend-decoding is incompatible with the EPD multimodal topology "
+                "(--multimodal-encode-worker / --multimodal-worker). The encode worker "
+                "needs URLs to run MMEncoder, while --frontend-decoding ships pre-decoded "
+                "pixels. Use --frontend-decoding on the default aggregated decode worker "
+                "(no --multimodal-* flag) instead."
+            )

components/src/dynamo/sglang/init_llm.py

@@ -119,7 +119,12 @@ async def init_decode(
     ready_event = asyncio.Event()
 
     handler = DecodeWorkerHandler(
-        engine, config, publisher, generate_endpoint, shutdown_event
+        engine,
+        config,
+        publisher,
+        generate_endpoint,
+        shutdown_event,
+        enable_frontend_decoding=dynamo_args.frontend_decoding,
     )
     handler.register_engine_routes(runtime)
 

components/src/dynamo/sglang/register.py

@@ -4,6 +4,7 @@
 import asyncio
 import json
 import logging
+import os
 from typing import Any, List, Optional
 
 import sglang as sgl
@@ -14,13 +15,37 @@
 
 from dynamo._core import Endpoint
 from dynamo.common.utils.output_modalities import get_output_modalities
-from dynamo.llm import ModelInput, ModelRuntimeConfig, ModelType, register_model
+from dynamo.llm import (
+    MediaDecoder,
+    MediaFetcher,
+    ModelInput,
+    ModelRuntimeConfig,
+    ModelType,
+    register_model,
+)
 from dynamo.sglang._compat import get_scheduler_info
 from dynamo.sglang.args import DynamoConfig
 
 SGLANG_HICACHE_MOONCAKE_RUNTIME_KEY = "sglang_hicache_mooncake"
 
 
+def _build_media_decoder_and_fetcher():
+    """Construct MediaDecoder/MediaFetcher for frontend-decoded multimodal.
+
+    Mirrors the vLLM backend pattern (components/src/dynamo/vllm/main.py).
+    """
+    media_decoder = MediaDecoder()
+    media_decoder.enable_image({"limits": {"max_alloc": 128 * 1024 * 1024}})
+
+    media_fetcher = MediaFetcher()
+    media_fetcher.timeout_ms(30000)
+    allow_internal = os.getenv("DYN_MM_ALLOW_INTERNAL", "0") == "1"
+    media_fetcher.allow_direct_ip(allow_internal)
+    media_fetcher.allow_direct_port(allow_internal)
+
+    return media_decoder, media_fetcher
+
+
 async def _register_model_with_runtime_config(
     engine: sgl.Engine,
     endpoint: Endpoint,
@@ -53,6 +78,13 @@ async def _register_model_with_runtime_config(
         if output_type != ModelType.Embedding:
             output_type = ModelType.Chat
 
+    # Configure the Rust frontend's media decoder so it ships pre-decoded
+    # images via NIXL RDMA instead of forwarding raw URLs / base64 to us.
+    media_decoder = None
+    media_fetcher = None
+    if getattr(dynamo_args, "frontend_decoding", False):
+        media_decoder, media_fetcher = _build_media_decoder_and_fetcher()
+
     try:
         await register_model(
             input_type,
@@ -64,6 +96,8 @@ async def _register_model_with_runtime_config(
             kv_cache_block_size=server_args.page_size,
             runtime_config=runtime_config,
             custom_template_path=dynamo_args.custom_jinja_template,
+            media_decoder=media_decoder,
+            media_fetcher=media_fetcher,
         )
         logging.info("Successfully registered LLM with runtime config")
         return True

components/src/dynamo/sglang/request_handlers/llm/decode_handler.py

@@ -9,9 +9,11 @@
 
 import pybase64
 import sglang as sgl
+from PIL.Image import Image as PILImage
 
 from dynamo._core import Context
 from dynamo.common.constants import DisaggregationMode
+from dynamo.common.multimodal.image_loader import ImageLoader
 from dynamo.common.utils.engine_response import normalize_finish_reason
 from dynamo.common.utils.otel_tracing import build_trace_headers
 from dynamo.sglang._compat import filter_supported_async_generate_kwargs
@@ -155,6 +157,7 @@ def __init__(
         publisher: Optional[DynamoSglangPublisher] = None,
         generate_endpoint=None,
         shutdown_event: Optional[asyncio.Event] = None,
+        enable_frontend_decoding: bool = False,
     ) -> None:
         """Initialize decode worker handler.
 
@@ -164,6 +167,10 @@ def __init__(
             publisher: Metrics publisher for the worker.
             shutdown_event: Optional event to signal shutdown.
             generate_endpoint: The endpoint handle for discovery registration.
+            enable_frontend_decoding: If True, multimodal images arrive as
+                ``Decoded`` variants over NIXL RDMA from the Rust frontend
+                and must be read+converted to PIL before passing to SGLang.
+                Off by default; the worker keeps the URL-string fast path.
         """
         super().__init__(
             engine,
@@ -180,12 +187,18 @@ def __init__(
         self._routed_experts_kwargs: Dict[
             str, Any
         ] = self._resolve_routed_experts_kwargs(self.engine, self.config.server_args)
+        self._enable_frontend_decoding = enable_frontend_decoding
+        self._image_loader: Optional[ImageLoader] = None
+        if self._enable_frontend_decoding:
+            # Lazy-inits a NIXL connector internally for Decoded variants.
+            self._image_loader = ImageLoader(enable_frontend_decoding=True)
         if self.serving_mode == DisaggregationMode.DECODE:
             logging.info(
                 "Decode worker handler initialized (disaggregated decode mode)"
             )
         else:
-            logging.info("Decode worker handler initialized (aggregated mode)")
+            mode = "frontend-decoded" if self._enable_frontend_decoding else "standard"
+            logging.info(f"Decode worker handler initialized (aggregated mode, {mode})")
 
     @staticmethod
     def _resolve_routed_experts_kwargs(engine: Any, server_args: Any) -> Dict[str, Any]:
@@ -492,9 +505,22 @@ async def generate(
             # Extract image/video URLs for multimodal requests. SGLang's mm_data_processor
             # handles loading/preprocessing, and the scheduler does vision encoding.
             mm_data = request.get("multi_modal_data", {})
-            image_data = _extract_media_urls(mm_data, "image_url")
             video_data = _extract_media_urls(mm_data, "video_url")
 
+            image_data: list[str] | list[PILImage] | None
+            if self._enable_frontend_decoding:
+                # Invariant from __init__: _image_loader is non-None iff
+                # _enable_frontend_decoding is True. Assert narrows the
+                # Optional for the type checker without runtime branching.
+                assert self._image_loader is not None
+                image_items = mm_data.get("image_url") or []
+                if image_items:
+                    image_data = await self._image_loader.load_image_batch(image_items)
+                else:
+                    image_data = None
+            else:
+                image_data = _extract_media_urls(mm_data, "image_url")
+
             trace_header = build_trace_headers(context) if self.enable_trace else None
 
             # Extract dp_rank from routing info (set by KV router)