Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion components/src/dynamo/frontend/prepost.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,10 @@
from functools import lru_cache, partial
from typing import TYPE_CHECKING, Any, Protocol, TypeGuard, cast

from vllm.entrypoints.chat_utils import make_tool_call_id
try:
from vllm.renderers.chat_utils import make_tool_call_id
except ImportError: # vLLM before vllm-project/vllm#60522 moved chat_utils
from vllm.entrypoints.chat_utils import make_tool_call_id
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionNamedFunction,
ChatCompletionNamedToolChoiceParam,
Expand Down
7 changes: 6 additions & 1 deletion components/src/dynamo/frontend/vllm_processor.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,12 @@

from msgspec.structs import replace as msgspec_replace
from vllm.config import CacheConfig, LoadConfig, ModelConfig, VllmConfig
from vllm.entrypoints.chat_utils import load_chat_template

try:
from vllm.renderers.chat_utils import load_chat_template
except ImportError: # vLLM before vllm-project/vllm#60522 moved chat_utils
from vllm.entrypoints.chat_utils import load_chat_template

from vllm.exceptions import VLLMClientError
from vllm.reasoning import ReasoningParser, ReasoningParserManager
from vllm.sampling_params import RequestOutputKind, SamplingParams
Expand Down
86 changes: 71 additions & 15 deletions components/src/dynamo/trtllm/multimodal_processor.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
from safetensors.torch import load as safetensors_load
from safetensors.torch import load_file as safetensors_load_file
from tensorrt_llm.inputs.multimodal_data import VideoData
from tensorrt_llm.inputs.utils import async_load_video
from tensorrt_llm.inputs.utils import async_load_video, load_video
from tensorrt_llm.llmapi.tokenizer import tokenizer_factory

from dynamo.common.http import HttpConfigurationError, HttpStatusError, fetch_bytes
Expand Down Expand Up @@ -58,19 +58,46 @@
_EMBEDDING_FETCH_READ_TIMEOUT_S = 300.0


def _nvdec_video_data(content: bytes, num_frames: int) -> VideoData:
"""Decode H.264/H.265 via NVDEC into TRT-LLM's ``VideoData`` (format="pt").
def _preferred_video_format(input_processor: Any) -> str:
"""The input processor's preferred video decode format: "np" or "pt".

Mirrors ``tensorrt_llm.inputs.media_io._load_video_by_cv2``'s "pt" output for
the pinned v1.3.0rc21: RGB, a list of ``(C, H, W)`` float32 [0,1] tensors,
with the same metadata keys. Synchronous -- call via ``asyncio.to_thread``.
trtllm-serve seeds its media-IO defaults from
``get_preferred_media_io_kwargs()``. MiniMax-M3 VL asks for uint8 "np"
frames because its video processor rescales by 1/255 itself; float [0,1]
"pt" frames would be rescaled twice and reach the vision tower near black.
Anything other than "np" keeps the "pt" frames.
"""
get_prefs = getattr(input_processor, "get_preferred_media_io_kwargs", None)
if get_prefs is None:
return "pt"
try:
prefs = get_prefs() or {}
except Exception as e:
logging.warning("Preferred media IO kwargs unavailable: %s", e)
return "pt"
return "np" if (prefs.get("video") or {}).get("format") == "np" else "pt"


def _nvdec_video_data(
content: bytes, num_frames: int, video_format: str = "pt"
) -> VideoData:
"""Decode H.264/H.265 via NVDEC into TRT-LLM's ``VideoData``.

Mirrors ``tensorrt_llm.inputs.media_io._load_video_by_cv2``: "pt" is a list
of ``(C, H, W)`` float32 [0,1] tensors, "np" the ``(N, H, W, 3)`` uint8 RGB
stack, with the same metadata keys. Synchronous -- call via
``asyncio.to_thread``.
"""
from dynamo.common.multimodal.nvdec_decoder import decode_video_nvdec

frames_np, meta = decode_video_nvdec(content, num_frames) # (N,H,W,3) uint8 RGB
stacked = frames_np.astype("float32") * (1.0 / 255.0)
nchw = torch.from_numpy(stacked).permute(0, 3, 1, 2).contiguous()
frames_pt = list(torch.unbind(nchw, dim=0))
frames: Any
if video_format == "np":
frames = frames_np
else:
stacked = frames_np.astype("float32") * (1.0 / 255.0)
nchw = torch.from_numpy(stacked).permute(0, 3, 1, 2).contiguous()
frames = list(torch.unbind(nchw, dim=0))
fps = float(meta.get("fps") or 0.0)
total = int(meta["total_num_frames"])
metadata = {
Expand All @@ -79,7 +106,7 @@ def _nvdec_video_data(content: bytes, num_frames: int) -> VideoData:
"duration": (total / fps) if fps > 0 else 0.0,
"frames_indices": meta["frames_indices"],
}
return VideoData(frames=frames_pt, metadata=metadata, audio=None)
return VideoData(frames=frames, metadata=metadata, audio=None)


class TokenizerProtocol(Protocol):
Expand Down Expand Up @@ -192,6 +219,7 @@ def __init__(
self.input_processor = create_input_processor(model_dir, self.tokenizer)
except Exception as e:
logging.warning("Input processor unavailable for max_tokens sizing: %s", e)
self.video_format = _preferred_video_format(self.input_processor)

def _expanded_prompt_len(
self, token_ids: List[int], images: Optional[List[Any]]
Expand All @@ -217,6 +245,18 @@ def _expanded_prompt_len(
logging.warning("Could not compute expanded prompt length: %s", e)
return None

async def _load_video_with_vendor_decoder(self, video: str) -> VideoData:
"""Decode with TRT-LLM's cv2 loader in ``self.video_format``.

