Skip to content
Merged
Show file tree
Hide file tree
Changes from 6 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
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,26 @@ def test_raises_for_audio_url(self):
def test_text_only_request_bypasses_guard(self):
raise_if_unextracted_multimodal({"token_ids": [10, 20, 30]})

def test_stripped_extra_args_still_uses_multi_modal_data(self):
data_url = "data:image/png;base64,AAAA"
request = {
"token_ids": [1, 2, 3],
"multi_modal_data": {"image_url": [{"Url": data_url}]},
"extra_args": {
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "describe"},
{"type": "image_url", "image_url": {"url": ""}},
],
}
]
},
}
raise_if_unextracted_multimodal(request)
assert build_disagg_mm_kwargs(request)["image_data"] == [data_url]


async def _stream(items):
for item in items:
Expand Down
6 changes: 1 addition & 5 deletions components/src/dynamo/trtllm/encode_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -409,15 +409,11 @@ async def process_encode_request(
yield {"error": "No multimodal_processor configured on encode worker"}
return

# Extract messages and determine which flow to use
messages = request.get("extra_args", {}).get(
"messages", request.get("messages", [])
)
(
_,
image_urls,
embedding_paths,
) = multimodal_processor.extract_prompt_and_media(messages)
) = multimodal_processor.extract_prompt_and_media_from_request(request)