``async_load_video`` accepts only "pt" and "pil", so "np" goes through
the synchronous ``load_video`` in a worker thread.
"""
if self.video_format == "np":
return await asyncio.to_thread(
load_video, video, self.num_video_frames, format="np"
)
return await async_load_video(video, self.num_video_frames)

def is_url(self, path: str) -> bool:
"""Check if a path is a URL."""
parsed = urlparse(path)
Expand Down Expand Up @@ -600,7 +640,10 @@ async def process_openai_request(
if should_use_nvdec(codec):
try:
nvdec_video = await asyncio.to_thread(
_nvdec_video_data, content, self.num_video_frames
_nvdec_video_data,
content,
self.num_video_frames,
self.video_format,
)
except Exception as exc: # noqa: BLE001 - fall back
logging.warning(
Expand All @@ -618,8 +661,8 @@ async def process_openai_request(
await asyncio.to_thread(video_file.flush)
try:
videos.append(
await async_load_video(
video_file.name, self.num_video_frames
await self._load_video_with_vendor_decoder(
video_file.name
)
)
except ImportError as exc:
Expand All @@ -638,8 +681,8 @@ async def process_openai_request(
else:
try:
videos.append(
await async_load_video(
normalized_url, self.num_video_frames
await self._load_video_with_vendor_decoder(
normalized_url
)
)
except ImportError as exc:
Expand Down Expand Up @@ -712,6 +755,19 @@ async def process_openai_request(
# Post-expansion prompt length, so an omitted max_tokens can be sized
# against the real context usage rather than the unexpanded placeholders.
mm_data = processed_inputs.get("multi_modal_data")

# The token IDs are the frontend's rendering of the chat template, one
# placeholder per media item already in place. A TRT-LLM processor
# without the tokenized+MM fast path gets them detokenized by TRT-LLM,
# which drops `chat_template_applied`, so a processor that templates a
# raw prompt (MiniMax-M3 VL) adds a second placeholder and rejects the
# request. Hand such processors the rendered text with the flag, as
# trtllm-serve does; fast-path processors keep the token IDs alone.
if mm_data and not getattr(
self.input_processor, "supports_token_id_mm_expansion", True
):
processed_inputs["prompt"] = self.tokenizer.decode(token_ids)
processed_inputs["chat_template_applied"] = True
# Skipped when the request overrides the processor: the sizing
# calculator is not override-aware (Qwen2-VL ignores the kwargs while
# counting) and is not guaranteed non-mutating (Gemma-4 writes them
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1142,6 +1142,12 @@ async def _generate_locally_impl(
sampling_params = self._override_sampling_params(
self.default_sampling_params, request
)
# A rendered prompt already carries its special tokens; TRT-LLM's own
# detokenize path clears this flag for the same reason.
if isinstance(processed_input, dict) and processed_input.get(
"chat_template_applied"
):
sampling_params.add_special_tokens = False

# Additional sampling params in output options
output_options = request.get("output_options", {})
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -188,6 +188,94 @@ def test_nvdec_video_data_builds_pt_videodata(monkeypatch) -> None:
assert vd.audio is None


def test_nvdec_video_data_keeps_uint8_frames_for_np(monkeypatch) -> None:
""" "np" hands over NVDEC's (T,H,W,3) uint8 stack untouched, as the vendor
loader's "np" format does; processors that rescale by 1/255 need it."""
import numpy as np

from dynamo.common.multimodal import nvdec_decoder