# Flow 1: Embedding-path flow (pre-computed embeddings via NIXL)
if embedding_paths:
Expand Down
10 changes: 10 additions & 0 deletions components/src/dynamo/trtllm/multimodal/embedding_fetcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,16 @@ def _create_request_with_urls(

modified_request = copy.deepcopy(original_request)

mm_data = modified_request.get("multi_modal_data")
if isinstance(mm_data, dict) and isinstance(mm_data.get("image_url"), list):
filtered_items = []
for item in mm_data["image_url"]:
if isinstance(item, dict) and item.get("Url") in image_urls:
filtered_items.append(item)
elif isinstance(item, str) and item in image_urls:
filtered_items.append(item)
mm_data["image_url"] = filtered_items

# Extract messages
messages = modified_request.get("extra_args", {}).get(
"messages", modified_request.get("messages", [])
Expand Down
61 changes: 59 additions & 2 deletions components/src/dynamo/trtllm/multimodal_processor.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,41 @@ def resolve_mm_processor_kwargs(request: Dict[str, Any]) -> Optional[Dict[str, A
return mm_kwargs


def _is_safetensors_url(url: str) -> bool:
"""True when the URL path (not query) ends with ``.safetensors``."""
return urlparse(url).path.lower().endswith(".safetensors")


def _urls_from_multi_modal_items(
items: Any,
) -> Tuple[List[str], List[str]]:
"""Split ``multi_modal_data`` image items into image URLs and embedding paths."""
Comment thread
indrajit96 marked this conversation as resolved.
Resolved
image_urls: List[str] = []
embedding_paths: List[str] = []
if not isinstance(items, list):
return image_urls, embedding_paths
for item in items:
if isinstance(item, dict) and isinstance(item.get("Url"), str):
url = item["Url"]
elif isinstance(item, str):
url = item
else:
continue
if not url:
continue
if _is_safetensors_url(url):
embedding_paths.append(url)
else:
image_urls.append(url)
return image_urls, embedding_paths


def request_messages(request: Dict[str, Any]) -> List[Dict]:
extra_args = request.get("extra_args") or {}
messages = extra_args.get("messages") or request.get("messages") or []
return messages if isinstance(messages, list) else []


class MultimodalRequestProcessor:
"""Simple processor for OpenAI format multimodal requests."""

Expand Down Expand Up @@ -303,13 +338,35 @@ def extract_prompt_and_media(
if not url:
continue
self.modality = "image"
if url.endswith(".safetensors"):
if _is_safetensors_url(url):
embedding_paths.append(url)
else:
image_urls.append(url)

return "".join(text_parts), image_urls, embedding_paths

def extract_prompt_and_media_from_request(
self, request: Dict[str, Any]
) -> Tuple[str, List[str], List[str]]:
"""Extract text and media URLs, preferring ``multi_modal_data``.

The frontend strips inline ``data:`` payloads from
``extra_args.messages`` so the request plane carries a single copy of
the media in ``multi_modal_data``. Chat-template structure still lives
in ``extra_args.messages``.
"""
text, image_urls, embedding_paths = self.extract_prompt_and_media(
request_messages(request)
)
mm_data = request.get("multi_modal_data")
if isinstance(mm_data, dict):
mm_urls, mm_emb = _urls_from_multi_modal_items(mm_data.get("image_url"))
if mm_urls:
image_urls = mm_urls
if mm_emb:
embedding_paths = mm_emb
return text, image_urls, embedding_paths

async def process_openai_request(
self, request: Dict, embeddings: Any, ep_disaggregated_params: Any
) -> Optional[Any]:
Expand Down Expand Up @@ -422,7 +479,7 @@ async def process_openai_request(
)
continue

if url.endswith(".safetensors"):
if _is_safetensors_url(url):
embedding_paths.append(url)
else:
# Keep original item format for load_image_batch
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,11 @@ async def generate(
embeddings: Optional[Union[torch.Tensor, dict]] = None
ep_disaggregated_params = None
if self.multimodal_processor and self.encode_client:
messages = request.get("extra_args", {}).get(
"messages", request.get("messages", [])
)
_, image_urls, _ = self.multimodal_processor.extract_prompt_and_media(
messages
)
(
_,
image_urls,
_,
) = self.multimodal_processor.extract_prompt_and_media_from_request(request)
if image_urls:
result = await fetch_embeddings_from_encoder(
image_urls,
Expand Down
6 changes: 1 addition & 5 deletions components/src/dynamo/trtllm/request_handlers/handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -158,15 +158,11 @@ async def generate(
ep_disaggregated_params = None

if self.multimodal_processor:
# Extract messages from extra_args (set by Rust preprocessor) or fall back to direct field
messages = request.get("extra_args", {}).get(
"messages", request.get("messages", [])
)
(
_,
image_urls,
embedding_paths,
) = self.multimodal_processor.extract_prompt_and_media(messages)
) = self.multimodal_processor.extract_prompt_and_media_from_request(request)
# Handle embedding paths (NIXL transfer of pre-computed embeddings)
if embedding_paths:
if self.encode_client and self.connector:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ def create_mock_context(request_id: str = "test-id") -> MagicMock:
def setup_multimodal_config(config: MagicMock, image_urls: List[str]) -> None:
"""Configure multimodal_processor and encode_client on config."""
config.multimodal_processor = MagicMock()
config.multimodal_processor.extract_prompt_and_media = MagicMock(
config.multimodal_processor.extract_prompt_and_media_from_request = MagicMock(
return_value=("text", image_urls, [])
)
config.encode_client = MagicMock()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -455,7 +455,7 @@ async def test_epd_non_object_kwargs_raises_client_error() -> None:
# URLs come from the processor, not the request; the engine must report an
# available encoder, or the flow short-circuits before the kwargs check.
processor = MagicMock()
processor.extract_prompt_and_media.return_value = (
processor.extract_prompt_and_media_from_request.return_value = (
"describe",
["http://example.invalid/a.png"],
[],
Expand Down Expand Up @@ -542,3 +542,114 @@ async def _encode(_req):

assert excinfo.value.status == 400
cache.get.assert_not_called()


def test_extract_prompt_and_media_from_request_uses_multi_modal_data() -> None:
"""Frontend-stripped extra_args.messages still resolve images from multi_modal_data."""
processor = MultimodalRequestProcessor(
model_type="multimodal",
model_dir="unused",
max_file_size_mb=10,
tokenizer=MagicMock(),
)
data_url = "data:image/png;base64,AAAA"
https_url = "https://example.com/img.png"
text, image_urls, embedding_paths = processor.extract_prompt_and_media_from_request(
{
"extra_args": {
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "describe"},
{"type": "image_url", "image_url": {"url": ""}},
{
"type": "image_url",
"image_url": {"url": https_url},
},
],
}
]
},
"multi_modal_data": {"image_url": [{"Url": data_url}, {"Url": https_url}]},
}
)
assert text == "describe"
assert image_urls == [data_url, https_url]
assert embedding_paths == []


def test_extract_prompt_and_media_from_request_classifies_signed_safetensors() -> None:
processor = MultimodalRequestProcessor(
model_type="multimodal",
model_dir="unused",
max_file_size_mb=10,
tokenizer=MagicMock(),
)
signed = "https://host/embedding.SAFETENSORS?sig=abc"
_, image_urls, embedding_paths = processor.extract_prompt_and_media_from_request(
{
"extra_args": {"messages": []},
"multi_modal_data": {"image_url": [{"Url": signed}]},
}
)
assert image_urls == []
assert embedding_paths == [signed]


def test_extract_prompt_and_media_from_request_keeps_message_fallback() -> None:
processor = MultimodalRequestProcessor(
model_type="multimodal",
model_dir="unused",
max_file_size_mb=10,
tokenizer=MagicMock(),
)
url = "https://example.com/legacy.png"
_, image_urls, _ = processor.extract_prompt_and_media_from_request(
{
"extra_args": {
"messages": [
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": url}},
],
}
]
}
}
)
assert image_urls == [url]


def test_extract_prompt_and_media_from_request_empty_or_malformed_mm_data() -> None:
processor = MultimodalRequestProcessor(
model_type="multimodal",
model_dir="unused",
max_file_size_mb=10,
tokenizer=MagicMock(),
)
fallback = "https://example.com/fallback.png"
request = {
"extra_args": {
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "hi"},
{"type": "image_url", "image_url": {"url": fallback}},
],
}
]
}
}
for mm_data in ({}, {"image_url": "not-a-list"}, {"image_url": [None, 1, {}]}):
request["multi_modal_data"] = mm_data
(
text,
image_urls,
embedding_paths,
) = processor.extract_prompt_and_media_from_request(request)
assert text == "hi"
assert image_urls == [fallback]
assert embedding_paths == []
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,13 @@ This page describes multimodal inference with the Dynamo vLLM backend.
| **HTTP/HTTPS** | `https://example.com/image.jpg` | Remote media files |
| **Data URL** | `data:image/jpeg;base64,/9j/4AAQ...` | Base64-encoded inline data |

> [!NOTE]
> Inline `data:` media is forwarded once on the request plane in `multi_modal_data`.
> The frontend strips the inline payload from `extra_args.messages` so the worker
> frame is not doubled. A serialized frame that still exceeds
> `DYN_TCP_MAX_MESSAGE_SIZE` (32 MiB by default) returns HTTP 400. See
> [Frontend Configuration](../../../../../reference/components/frontend-configuration.mdx).

> [!NOTE]
> Media URLs are validated against a default-deny policy. `https://` and `data:` sources
> pass; plain `http://` and hostnames that resolve to private or loopback addresses are
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,17 @@ see the [TLS reference](tls-configuration.mdx#http-tls-and-mtls).
The Rust HTTP server also reads these environment variables, which are not exposed as CLI arguments:

<ParamField path="DYN_HTTP_BODY_LIMIT_MB" type="integer" default="192">
Maximum request body size in MB.
Maximum HTTP request body size in MB. This is the client-facing ingress cap.
After preprocessing, the frontend sends the worker-bound frame over the request
plane. On the default TCP plane that frame is independently limited by
`DYN_TCP_MAX_MESSAGE_SIZE` (32 MiB by default). Inline `data:` media is
forwarded once in `multi_modal_data`; a serialized frame that still exceeds the
TCP cap is rejected with HTTP 400 before the connection is written. Raise
`DYN_TCP_MAX_MESSAGE_SIZE` together with this value when operators need larger
inline-media requests, or keep HTTP images as `https://` URLs so only a
reference crosses the request plane.

See [Runtime Configuration](runtime-configuration.mdx) for `DYN_TCP_MAX_MESSAGE_SIZE`.
</ParamField>

<ParamField path="DYN_HTTP_GRACEFUL_SHUTDOWN_TIMEOUT_SECS" type="integer" default="5">
Expand Down
12 changes: 12 additions & 0 deletions docs/fern/pages/reference/components/runtime-configuration.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,18 @@ Unless a field is marked environment-only, it has both a CLI flag and an environ
<span className="enum-values"><span className="enum-label">Allowed values:</span> <Badge intent="note" minimal>json</Badge> <Badge intent="note" minimal>msgpack</Badge></span>
</ParamField>

<ParamField path="DYN_TCP_MAX_MESSAGE_SIZE" type="integer" default="33554432">
Maximum TCP request-plane frame size in bytes (default 32 MiB). The frontend
rejects an oversized frame with HTTP 400 before writing it. This limit is
independent of `DYN_HTTP_BODY_LIMIT_MB` (default 192 MB) on the HTTP ingress:
a chat-completions body can be accepted by the HTTP server and still fail at
this cap after serialization. Inline `data:` media is sent once in
`multi_modal_data`; raise this value when operators must accept larger
inline-media worker frames.

See [Frontend Configuration](frontend-configuration.mdx) for `DYN_HTTP_BODY_LIMIT_MB`.
</ParamField>

<ParamField path="DYN_TCP_RESPONSE_STREAM_HOST" type="string" default="auto-detected local address">
IPv4 or IPv6 address, or exact network interface name, used to bind and advertise the TCP response
stream server. If an interface name occurs more than once in the operating system's interface
Expand Down
Loading
Loading