frames_np = np.full((2, 4, 5, 3), 200, dtype=np.uint8)
meta = {"total_num_frames": 2, "fps": 1.0, "frames_indices": [0, 1]}
monkeypatch.setattr(
nvdec_decoder,
"decode_video_nvdec",
lambda content, num_frames: (frames_np, meta),
)

vd = mmp._nvdec_video_data(b"h264-bytes", 2, "np")

assert vd.frames is frames_np
assert vd.metadata["frames_indices"] == [0, 1]


@pytest.mark.parametrize(
"input_processor, expected",
[
(None, "pt"),
(SimpleNamespace(), "pt"),
(SimpleNamespace(get_preferred_media_io_kwargs=lambda: {}), "pt"),
(
SimpleNamespace(
get_preferred_media_io_kwargs=lambda: {"image": {"format": "np"}}
),
"pt",
),
(
SimpleNamespace(
get_preferred_media_io_kwargs=lambda: {"video": {"format": "np"}}
),
"np",
),
],
)
def test_preferred_video_format(input_processor, expected) -> None:
assert mmp._preferred_video_format(input_processor) == expected


@pytest.mark.asyncio
@pytest.mark.parametrize("codec", ["h264", "vp9"])
async def test_video_decodes_in_the_preferred_format(codec, monkeypatch) -> None:
"""With "np" preferred, NVDEC is asked for "np" and the vendor fallback goes
through load_video(format="np"), since async_load_video rejects "np"."""
monkeypatch.setenv("DYN_MM_ALLOW_INTERNAL", "1")
processor = MultimodalRequestProcessor(
model_type="multimodal",
model_dir="unused",
max_file_size_mb=10,
tokenizer=MagicMock(),
)
processor.video_format = "np"
monkeypatch.setattr(mmp, "fetch_bytes", AsyncMock(return_value=b"video bytes"))
monkeypatch.setattr(mmp, "probe_video_codec", lambda content: codec)
monkeypatch.setattr(mmp, "should_use_nvdec", lambda codec: codec == "h264")
nvdec = MagicMock(return_value=object())
monkeypatch.setattr(mmp, "_nvdec_video_data", nvdec)
sync_load = MagicMock(return_value=object())
monkeypatch.setattr(mmp, "load_video", sync_load)
async_load = AsyncMock(return_value=object())
monkeypatch.setattr(mmp, "async_load_video", async_load)

await processor.process_openai_request(
{
"multi_modal_data": {
"video_url": [{"Url": "http://169.254.169.254/v.mp4"}]
},
"token_ids": [1],
},
embeddings=None,
ep_disaggregated_params=None,
)

async_load.assert_not_awaited()
if codec == "h264":
assert nvdec.call_args.args[2] == "np"
sync_load.assert_not_called()
else:
nvdec.assert_not_called()
assert sync_load.call_args.kwargs == {"format": "np"}


@pytest.mark.asyncio
async def test_h264_video_routes_through_nvdec(monkeypatch) -> None:
"""H.264 video input is hardware-decoded via NVDEC; the vendor
Expand Down Expand Up @@ -529,6 +617,45 @@ async def test_mm_processor_kwargs_forwarded(request_extra, expected) -> None:
assert processed["mm_processor_kwargs"] == expected


@pytest.mark.asyncio
@pytest.mark.parametrize("fast_path", [False, True])
async def test_rendered_prompt_reaches_processors_without_the_token_fast_path(
fast_path,
) -> None:
"""Without the tokenized+MM fast path, media requests carry the rendered
text and chat_template_applied; with it, the token IDs alone."""
tokenizer = MagicMock()
tokenizer.decode.return_value = "rendered <image> prompt"
processor = MultimodalRequestProcessor(
model_type="multimodal",
model_dir="unused",
max_file_size_mb=10,
tokenizer=tokenizer,
)
processor.input_processor = SimpleNamespace(
supports_token_id_mm_expansion=fast_path
)
processor.image_loader.load_image_batch = AsyncMock(return_value=[object()])

processed = await processor.process_openai_request(
{
"multi_modal_data": {"image_url": [{"Url": "https://example.com/x.png"}]},
"token_ids": [1, 2, 3],
},
embeddings=None,
ep_disaggregated_params=None,
)

assert processed["prompt_token_ids"] == [1, 2, 3]
if fast_path:
assert "prompt" not in processed
assert "chat_template_applied" not in processed
else:
assert processed["prompt"] == "rendered <image> prompt"
assert processed["chat_template_applied"] is True
tokenizer.decode.assert_called_once_with([1, 2, 3])


@pytest.mark.asyncio
async def test_mm_processor_kwargs_non_object_is_rejected() -> None:
"""A non-object value is a client error, not a silently ignored field."""
Expand Down
Loading
Loading