@tduncan / Nvidia-Optimization-Stack-VLLM / commits / b30291cf0c

[EPD] Remove duplicate image preprocessing in EPD and enable preprocess on GPU (#50390)

Signed-off-by: Tianyu Guo <guoty@inferact.ai>

Tianyu Guo committed Aug 6, 2026 at 06:59 UTC b30291cf0c9a15898589f0347906bad07f3b538b
22 files changed +903 -44
examples/disaggregated/disaggregated_encoder/disagg_epd_proxy.py
+162 -6
@@ -21,13 +21,18 @@ from __future__ import annotations
21
22 import argparse
23 import asyncio
24 +import hashlib
25 +import io
26 +import json
27 import logging
28 import os
29 import random
30 +import time
31 import uuid
32 from collections.abc import AsyncIterator
33
34 import aiohttp
35 +import pybase64 as base64
36 import uvicorn
37 from fastapi import FastAPI, HTTPException, Request
38 from fastapi.responses import JSONResponse, StreamingResponse
@@ -53,6 +58,89 @@ decode_session: aiohttp.ClientSession | None = None
58
59 MM_TYPES = {"image_url", "audio_url", "input_audio"}
60
61 +# Diagnostic switch: forward the original request to the decoder so the
62 +# only difference from the rewrite path is the rewrite itself.
63 +NO_REWRITE = False
64 +
65 +
66 +# Grid metadata reported by the encoder instance, keyed by item index.
67 +# Empty when the encoder did not report any (then nothing is rewritten).
68 +def content_uuid(item: dict) -> str:
69 + """Cache key for a multimodal item, derived from its content.
70 +
71 + Must be content-derived, not request-derived: the EC cache is keyed by this
72 + value, so a per-request key (a request id, say) would make every request a
73 + miss and throw away cross-request reuse of already-encoded media -- while
74 + the unmodified path, which hashes the content, would keep it. That asymmetry
75 + silently biases any comparison between the two.
76 + """
77 + url = (item.get("image_url") or item.get("audio_url") or {}).get("url") or ""
78 + payload = url or json.dumps(item, sort_keys=True)
79 + return hashlib.sha256(payload.encode()).hexdigest()
80 +
81 +
82 +def _b64_tensor(values: list) -> str:
83 + import torch
84 +
85 + buf = io.BytesIO()
86 + grid = torch.tensor(values, dtype=torch.long)
87 + # Downstream stacks per item, so hand over a flat (t, h, w).
88 + torch.save(grid.reshape(-1)[:3], buf)
89 + return base64.b64encode(buf.getvalue()).decode()
90 +
91 +
92 +def rewrite_for_decode(req_data: dict, item_meta: dict[int, dict]) -> dict:
93 + """Replace each image item with a metadata-only reference for the decoder.
94 +
95 + The decoder does not need the pixels: the encoder instance already produced
96 + the embedding and published it through the EC connector under the same uuid.
97 + Sending only the grid lets the decoder size the placeholder range without
98 + re-running the image transform.
99 +
100 + `item_meta` holds what the encoder reported for each item (its cache key and
101 + the grid its processor actually produced), so the grid is never re-derived
102 + here -- a second derivation could disagree with the encoder's.
103 + """
104 + rewritten = 0
105 + idx = 0
106 + new_messages = []
107 + for msg in req_data.get("messages", []):
108 + content = msg.get("content")
109 + if not isinstance(content, list):
110 + new_messages.append(msg)
111 + continue
112 + new_content = []
113 + for item in content:
114 + if item.get("type") not in MM_TYPES:
115 + new_content.append(item)
116 + continue
117 + meta = dict(item_meta.get(idx) or {})
118 + idx += 1
119 + item_uuid = meta.pop("mm_hash", None)
120 + # Whatever keys the encoder reported are the metadata its model
121 + # declared as needed to size the placeholder range; the proxy does
122 + # not need to know their names.
123 + metadata = {k: _b64_tensor(v) for k, v in meta.items()}
124 + if not metadata or not item_uuid:
125 + # The encoder reported no metadata (e.g. the item came from its
126 + # processor cache); let the decoder process the media itself.
127 + new_content.append(item)
128 + continue
129 + new_content.append(
130 + {
131 + "type": "image_embeds",
132 + "image_embeds": metadata,
133 + "uuid": item_uuid,
134 + }
135 + )
136 + rewritten += 1
137 + new_messages.append({**msg, "content": new_content})
138 +
139 + if not rewritten:
140 + return req_data
141 + logger.info("Rewrote %d image item(s) as metadata references", rewritten)
142 + return {**req_data, "messages": new_messages}
143 +
144
145 def extract_mm_items(request_data: dict) -> list[dict]:
146 """
@@ -77,22 +165,29 @@ async def fanout_encoder_primer(
165 orig_request: dict,
166 e_urls: list[str],
167 req_id: str,
80 -) -> None:
168 +) -> dict[int, dict]:
169 """
170 1. Build one request *per MM item* with all text removed.
171 2. Send them concurrently to the encode cluster.
172 3. Raise if any of them fails.
173 +
174 + Returns, per item index, the metadata the encoder reported in
175 + `ec_transfer_params`: its EC cache key and the grid its processor produced.
176 + The proxy still supplies the uuid so both sides key the cache the same way;
177 + the grid can only come from the encoder, which is the side that computed it.
178 """
179 logger.info("[%s] Processing multimodal items...", req_id)
180
181 mm_items = extract_mm_items(orig_request)
182 if not mm_items:
183 logger.info("[%s] No multimodal items, skipping encoder", req_id)
91 - return # nothing to do
184 + return {} # nothing to do
185
186 logger.info("[%s] got %d multimodal items...", req_id, len(mm_items))
187
188 tasks = []
189 + item_uuids: dict[int, str] = {}
190 + item_meta: dict[int, dict] = {}
191
192 # Round-robin over encode servers to distribute load a bit
193 url_cycle = (e_urls[i % len(e_urls)] for i in range(len(mm_items)))
@@ -102,11 +197,24 @@ async def fanout_encoder_primer(
197 child_req_id = f"{req_id}:{idx}:{uuid.uuid4().hex[:6]}"
198 headers = {"x-request-id": child_req_id}
199
200 + # With --no-rewrite the decoder still receives the raw image and derives
201 + # the cache key by hashing it, so the encoder must do the same -- passing
202 + # a uuid here would make the two disagree and silently defeat the EC
203 + # transfer, leaving the decoder to encode the image itself.
204 + item_uuid = None if NO_REWRITE else content_uuid(item)
205 + if item_uuid is not None:
206 + item_uuids[idx] = item_uuid
207 +
208 encoder_req = {
209 # You *may* need to keep additional fields
210 "model": orig_request.get("model"),
211 "messages": [
109 - {"role": "user", "content": [item]},
212 + {
213 + "role": "user",
214 + "content": [
215 + item if item_uuid is None else {**item, "uuid": item_uuid}
216 + ],
217 + },
218 ],
219 # No max_tokens cap: the encoder instance never samples, it finishes
220 # once the prompt is encoded and its embeddings are published.
@@ -152,9 +260,21 @@ async def fanout_encoder_primer(
260 detail=f"Encoder request failed: {detail}",
261 )
262
263 + # The encoder reports each item's cache key and grid here.
264 + try:
265 + params = (await r.json()).get("ec_transfer_params") or {}
266 + reported = params.get("ec_items") or []
267 + except Exception:
268 + logger.warning("[%s] Could not read encoder metadata #%d", req_id, idx)
269 + reported = []
270 + if reported and idx in item_uuids:
271 + # One item per encoder request, so the first entry is this item's.
272 + item_meta[idx] = {**reported[0], "mm_hash": item_uuids[idx]}
273 +
274 logger.info(
275 "[%s] All %d encoder requests completed successfully", req_id, len(mm_items)
276 )
277 + return item_meta
278
279
280 async def maybe_prefill(
@@ -320,7 +440,11 @@ async def forward_non_stream(
440 ) -> dict:
441 try:
442 # Step 1: Process through Encoder instance (if has MM input)
323 - await fanout_encoder_primer(req_data, e_urls, req_id)
443 + _t0 = time.perf_counter()
444 + item_meta = await fanout_encoder_primer(req_data, e_urls, req_id)
445 + _t1 = time.perf_counter()
446 + req_data = req_data if NO_REWRITE else rewrite_for_decode(req_data, item_meta)
447 + _t2 = time.perf_counter()
448
449 # Step 2: Process through Prefill instance
450 req_data = await maybe_prefill(req_data, p_url, req_id)
@@ -334,7 +458,17 @@ async def forward_non_stream(
458 f"{d_url}/v1/chat/completions", json=req_data, headers=headers
459 ) as resp:
460 resp.raise_for_status()
337 - return await resp.json()
461 + out = await resp.json()
462 + _t3 = time.perf_counter()
463 + logger.info(
464 + "STAGE %s encode=%.1f rewrite=%.1f decode=%.1f total=%.1f",
465 + "no-rewrite" if NO_REWRITE else "rewrite",
466 + (_t1 - _t0) * 1e3,
467 + (_t2 - _t1) * 1e3,
468 + (_t3 - _t2) * 1e3,
469 + (_t3 - _t0) * 1e3,
470 + )
471 + return out
472
473 except HTTPException:
474 raise
@@ -348,7 +482,11 @@ async def forward_stream(
482 ) -> AsyncIterator[str]:
483 try:
484 # Step 1: Process through Encoder instance (if has MM input)
351 - await fanout_encoder_primer(req_data, e_urls, req_id)
485 + _t0 = time.perf_counter()
486 + item_meta = await fanout_encoder_primer(req_data, e_urls, req_id)
487 + _t1 = time.perf_counter()
488 + req_data = req_data if NO_REWRITE else rewrite_for_decode(req_data, item_meta)
489 + _t2 = time.perf_counter()
490
491 # Step 2: Process through Prefill instance
492 req_data = await maybe_prefill(req_data, p_url, req_id)
@@ -358,6 +496,7 @@ async def forward_stream(
496 headers = {"x-request-id": req_id}
497
498 # Streaming response
499 + _first = None
500 async with decode_session.post(
501 f"{d_url}/v1/chat/completions",
502 json=req_data,
@@ -366,8 +505,19 @@ async def forward_stream(
505 resp.raise_for_status()
506 async for chunk in resp.content.iter_chunked(1024):
507 if chunk:
508 + if _first is None:
509 + _first = time.perf_counter()
510 yield chunk.decode("utf-8", errors="ignore")
511 + _t3 = time.perf_counter()
512
513 + logger.info(
514 + "STAGE %s encode=%.1f rewrite=%.2f decode_ttfb=%.1f decode_total=%.1f",
515 + "no-rewrite" if NO_REWRITE else "rewrite",
516 + (_t1 - _t0) * 1e3,
517 + (_t2 - _t1) * 1e3,
518 + ((_first or _t3) - _t2) * 1e3,
519 + (_t3 - _t2) * 1e3,
520 + )
521 logger.info("[%s] Streaming completed", req_id)
522
523 except HTTPException:
@@ -553,6 +703,11 @@ if __name__ == "__main__":
703 parser = argparse.ArgumentParser()
704 parser.add_argument("--host", default="0.0.0.0")
705 parser.add_argument("--port", type=int, default=8000)
706 + parser.add_argument(
707 + "--no-rewrite",
708 + action="store_true",
709 + help="Forward images to the decoder unchanged (for stage-timing A/B).",
710 + )
711 parser.add_argument(
712 "--encode-servers-urls",
713 required=True,
@@ -573,6 +728,7 @@ if __name__ == "__main__":
728 )
729
730 args = parser.parse_args()
731 + NO_REWRITE = args.no_rewrite
732 app.state.e_urls = [
733 u.strip() for u in args.encode_servers_urls.split(",") if u.strip()
734 ]
tests/config/test_multimodal_config.py
+174
@@ -4,10 +4,13 @@
4 from unittest.mock import MagicMock, patch
5
6 import pytest
7 +import torch
8 from transformers import PretrainedConfig
9
10 +from vllm.config.ec_transfer import ECRole, ECTransferConfig
11 from vllm.config.model import ModelConfig
12 from vllm.config.multimodal import MultiModalConfig
13 +from vllm.config.vllm import VllmConfig
14 from vllm.transformers_utils.model_arch_config_convertor import (
15 ModelArchConfigConvertorBase,
16 )
@@ -173,3 +176,174 @@ def test_sticky_cache_survives_text_subconfig_regeneration():
176 side_effect=AssertionError("must not re-query registry"),
177 ):
178 assert text_config._supports_multimodal_for_mm_prefix() is False
179 +
180 +
181 +@pytest.mark.parametrize(
182 + ("device", "expected"),
183 + [
184 + (None, None),
185 + ("cpu", "cpu"),
186 + ("cuda", "cuda"),
187 + # Callers compare against a bare device type, so an indexed device and a
188 + # torch.device must normalise to the same thing as "cuda".
189 + ("cuda:1", "cuda"),
190 + (torch.device("cuda", 1), "cuda"),
191 + ],
192 +)
193 +def test_mm_processor_device_type_normalizes(device: object, expected: str | None):
194 + kwargs = {} if device is None else {"device": device}
195 + config = MultiModalConfig(mm_processor_kwargs=kwargs)
196 + assert config.get_mm_processor_device_type() == expected
197 +
198 +
199 +def _validate_mm_processor_device(*, device: str, ec_role: ECRole | None) -> None:
200 + ec_config = (
201 + None
202 + if ec_role is None
203 + # `is_ec_producer`/`is_ec_consumer` are False without a connector, so an
204 + # unset one would make every role look like "no EC role at all".
205 + else ECTransferConfig(ec_connector="ECExampleConnector", ec_role=ec_role)
206 + )
207 + MultiModalConfig(
208 + mm_processor_kwargs={"device": device}
209 + ).validate_mm_processor_device(ec_config)
210 +
211 +
212 +@pytest.mark.parametrize("ec_role", [None, "ec_producer"])
213 +def test_bad_mm_processor_device_rejected(ec_role: ECRole | None):
214 + """A bad device must fail during startup, not mid-request.
215 +
216 + Rejected for every role, and on a CPU-only platform too, so a typo can never
217 + silently fall through to running the processor somewhere unintended.
218 + """
219 + with (
220 + patch("vllm.platforms.current_platform.device_type", "cpu"),
221 + pytest.raises(ValueError, match='Invalid "device" in mm_processor_kwargs'),
222 + ):
223 + _validate_mm_processor_device(device="not-a-device", ec_role=ec_role)
224 +
225 +
226 +@pytest.mark.parametrize("ec_role", [None, "ec_consumer", "ec_both"])
227 +def test_accelerator_mm_processor_rejected_outside_encoder_instance(
228 + ec_role: ECRole | None,
229 +):
230 + """Only an encode-only EPD instance has the device to itself.
231 +
232 + Every other role runs the language model in the same process, so frontend
233 + accelerator work would contend with the forward pass and allocate outside
234 + the memory profiled for the KV cache.
235 + """
236 + with (
237 + patch("vllm.platforms.current_platform.device_type", "cuda"),
238 + pytest.raises(ValueError, match="also runs the language model"),
239 + ):
240 + _validate_mm_processor_device(device="cuda", ec_role=ec_role)
241 +
242 +
243 +def test_accelerator_mm_processor_allowed_on_encoder_instance():
244 + with patch("vllm.platforms.current_platform.device_type", "cuda"):
245 + _validate_mm_processor_device(device="cuda", ec_role="ec_producer")
246 +
247 +
248 +def test_cpu_mm_processor_needs_no_ec_role():
249 + """The gate only applies to the accelerator, so CPU stays unrestricted."""
250 + with patch("vllm.platforms.current_platform.device_type", "cuda"):
251 + _validate_mm_processor_device(device="cpu", ec_role=None)
252 +
253 +
254 +@pytest.mark.parametrize(
255 + ("flag", "kwargs", "expected"),
256 + [
257 + # "auto" is only settled once the EC role is known, so it must leave no
258 + # device behind for `VllmConfig` to mistake for an explicit request.
259 + ("auto", None, None),
260 + (None, None, None),
261 + ("cpu", None, "cpu"),
262 + ("cuda", None, "cuda"),
263 + # Any explicit value other than "cpu" means "whatever this platform
264 + # calls its accelerator".
265 + ("gpu", None, "cuda"),
266 + # An explicit device in the kwargs wins over the convenience flag.
267 + ("cpu", {"device": "cuda"}, "cuda"),
268 + ("cuda", {"device": "cpu"}, "cpu"),
269 + ],
270 +)
271 +def test_fold_mm_processor_device(
272 + flag: str | None, kwargs: dict | None, expected: str | None
273 +):
274 + with patch("vllm.platforms.current_platform.device_type", "cuda"):
275 + folded = MultiModalConfig.fold_mm_processor_device(kwargs, flag)
276 +
277 + config = MultiModalConfig(mm_processor_kwargs=folded or {})
278 + assert config.get_mm_processor_device_type() == expected
279 +
280 +
281 +def _resolve_mm_processor_device(
282 + *,
283 + ec_role: ECRole | None,
284 + mm_tensor_ipc: str = "torch_shm",
285 + device: str | None = None,
286 +) -> str | None:
287 + """Run the `auto` resolution and report where the processor ended up."""
288 + mm_config = MultiModalConfig(
289 + mm_processor_kwargs={} if device is None else {"device": device},
290 + mm_tensor_ipc=mm_tensor_ipc, # type: ignore[arg-type]
291 + )
292 + model_config = MagicMock(spec=ModelConfig)
293 + model_config.multimodal_config = mm_config
294 + vllm_config = MagicMock(spec=VllmConfig)
295 + vllm_config.model_config = model_config
296 + vllm_config.ec_transfer_config = (
297 + None
298 + if ec_role is None
299 + else ECTransferConfig(ec_connector="ECExampleConnector", ec_role=ec_role)
300 + )
301 +
302 + with patch("vllm.platforms.current_platform.device_type", "cuda"):
303 + VllmConfig._resolve_mm_processor_device(vllm_config)
304 + return mm_config.get_mm_processor_device_type()
305 +
306 +
307 +def test_auto_mm_processor_device_uses_accelerator_on_encoder_instance():
308 + """The one deployment that has the device to itself and can hand it over."""
309 + assert _resolve_mm_processor_device(ec_role="ec_producer") == "cuda"
310 +
311 +
312 +@pytest.mark.parametrize("ec_role", [None, "ec_consumer", "ec_both"])
313 +def test_auto_mm_processor_device_stays_on_cpu_off_encoder_instance(
314 + ec_role: ECRole | None,
315 +):
316 + """Every other role runs the language model in the same process."""
317 + assert _resolve_mm_processor_device(ec_role=ec_role) is None
318 +
319 +
320 +def test_auto_mm_processor_device_needs_a_device_capable_transport():
321 + """Other transports serialize host bytes, so the output is copied back."""
322 + assert (
323 + _resolve_mm_processor_device(ec_role="ec_producer", mm_tensor_ipc="direct_rpc")
324 + is None
325 + )
326 +
327 +
328 +def test_auto_mm_processor_device_leaves_an_explicit_request_alone():
329 + assert _resolve_mm_processor_device(ec_role="ec_producer", device="cpu") == "cpu"
330 +
331 +
332 +def test_vllm_config_runs_the_mm_processor_device_check():
333 + """Startup must reach the check; the rule itself is covered above.
334 +
335 + Guards the wiring, which no other test would notice going missing.
336 + """
337 + model_config = MagicMock(spec=ModelConfig)
338 + model_config.multimodal_config = MultiModalConfig(
339 + mm_processor_kwargs={"device": "cuda"}
340 + )
341 + vllm_config = MagicMock(spec=VllmConfig)
342 + vllm_config.model_config = model_config
343 + vllm_config.ec_transfer_config = None
344 +
345 + with (
346 + patch("vllm.platforms.current_platform.device_type", "cuda"),
347 + pytest.raises(ValueError, match="also runs the language model"),
348 + ):
349 + VllmConfig._validate_mm_processor_device(vllm_config)
vllm/config/ec_transfer.py
+10
@@ -103,5 +103,15 @@ class ECTransferConfig:
103 def is_ec_consumer(self) -> bool:
104 return self.ec_connector is not None and self.ec_role in get_args(ECConsumer)
105
106 + @property
107 + def is_encode_only(self) -> bool:
108 + """Whether this instance encodes but does not run the language model.
109 +
110 + It allocates no KV cache either -- `GPUModelRunner.get_kv_cache_spec`
111 + returns {} for it -- so it is the one role that can spend accelerator
112 + time and memory on frontend work.
113 + """
114 + return self.is_ec_producer and not self.is_ec_consumer
115 +
116 def get_from_extra_config(self, key, default) -> Any:
117 return self.ec_connector_extra_config.get(key, default)
vllm/config/model.py
+7
@@ -18,6 +18,7 @@ from vllm.config.multimodal import (
18 MMCacheType,
19 MMEncoderTPMode,
20 MMHasherAlgorithm,
21 + MMProcessorDevice,
22 MMTensorIPC,
23 MultiModalConfig,
24 )
@@ -395,6 +396,7 @@ class ModelConfig:
396 video_pruning_method: InitVar[str | None] = None
397 mm_tensor_ipc: InitVar[MMTensorIPC] = None
398 mm_ipc_gpu_memory_gb: InitVar[float | None] = None
399 + mm_processor_device: InitVar[MMProcessorDevice | None] = None
400
401 def compute_hash(self) -> str:
402 """
@@ -525,6 +527,7 @@ class ModelConfig:
527 video_pruning_method: str | None,
528 mm_tensor_ipc: MMTensorIPC,
529 mm_ipc_gpu_memory_gb: float | None,
530 + mm_processor_device: MMProcessorDevice | None,
531 ) -> None:
532 # Keep set served_model_name before maybe_model_redirect(self.model)
533 self.served_model_name = get_served_model_name(
@@ -763,6 +766,10 @@ class ModelConfig:
766 )
767 mm_encoder_tp_mode = "weights"
768
769 + mm_processor_kwargs = MultiModalConfig.fold_mm_processor_device(
770 + mm_processor_kwargs, mm_processor_device
771 + )
772 +
773 mm_config_kwargs = dict(
774 language_model_only=language_model_only,
775 limit_per_prompt=limit_mm_per_prompt,
vllm/config/multimodal.py
+136
@@ -5,14 +5,19 @@ from collections.abc import Mapping
5 from pathlib import Path
6 from typing import Any, Literal, TypeAlias, TypedDict, cast, final
7
8 +import torch
9 from pydantic import ConfigDict, Field, field_validator, model_validator
10 from pydantic.dataclasses import dataclass
11
12 import vllm.envs as envs
13 +from vllm.config.ec_transfer import ECTransferConfig
14 from vllm.config.utils import config, get_from_deprecated_env_if_set
15 +from vllm.logger import init_logger
16 from vllm.utils.hashing import safe_hash
17 from vllm.v1.attention.backends.registry import AttentionBackendEnum
18
19 +logger = init_logger(__name__)
20 +
21
22 @dataclass
23 class BaseDummyOptions:
@@ -64,6 +69,10 @@ MMCacheType = Literal["shm", "lru"]
69 VideoPruningMethod = Literal["evs", "vidcom2"]
70 MMTensorIPC = Literal["direct_rpc", "torch_shm"]
71 MMHasherAlgorithm = Literal["blake3", "sha256", "sha512"]
72 +MMProcessorDevice: TypeAlias = str
73 +"""`"auto"`, `"cpu"`, or the platform's own accelerator name
74 +(`current_platform.device_type`, e.g. `"cuda"` on CUDA and ROCm,
75 +`"xpu"` on XPU). Validated against that set by the CLI."""
76
77
78 def _get_mm_hasher_algorithm() -> MMHasherAlgorithm:
@@ -222,6 +231,20 @@ class MultiModalConfig:
231 - "direct_rpc": Use msgspec serialization via RPC
232 - "torch_shm": Use torch.multiprocessing shared memory for zero-copy IPC
233 Defaults to "direct_rpc". """
234 + mm_embeds_from_ec_connector: bool = False
235 + """Whether a pre-computed-embedding input may omit the `*_embeds` tensor.
236 +
237 + In an encode/prefill/decode (EPD) deployment the encoder instance publishes
238 + embeddings through the EC connector, so the request that reaches the
239 + prefill/decode instance only needs to carry the grid/size metadata that
240 + sizes the placeholder range — the embeddings themselves come from the
241 + connector, keyed by `mm_hash`.
242 +
243 + Derived, not user-settable: `VllmConfig.__post_init__` sets this to True
244 + exactly on EC consumers. Everywhere else it stays False so that a request
245 + which forgets its embeddings still fails fast in the frontend, with a clear
246 + error, rather than deep inside the model."""
247 +
248 mm_ipc_gpu_memory_gb: float = Field(default=0, ge=0)
249 """Amount of GPU memory (in GiB) sequestered on the engine's device for
250 GPU-side multimodal work in the API-server (frontend) process, such as
@@ -319,6 +342,119 @@ class MultiModalConfig:
342 )
343 return self
344
345 + @staticmethod
346 + def fold_mm_processor_device(
347 + mm_processor_kwargs: dict[str, Any] | None,
348 + mm_processor_device: MMProcessorDevice | None,
349 + ) -> dict[str, Any] | None:
350 + """Fold the `mm_processor_device` convenience flag into the kwargs.
351 +
352 + The flag keeps no state of its own: `mm_processor_kwargs["device"]` is
353 + the only representation of where the processor runs, so an explicit
354 + `device` there always wins and `"auto"` stays unresolved for
355 + `VllmConfig`, which is where the EC role needed to resolve it lives.
356 +
357 + Args:
358 + mm_processor_kwargs: The kwargs as given, or None.
359 + mm_processor_device: The flag's value, or None when unset.
360 +
361 + Returns:
362 + The kwargs to build the config with, unchanged unless the flag adds
363 + a `device`.
364 + """
365 + if mm_processor_device in (None, "auto"):
366 + return mm_processor_kwargs
367 + if (mm_processor_kwargs or {}).get("device") is not None:
368 + return mm_processor_kwargs
369 +
370 + from vllm.platforms import current_platform
371 +
372 + # Any explicit value other than "cpu" means "the accelerator", so a
373 + # programmatically-set "cuda" still works on a platform whose device type
374 + # is named differently ("xpu"), and degrades to CPU where there is none.
375 + device = (
376 + "cpu"
377 + if mm_processor_device == "cpu"
378 + else (current_platform.device_type or "cpu")
379 + )
380 + return {**(mm_processor_kwargs or {}), "device": device}
381 +
382 + def get_mm_processor_device_type(self) -> str | None:
383 + """The torch device type `mm_processor_kwargs["device"]` names.
384 +
385 + `mm_processor_kwargs` is untyped, so `device` may be any form torch
386 + accepts -- `"cuda"`, `"cuda:1"`, `torch.device(...)`, or a bare index.
387 + Normalising through torch rather than parsing the string keeps the
388 + non-string forms from slipping past a caller's comparison.
389 +
390 + Returns:
391 + The device type, or None when no device is requested.
392 +
393 + Raises:
394 + ValueError: If `device` is not something `torch.device` accepts.
395 + `validate_mm_processor_device` is what surfaces this during
396 + startup, so the value is only parsed once.
397 + """
398 + device = (self.mm_processor_kwargs or {}).get("device")
399 + if device is None:
400 + return None
401 + try:
402 + return torch.device(device).type # type: ignore[arg-type]
403 + except (RuntimeError, TypeError, ValueError):
404 + raise ValueError(
405 + f'Invalid "device" in mm_processor_kwargs: {device!r}. Expected a '
406 + 'torch device such as "cpu", "cuda" or "cuda:0".'
407 + ) from None
408 +
409 + def validate_mm_processor_device(self, ec_config: ECTransferConfig | None) -> None:
410 + """Check `mm_processor_kwargs["device"]` for this deployment.
411 +
412 + The only place the requested device is validated, so it runs even on a
413 + CPU-only platform: the value is parsed before any early return.
414 +
415 + Args:
416 + ec_config: The deployment's EC config, or None when it is not an
417 + encode/prefill/decode deployment. Passed in because it is not
418 + reachable from here, and because a field assigned after
419 + construction would not re-trigger this config's validators.
420 +
421 + Raises:
422 + ValueError: If the requested device is not a torch device, or if it
423 + is the accelerator on an instance that also runs the language
424 + model.
425 + """
426 + from vllm.platforms import current_platform
427 +
428 + device_type = self.get_mm_processor_device_type()
429 + accelerator = current_platform.device_type
430 + if device_type is None or accelerator in ("", "cpu"):
431 + return
432 + if device_type != accelerator:
433 + return
434 +
435 + if ec_config is None or not ec_config.is_encode_only:
436 + raise ValueError(
437 + f"Cannot run the multi-modal processor on {device_type!r}: this "
438 + "instance also runs the language model. The processor would "
439 + "share the device with the model's forward pass, so its "
440 + "transform kernels contend with that compute, and because it "
441 + "runs in the API-server process its allocations are outside the "
442 + "memory the engine profiled for its KV cache -- risking OOM or "
443 + "a silently shrunken cache.\n"
444 + "Accelerator preprocessing is only supported on an encode-only "
445 + "instance of an encode/prefill/decode deployment (an EC "
446 + "producer that is not also a consumer), which runs no forward "
447 + "pass and allocates no KV cache.\n"
448 + 'Use --mm-processor-device=cpu, or drop "device" from '
449 + "--mm-processor-kwargs."
450 + )
451 +
452 + logger.info_once(
453 + "Running the multi-modal processor on %s. Override with "
454 + "--mm-processor-device=cpu.",
455 + device_type,
456 + )
457 +
458 def compute_hash(self) -> str:
459 """
460 WARNING: Whenever a new field is added to this config,
vllm/config/vllm.py
+99
@@ -1509,6 +1509,10 @@ class VllmConfig:
1509 )
1510 current_platform.check_and_update_config(self)
1511
1512 + self._resolve_mm_embeds_from_ec_connector()
1513 + self._resolve_mm_processor_device()
1514 + self._validate_mm_processor_device()
1515 +
1516 if self.use_v2_model_runner:
1517 self._validate_v2_model_runner()
1518 elif self.parallel_config.prefill_context_parallel_size > 1:
@@ -2185,6 +2189,101 @@ class VllmConfig:
2189 f"kernel_config={self.kernel_config!r}"
2190 )
2191
2192 + def _resolve_mm_embeds_from_ec_connector(self) -> None:
2193 + """Allow `*_embeds` to be omitted only where the connector supplies them.
2194 +
2195 + That is exactly an EC consumer: the encoder instance publishes the
2196 + embeddings through the EC connector, so the request only has to carry
2197 + the grid metadata that sizes the placeholder range. On every other
2198 + deployment a missing `*_embeds` is a client error and must keep failing
2199 + fast in the frontend.
2200 + """
2201 + model_config = self.model_config
2202 + if model_config is None:
2203 + return
2204 + mm_config = model_config.multimodal_config
2205 + if mm_config is None:
2206 + return
2207 +
2208 + ec_config = self.ec_transfer_config
2209 + # Derived, so overwrite unconditionally rather than honouring a value
2210 + # that was set by hand.
2211 + mm_config.mm_embeds_from_ec_connector = (
2212 + ec_config is not None and ec_config.is_ec_consumer
2213 + )
2214 + if mm_config.mm_embeds_from_ec_connector:
2215 + logger.info_once(
2216 + "EC consumer: pre-computed-embedding inputs may omit the "
2217 + "embedding tensor; embeddings are loaded from the EC connector."
2218 + )
2219 +
2220 + def _resolve_mm_processor_device(self) -> None:
2221 + """Settle `--mm-processor-device=auto` now that the EC role is known.
2222 +
2223 + "auto" means "the accelerator, but only where the processor has it to
2224 + itself and its output can be handed over without a copy back to host":
2225 + an encode-only instance whose tensor transport carries device tensors.
2226 + Every other deployment keeps the processor on CPU.
2227 +
2228 + An explicit device -- from `--mm-processor-device` or straight from
2229 + `mm_processor_kwargs` -- is already folded in by `MultiModalConfig`, so
2230 + it is left alone here and validated by `_validate_mm_processor_device`.
2231 + """
2232 + model_config = self.model_config
2233 + if model_config is None:
2234 + return
2235 + mm_config = model_config.multimodal_config
2236 + if mm_config is None:
2237 + return
2238 + if mm_config.get_mm_processor_device_type() is not None:
2239 + return
2240 +
2241 + from vllm.platforms import current_platform
2242 +
2243 + device_type = current_platform.device_type
2244 + if device_type in ("", "cpu"):
2245 + return
2246 +
2247 + ec_config = self.ec_transfer_config
2248 + # An EC producer that is not also a consumer runs no forward pass and
2249 + # allocates no KV cache, so frontend accelerator work has the device to
2250 + # itself.
2251 + if ec_config is None or not ec_config.is_encode_only:
2252 + return
2253 +
2254 + if mm_config.mm_tensor_ipc != "torch_shm":
2255 + # Any other transport serializes host bytes, so the output would be
2256 + # copied back, and that copy costs more than running the transform
2257 + # on device saves.
2258 + logger.info_once(
2259 + "EPD encoder instance: keeping the multi-modal processor on CPU "
2260 + "because mm_tensor_ipc=%s cannot carry device tensors. Add "
2261 + "--mm-tensor-ipc=torch_shm to run it on the accelerator.",
2262 + mm_config.mm_tensor_ipc,
2263 + )
2264 + return
2265 +
2266 + mm_config.mm_processor_kwargs = {
2267 + **(mm_config.mm_processor_kwargs or {}),
2268 + "device": device_type,
2269 + }
2270 + logger.info_once(
2271 + "EPD encoder instance: running the multi-modal processor on %s. "
2272 + "Override with --mm-processor-device=cpu.",
2273 + device_type,
2274 + )
2275 +
2276 + def _validate_mm_processor_device(self) -> None:
2277 + """Hand the EC config to `MultiModalConfig`, which owns the rule."""
2278 + model_config = self.model_config
2279 + if model_config is None:
2280 + return
2281 + mm_config = model_config.multimodal_config
2282 + if mm_config is None:
2283 + return
2284 +
2285 + mm_config.validate_mm_processor_device(self.ec_transfer_config)
2286 +
2287 def _get_v2_model_runner_unsupported_features(self) -> list[str]:
2288 """Collect features not yet supported by the V2 model runner."""
2289 unsupported: list[str] = []
vllm/distributed/ec_transfer/ec_connector/example_connector.py
+68 -1
@@ -2,9 +2,10 @@
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 import os
4 from dataclasses import dataclass
5 -from typing import TYPE_CHECKING
5 +from typing import TYPE_CHECKING, Any
6
7 import safetensors
8 +import torch
9
10 from vllm.config import VllmConfig
11 from vllm.distributed.ec_transfer.ec_connector.base import (
@@ -50,6 +51,8 @@ class ECExampleConnector(ECConnectorBase):
51 super().__init__(vllm_config=vllm_config, role=role)
52 # req_id -> index
53 self._mm_datas_need_loads: dict[str, int] = {}
54 + self._model_config = vllm_config.model_config
55 + self._metadata_fields_cache: dict[str, set[str]] = {}
56 transfer_config = vllm_config.ec_transfer_config
57 if transfer_config is not None:
58 self._storage_path = transfer_config.get_from_extra_config(
@@ -165,6 +168,70 @@ class ECExampleConnector(ECConnectorBase):
168 self._mm_datas_need_loads.clear()
169 return meta
170
171 + def _placeholder_metadata_fields(self, modality: str) -> set[str]:
172 + """Which processed keys this model needs published for `modality`.
173 +
174 + Read from `MultiModalDataParser.embedding_fields`, the same declaration
175 + the consumer's parser requires, so the two cannot drift. An empty set
176 + means the modality cannot be delivered out of band, and the consumer
177 + will process the media itself.
178 + """
179 + if modality in self._metadata_fields_cache:
180 + return self._metadata_fields_cache[modality]
181 +
182 + fields: set[str] = set()
183 + try:
184 + from vllm.multimodal import MULTIMODAL_REGISTRY
185 +
186 + info = MULTIMODAL_REGISTRY.create_processor(self._model_config).info
187 + fields = info.data_parser.placeholder_metadata_fields(modality)
188 + except Exception:
189 + # Reporting nothing is a safe degradation: the consumer falls back to
190 + # processing the media itself.
191 + logger.warning(
192 + "Could not determine the placeholder metadata fields for "
193 + "modality %s; the consumer will preprocess the media itself.",
194 + modality,
195 + exc_info=True,
196 + )
197 +
198 + self._metadata_fields_cache[modality] = fields
199 + return fields
200 +
201 + def request_finished(
202 + self,
203 + request: "Request",
204 + ) -> tuple[bool, dict[str, Any] | None]:
205 + """Report each item's cache key and grid so a consumer can skip the
206 + image transform.
207 +
208 + A consumer only needs the grid to size the prompt's placeholder range;
209 + the embedding itself arrives through this connector. Reporting the grid
210 + the producer actually computed keeps the two sides in agreement without
211 + the caller re-deriving it from the raw media.
212 + """
213 + if not self.is_producer:
214 + return False, None
215 +
216 + items = []
217 + for feature in request.mm_features:
218 + metadata = {}
219 + # `data` is None for items served from the processor cache, in which
220 + # case the metadata is unavailable here and the consumer has to fall
221 + # back to processing the media itself.
222 + if feature.data is not None:
223 + wanted = self._placeholder_metadata_fields(feature.modality)
224 + metadata = {
225 + key: value.tolist()
226 + for key, value in feature.data.get_data().items()
227 + if key in wanted and isinstance(value, torch.Tensor)
228 + }
229 + items.append({"mm_hash": feature.identifier, **metadata})
230 +
231 + if not items:
232 + return False, None
233 + return False, {"ec_items": items}
234 +
235 # ==============================
236 # Helper functions
237 # ==============================
vllm/engine/arg_utils.py
+27
@@ -90,6 +90,7 @@ from vllm.config.multimodal import (
90 MMCacheType,
91 MMEncoderTPMode,
92 MMHasherAlgorithm,
93 + MMProcessorDevice,
94 MMTensorIPC,
95 )
96 from vllm.config.observability import DetailedTraceModules
@@ -596,6 +597,7 @@ class EngineArgs:
597 video_pruning_rate: float | None = MultiModalConfig.video_pruning_rate
598 video_pruning_method: str = MultiModalConfig.video_pruning_method
599 mm_tensor_ipc: MMTensorIPC = MultiModalConfig.mm_tensor_ipc
600 + mm_processor_device: MMProcessorDevice = "auto"
601 mm_ipc_gpu_memory_gb: float = MultiModalConfig.mm_ipc_gpu_memory_gb
602 # LoRA fields
603 enable_lora: bool = False
@@ -1353,6 +1355,30 @@ class EngineArgs:
1355 multimodal_group.add_argument(
1356 "--mm-tensor-ipc", **multimodal_kwargs["mm_tensor_ipc"]
1357 )
1358 + multimodal_group.add_argument(
1359 + "--mm-processor-device",
1360 + choices=["auto", "cpu"]
1361 + + (
1362 + [current_platform.device_type]
1363 + if current_platform.device_type not in ("", "cpu")
1364 + else []
1365 + ),
1366 + default="auto",
1367 + help="Device the HF multi-modal processor runs the image/video "
1368 + "transform on. Convenience for `--mm-processor-kwargs "
1369 + "'{\"device\": ...}'`: the value is resolved here and stored there, "
1370 + "it is not kept as separate state. Only takes effect for HF "
1371 + '"fast" (torchvision-backed) processors, which accept a `device` '
1372 + "argument; the others ignore it and stay on CPU.\n\n"
1373 + '"auto" uses the accelerator on encoder instances of an '
1374 + "encode/prefill/decode deployment -- an EC producer that is not "
1375 + "also a consumer allocates no KV cache, so its accelerator is not "
1376 + "contended by the language model -- and then only when "
1377 + "`--mm-tensor-ipc=torch_shm` can carry device tensors, since every "
1378 + "other transport would copy the result back to the host and that "
1379 + 'copy costs more than it saves. "auto" resolves to "cpu" '
1380 + "everywhere else.",
1381 + )
1382 multimodal_group.add_argument(
1383 "--mm-ipc-gpu-memory-gb",
1384 **multimodal_kwargs["mm_ipc_gpu_memory_gb"],
@@ -1743,6 +1769,7 @@ class EngineArgs:
1769 video_pruning_method=self.video_pruning_method,
1770 mm_tensor_ipc=self.mm_tensor_ipc,
1771 mm_ipc_gpu_memory_gb=self.mm_ipc_gpu_memory_gb,
1772 + mm_processor_device=self.mm_processor_device,
1773 io_processor_plugin=self.io_processor_plugin,
1774 renderer_num_workers=self.renderer_num_workers,
1775 )
vllm/model_executor/models/colqwen3.py
+1
@@ -104,6 +104,7 @@ class ColQwen3ProcessingInfo(Qwen3VLProcessingInfo):
104 spatial_merge_size,
105 video_needs_metadata=self._supports_video,
106 expected_hidden_size=self._get_expected_hidden_size(),
107 + embeds_from_ec_connector=self.embeds_from_ec_connector,
108 )
109
110
vllm/model_executor/models/colqwen3_5.py
+1
@@ -101,6 +101,7 @@ class ColQwen3_5ProcessingInfo(Qwen3_5ProcessingInfo):
101 spatial_merge_size,
102 video_needs_metadata=self._supports_video,
103 expected_hidden_size=self._get_expected_hidden_size(),
104 + embeds_from_ec_connector=self.embeds_from_ec_connector,
105 )
106
107
vllm/model_executor/models/hunyuan_vision.py
+9 -1
@@ -551,15 +551,22 @@ def _hunyuan_vl_field_config(hf_inputs: Mapping[str, torch.Tensor]):
551
552
553 class HunYuanVLMultiModalDataParser(MultiModalDataParser):
554 + # The patch grid is what sizes the placeholder range.
555 + embedding_fields = {
556 + "image": {"image_embeds": "values", "image_grid_thw": "metadata"},
557 + }
558 +
559 def _parse_image_data(
560 self,
561 data: dict[str, torch.Tensor] | ModalityData[ImageItem],
562 ) -> ModalityDataItems[Any, Any] | None:
563 if isinstance(data, dict):
564 + required, optional = self.embedding_field_sets("image")
565 return DictEmbeddingItems(
566 data,
567 modality="image",
562 - required_fields={"image_embeds", "image_grid_thw"},
568 + required_fields=required,
569 + optional_fields=optional,
570 fields_factory=_hunyuan_vl_field_config,
571 )
572
@@ -592,6 +599,7 @@ class HunYuanVLProcessingInfo(BaseProcessingInfo):
599 def get_data_parser(self):
600 return HunYuanVLMultiModalDataParser(
601 expected_hidden_size=self._get_expected_hidden_size(),
602 + embeds_from_ec_connector=self.embeds_from_ec_connector,
603 )
604
605 def get_supported_mm_limits(self) -> Mapping[str, int | None]:
vllm/model_executor/models/keye.py
+13 -8
@@ -881,18 +881,23 @@ def _keye_field_config(
881
882
883 class KeyeMultiModalDataParser(MultiModalDataParser):
884 + # The patch grid is what sizes the placeholder range.
885 + embedding_fields = {
886 + "image": {"image_embeds": "values", "image_grid_thw": "metadata"},
887 + "video": {"video_embeds": "values", "video_grid_thw": "metadata"},
888 + }
889 +
890 def _parse_image_data(
891 self,
892 data: dict[str, torch.Tensor] | ModalityData[ImageItem],
893 ) -> ModalityDataItems[Any, Any] | None:
894 if isinstance(data, dict):
895 + required, optional = self.embedding_field_sets("image")
896 return DictEmbeddingItems(
897 data,
898 modality="image",
892 - required_fields={
893 - "image_embeds",
894 - "image_grid_thw",
895 - },
899 + required_fields=required,
900 + optional_fields=optional,
901 fields_factory=_keye_field_config,
902 )
903
@@ -903,13 +908,12 @@ class KeyeMultiModalDataParser(MultiModalDataParser):
908 data: dict[str, torch.Tensor] | ModalityData[VideoItem],
909 ) -> ModalityDataItems[Any, Any] | None:
910 if isinstance(data, dict):
911 + required, optional = self.embedding_field_sets("video")
912 return DictEmbeddingItems(
913 data,
914 modality="video",
909 - required_fields={
910 - "video_embeds",
911 - "video_grid_thw",
912 - },
915 + required_fields=required,
916 + optional_fields=optional,
917 fields_factory=_keye_field_config,
918 )
919
@@ -929,6 +933,7 @@ class KeyeProcessingInfo(BaseProcessingInfo):
933 def get_data_parser(self):
934 return KeyeMultiModalDataParser(
935 expected_hidden_size=self._get_expected_hidden_size(),
936 + embeds_from_ec_connector=self.embeds_from_ec_connector,
937 )
938
939 def get_supported_mm_limits(
vllm/model_executor/models/keye_vl1_5.py
+13 -8
@@ -320,18 +320,23 @@ def _keye_field_config(
320
321
322 class KeyeVL1_5MultiModalDataParser(MultiModalDataParser):
323 + # The patch grid is what sizes the placeholder range.
324 + embedding_fields = {
325 + "image": {"image_embeds": "values", "image_grid_thw": "metadata"},
326 + "video": {"video_embeds": "values", "video_grid_thw": "metadata"},
327 + }
328 +
329 def _parse_image_data(
330 self,
331 data: dict[str, torch.Tensor] | ModalityData[ImageItem],
332 ) -> ModalityDataItems[Any, Any] | None:
333 if isinstance(data, dict):
334 + required, optional = self.embedding_field_sets("image")
335 return DictEmbeddingItems(
336 data,
337 modality="image",
331 - required_fields={
332 - "image_embeds",
333 - "image_grid_thw",
334 - },
338 + required_fields=required,
339 + optional_fields=optional,
340 fields_factory=_keye_field_config,
341 )
342
@@ -342,13 +347,12 @@ class KeyeVL1_5MultiModalDataParser(MultiModalDataParser):
347 data: dict[str, torch.Tensor] | ModalityData[VideoItem],
348 ) -> ModalityDataItems[Any, Any] | None:
349 if isinstance(data, dict):
350 + required, optional = self.embedding_field_sets("video")
351 return DictEmbeddingItems(
352 data,
353 modality="video",
348 - required_fields={
349 - "video_embeds",
350 - "video_grid_thw",
351 - },
354 + required_fields=required,
355 + optional_fields=optional,
356 fields_factory=_keye_field_config,
357 )
358
@@ -359,6 +363,7 @@ class KeyeVL1_5ProcessingInfo(KeyeProcessingInfo):
363 def get_data_parser(self):
364 return KeyeVL1_5MultiModalDataParser(
365 expected_hidden_size=self._get_expected_hidden_size(),
366 + embeds_from_ec_connector=self.embeds_from_ec_connector,
367 )
368
369 def get_max_frame_per_video(self) -> int:
vllm/model_executor/models/llava_onevision2.py
+9 -1
@@ -1279,6 +1279,7 @@ class LlavaOnevision2ProcessingInfo(BaseProcessingInfo):
1279 return LlavaOnevision2MultiModalDataParser(
1280 self.get_hf_config().vision_config.spatial_merge_size,
1281 video_needs_metadata=True,
1282 + embeds_from_ec_connector=self.embeds_from_ec_connector,
1283 )
1284
1285 def get_hf_processor(self, **kwargs: object):
@@ -1510,6 +1511,11 @@ class LlavaOnevision2DummyInputsBuilder(
1511
1512
1513 class LlavaOnevision2MultiModalDataParser(MultiModalDataParser):
1514 + # The patch grid is what sizes the placeholder range.
1515 + embedding_fields = {
1516 + "image": {"image_embeds": "values", "image_grid_thw": "metadata"},
1517 + }
1518 +
1519 def __init__(self, spatial_merge_size: int, *args, **kwargs):
1520 self._spatial_merge_size = spatial_merge_size
1521 super().__init__(*args, **kwargs)
@@ -1519,10 +1525,12 @@ class LlavaOnevision2MultiModalDataParser(MultiModalDataParser):
1525 data: dict[str, torch.Tensor] | ModalityData[ImageItem],
1526 ) -> ModalityDataItems[Any, Any] | None:
1527 if isinstance(data, dict):
1528 + required, optional = self.embedding_field_sets("image")
1529 return DictEmbeddingItems(
1530 data,
1531 modality="image",
1525 - required_fields={"image_embeds", "image_grid_thw"},
1532 + required_fields=required,
1533 + optional_fields=optional,
1534 fields_factory=_create_field_factory(self._spatial_merge_size),
1535 )
1536 return super()._parse_image_data(data)
vllm/model_executor/models/minicpmv.py
+22 -3
@@ -26,7 +26,7 @@
26
27 import math
28 from collections import defaultdict
29 -from collections.abc import Callable, Iterable, Mapping, Sequence
29 +from collections.abc import Callable, Iterable, Mapping, Sequence, Set
30 from functools import partial
31 from itertools import chain
32 from typing import Annotated, Any, Literal, TypeAlias
@@ -474,11 +474,14 @@ class MiniCPMVImageEmbeddingItems(DictEmbeddingItems):
474 [Mapping[str, torch.Tensor]],
475 Mapping[str, MultiModalFieldConfig],
476 ],
477 + required_fields: Set[str],
478 + optional_fields: Set[str] = frozenset(),
479 ) -> None:
480 super().__init__(
481 data,
482 modality="image",
481 - required_fields={"image_embeds", "image_sizes"},
483 + required_fields=required_fields,
484 + optional_fields=optional_fields,
485 fields_factory=fields_factory,
486 )
487
@@ -495,11 +498,14 @@ class MiniCPMVVideoEmbeddingItems(DictEmbeddingItems):
498 [Mapping[str, torch.Tensor]],
499 Mapping[str, MultiModalFieldConfig],
500 ],
501 + required_fields: Set[str],
502 + optional_fields: Set[str] = frozenset(),
503 ) -> None:
504 super().__init__(
505 data,
506 modality="video",
502 - required_fields={"video_embeds", "video_image_sizes"},
507 + required_fields=required_fields,
508 + optional_fields=optional_fields,
509 fields_factory=fields_factory,
510 )
511
@@ -512,14 +518,23 @@ class MiniCPMVVideoEmbeddingItems(DictEmbeddingItems):
518
519
520 class MiniCPMVMultiModalDataParser(MultiModalDataParser):
521 + # The per-item sizes are what size the placeholder range.
522 + embedding_fields = {
523 + "image": {"image_embeds": "values", "image_sizes": "metadata"},
524 + "video": {"video_embeds": "values", "video_image_sizes": "metadata"},
525 + }
526 +
527 def _parse_image_data(
528 self,
529 data: dict[str, torch.Tensor] | ModalityData[ImageItem],
530 ) -> ModalityDataItems[Any, Any] | None:
531 if isinstance(data, dict):
532 + required, optional = self.embedding_field_sets("image")
533 return MiniCPMVImageEmbeddingItems(
534 data,
535 fields_factory=_minicpmv_field_config,
536 + required_fields=required,
537 + optional_fields=optional,
538 )
539
540 return super()._parse_image_data(data)
@@ -529,9 +544,12 @@ class MiniCPMVMultiModalDataParser(MultiModalDataParser):
544 data: dict[str, torch.Tensor] | ModalityData[VideoItem],
545 ) -> ModalityDataItems[Any, Any] | None:
546 if isinstance(data, dict):
547 + required, optional = self.embedding_field_sets("video")
548 return MiniCPMVVideoEmbeddingItems(
549 data,
550 fields_factory=_minicpmv_field_config,
551 + required_fields=required,
552 + optional_fields=optional,
553 )
554
555 return super()._parse_video_data(data)
@@ -573,6 +591,7 @@ class MiniCPMVProcessingInfo(BaseProcessingInfo):
591 def get_data_parser(self):
592 return MiniCPMVMultiModalDataParser(
593 expected_hidden_size=self._get_expected_hidden_size(),
594 + embeds_from_ec_connector=self.embeds_from_ec_connector,
595 )
596
597 def get_model_version(self):
vllm/model_executor/models/opencua.py
+1
@@ -57,6 +57,7 @@ class OpenCUAProcessingInfo(Qwen2VLProcessingInfo):
57 return Qwen2VLMultiModalDataParser(
58 self.get_hf_config().vision_config.spatial_merge_size,
59 expected_hidden_size=self._get_expected_hidden_size(),
60 + embeds_from_ec_connector=self.embeds_from_ec_connector,
61 )
62
63 def get_hf_config(self):
vllm/model_executor/models/qwen2_5_omni_thinker.py
+1
@@ -372,6 +372,7 @@ class Qwen2_5OmniThinkerProcessingInfo(
372 target_sr=feature_extractor.sampling_rate,
373 target_channels=self.get_target_channels(),
374 expected_hidden_size=self._get_expected_hidden_size(),
375 + embeds_from_ec_connector=self.embeds_from_ec_connector,
376 )
377
378 def get_target_channels(self) -> int:
vllm/model_executor/models/qwen2_vl.py
+13 -2
@@ -795,6 +795,12 @@ def _create_qwen2vl_field_factory(
795
796
797 class Qwen2VLMultiModalDataParser(MultiModalDataParser):
798 + # The patch grid is what sizes the placeholder range.
799 + embedding_fields = {
800 + "image": {"image_embeds": "values", "image_grid_thw": "metadata"},
801 + "video": {"video_embeds": "values", "video_grid_thw": "metadata"},
802 + }
803 +
804 def __init__(self, spatial_merge_size: int, *args, **kwargs):
805 self._spatial_merge_size = spatial_merge_size
806 super().__init__(*args, **kwargs)
@@ -804,10 +810,12 @@ class Qwen2VLMultiModalDataParser(MultiModalDataParser):
810 data: dict[str, torch.Tensor] | ModalityData[ImageItem],
811 ) -> ModalityDataItems[Any, Any] | None:
812 if isinstance(data, dict):
813 + required, optional = self.embedding_field_sets("image")
814 return DictEmbeddingItems(
815 data,
816 modality="image",
810 - required_fields={"image_embeds", "image_grid_thw"},
817 + required_fields=required,
818 + optional_fields=optional,
819 fields_factory=_create_qwen2vl_field_factory(self._spatial_merge_size),
820 )
821
@@ -818,10 +826,12 @@ class Qwen2VLMultiModalDataParser(MultiModalDataParser):
826 data: dict[str, torch.Tensor] | ModalityData[VideoItem],
827 ) -> ModalityDataItems[Any, Any] | None:
828 if isinstance(data, dict):
829 + required, optional = self.embedding_field_sets("video")
830 return DictEmbeddingItems(
831 data,
832 modality="video",
824 - required_fields={"video_embeds", "video_grid_thw"},
833 + required_fields=required,
834 + optional_fields=optional,
835 fields_factory=_create_qwen2vl_field_factory(self._spatial_merge_size),
836 )
837
@@ -846,6 +856,7 @@ class Qwen2VLProcessingInfo(BaseProcessingInfo):
856 return Qwen2VLMultiModalDataParser(
857 self.get_hf_config().vision_config.spatial_merge_size,
858 expected_hidden_size=self._get_expected_hidden_size(),
859 + embeds_from_ec_connector=self.embeds_from_ec_connector,
860 )
861
862 def get_supported_mm_limits(self) -> Mapping[str, int | None]:
vllm/model_executor/models/qwen3_vl.py
+1
@@ -879,6 +879,7 @@ class Qwen3VLProcessingInfo(Qwen2VLProcessingInfo):
879 self.get_hf_config().vision_config.spatial_merge_size,
880 video_needs_metadata=True,
881 expected_hidden_size=self._get_expected_hidden_size(),
882 + embeds_from_ec_connector=self.embeds_from_ec_connector,
883 )
884
885 def _get_vision_info(
vllm/multimodal/inputs.py
+13 -1
@@ -23,7 +23,7 @@ from typing_extensions import TypeVar
23
24 from vllm.utils.collection_utils import is_list_of
25 from vllm.utils.import_utils import LazyLoader
26 -from vllm.utils.jsontree import json_map_leaves
26 +from vllm.utils.jsontree import json_iter_leaves, json_map_leaves
27
28 from .media import MediaWithBytes
29
@@ -286,6 +286,13 @@ def nested_tensors_equal(
286 return a == b
287
288
289 +def _nested_tensors_are_cpu(tensors: NestedTensors) -> bool:
290 + """Whether every tensor in `tensors` lives in host memory."""
291 + return not any(
292 + isinstance(x, torch.Tensor) and not x.is_cpu for x in json_iter_leaves(tensors)
293 + )
294 +
295 +
296 def _nested_tensors_h2d(
297 tensors: NestedTensors,
298 device: torch.types.Device,
@@ -478,6 +485,11 @@ class BaseMultiModalField(ABC):
485 pin_memory = False
486
487 batch = [elem.data for elem in elems]
488 + if pin_memory and not _nested_tensors_are_cpu(batch):
489 + # Data already sits on an accelerator (e.g. a device-side HF
490 + # processor handed it over by IPC). Pinning is a host-memory
491 + # concept, and `pin_memory()` rejects device tensors outright.
492 + pin_memory = False
493 out = self._reduce_data(batch, pin_memory=pin_memory)
494 return _nested_tensors_h2d(out, device=device)
495
vllm/multimodal/parse.py
+89 -6
@@ -40,8 +40,18 @@ from .media import MediaWithBytes
40 _T = TypeVar("_T")
41 _I = TypeVar("_I")
42
43 +EmbeddingFieldRole: TypeAlias = Literal["values", "metadata"]
44 +"""What a field of a pre-computed-embedding input carries.
45 +
46 +`"values"` is the embedding tensor itself; `"metadata"` is a key that sizes the
47 +prompt's placeholder range. The distinction is what lets one declaration serve
48 +both an ordinary request (everything present) and an EC consumer (embeddings
49 +arrive through the connector, metadata still in the request).
50 +"""
51 +
52 if TYPE_CHECKING:
53 import PIL.Image as PILImage
54 +
55 else:
56 PILImage = LazyLoader("PILImage", globals(), "PIL.Image")
57
@@ -250,34 +260,62 @@ class DictEmbeddingItems(
260 self,
261 data: Mapping[str, torch.Tensor],
262 modality: str,
253 - required_fields: set[str],
263 + required_fields: Set[str],
264 fields_factory: Callable[
265 [Mapping[str, torch.Tensor]],
266 Mapping[str, MultiModalFieldConfig],
267 ],
268 + optional_fields: Set[str] = frozenset(),
269 ) -> None:
270 + """
271 + Args:
272 + data: The dictionary of tensors for this modality.
273 + modality: The modality these items belong to.
274 + required_fields: Fields `data` must contain.
275 + fields_factory: Builds the field config from the data.
276 + optional_fields: Fields `data` may omit. Which fields these are is
277 + the caller's decision -- see
278 + `MultiModalDataParser.embedding_field_sets`, where a deployment
279 + that receives embeddings through an EC connector makes the
280 + embeddings optional. They still need a field config, since they
281 + are used whenever they *are* supplied.
282 + """
283 from transformers.feature_extraction_utils import BatchFeature
284
285 super().__init__(data, modality)
286
263 - missing_required_data_keys = required_fields - data.keys()
287 + # Nothing required would leave nothing to size the placeholder range
288 + # from, so the item would parse into zero entries and silently produce a
289 + # wrong prompt instead of failing here.
290 + if not required_fields:
291 + raise ValueError(
292 + f"Cannot parse {modality!r} embeddings: every declared field is "
293 + f"optional ({sorted(optional_fields)}), so nothing is left to "
294 + "size the placeholder range."
295 + )
296 +
297 + declared_fields = set(required_fields) | set(optional_fields)
298 +
299 + missing_required_data_keys = set(required_fields) - data.keys()
300 if missing_required_data_keys:
301 data_keys = set(data.keys())
302 msg = (
267 - f"The data should contain the fields: {required_fields}, "
303 + f"The data should contain the fields: {set(required_fields)}, "
304 f"but only found the following keys: {data_keys}"
305 )
306 raise ValueError(msg)
307
308 fields_config = fields_factory(data)
273 - missing_required_fields = required_fields - fields_config.keys()
309 + # Check every declared field, not just the required ones: an optional
310 + # field still needs a config for when it is supplied.
311 + missing_required_fields = declared_fields - fields_config.keys()
312 if missing_required_fields:
313 fields = set(fields_config.keys())
276 - msg = f"{required_fields=} should be a subset of {fields=}"
314 + msg = f"{declared_fields=} should be a subset of {fields=}"
315 raise ValueError(msg)
316
317 self.fields_config = fields_config
280 - self.required_fields = required_fields
318 + self.required_fields = set(required_fields)
319
320 self._kwargs = MultiModalKwargsItems.from_hf_inputs(
321 BatchFeature(dict(data)),
@@ -516,8 +554,50 @@ class MultiModalDataParser:
554 embedding inputs. If provided, validates that user-supplied
555 embeddings have the correct hidden size to prevent crashes
556 during model inference.
557 + embeds_from_ec_connector (bool): Whether pre-computed embeddings may be
558 + absent from the request because an encode/prefill/decode encoder
559 + instance publishes them through an EC connector instead. Derived by
560 + `BaseProcessingInfo.embeds_from_ec_connector`.
561 """
562
563 + embedding_fields: Mapping[str, Mapping[str, EmbeddingFieldRole]] = {}
564 + """Per-modality field roles for pre-computed-embedding inputs.
565 +
566 + Declared here rather than inside `_parse_*_data` so an EC producer can read
567 + it too: on a producer the request carries real media, so the branch that
568 + builds `DictEmbeddingItems` never runs, yet the producer still has to know
569 + which processed keys to publish. One declaration, both sides, no drift.
570 +
571 + A modality absent from this mapping can only be sent whole in the request.
572 + """
573 +
574 + @classmethod
575 + def placeholder_metadata_fields(cls, modality: str) -> set[str]:
576 + """The keys that size `modality`'s placeholder range.
577 +
578 + What an EC producer publishes alongside the embedding, and what a
579 + consumer keeps requiring once the embedding itself is gone.
580 + """
581 + return {
582 + field
583 + for field, role in cls.embedding_fields.get(modality, {}).items()
584 + if role == "metadata"
585 + }
586 +
587 + def embedding_field_sets(self, modality: str) -> tuple[set[str], set[str]]:
588 + """`modality`'s (required, optional) fields for this deployment.
589 +
590 + Resolves the static roles in `embedding_fields` against where the
591 + embeddings actually come from: on an EC consumer they arrive through the
592 + connector, so the request may omit them; anywhere else a request that
593 + claims to carry pre-computed embeddings has to actually carry them.
594 + """
595 + metadata = self.placeholder_metadata_fields(modality)
596 + values = set(self.embedding_fields.get(modality, {})) - metadata
597 + if self.embeds_from_ec_connector:
598 + return metadata, values
599 + return metadata | values, set()
600 +
601 def __init__(
602 self,
603 *,
@@ -526,9 +606,12 @@ class MultiModalDataParser:
606 audio_resample_method: Literal["pyav", "scipy", "soxr"] = "pyav",
607 video_needs_metadata: bool = False,
608 expected_hidden_size: int | None = None,
609 + embeds_from_ec_connector: bool = False,
610 ) -> None:
611 super().__init__()
612
613 + self.embeds_from_ec_connector = embeds_from_ec_connector
614 +
615 self.audio_resampler = AudioResampler(
616 target_sr=target_sr,
617 method=audio_resample_method,
vllm/multimodal/processing/context.py
+34 -7
@@ -1,7 +1,6 @@
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 import time
4 -from abc import abstractmethod
4 from collections.abc import Callable, Mapping
5 from contextlib import contextmanager
6 from dataclasses import dataclass, field
@@ -227,13 +226,30 @@ class InputProcessingContext:
226 self,
227 output: JSONTree,
228 ) -> JSONTree:
229 + # "torch_shm" puts tensors on a torch.multiprocessing queue, which
230 + # shares device tensors by CUDA IPC handle, so a device-side processor
231 + # can hand `pixel_values` straight to the worker. Every other transport
232 + # serializes host bytes, so the result has to be copied back first.
233 + keep_on_device = (
234 + self.model_config.get_multimodal_config().mm_tensor_ipc == "torch_shm"
235 + )
236 +
237 def _postprocess_one(x: object):
231 - if isinstance(x, torch.Tensor): # noqa: SIM102
232 - # This mimics the behavior of transformers.BatchFeature
233 - if x.is_floating_point():
234 - x = x.to(dtype=self.model_config.dtype)
238 + if not isinstance(x, torch.Tensor):
239 + return x
240 +
241 + # Bind to a Tensor-typed local: reassigning the `object`-typed
242 + # parameter would discard the isinstance narrowing.
243 + tensor = x
244 +
245 + # This mimics the behavior of transformers.BatchFeature
246 + if tensor.is_floating_point():
247 + tensor = tensor.to(dtype=self.model_config.dtype)
248 +
249 + if not tensor.is_cpu and not keep_on_device:
250 + tensor = tensor.cpu()
251
236 - return x
252 + return tensor
253
254 return json_map_leaves(_postprocess_one, output)
255
@@ -350,6 +366,17 @@ class BaseProcessingInfo:
366
367 return None
368
369 + @property
370 + def embeds_from_ec_connector(self) -> bool:
371 + """Whether pre-computed embeddings may arrive outside the request.
372 +
373 + True only on an EC consumer, where an encode/prefill/decode encoder
374 + instance publishes them through the connector instead, so the request
375 + carries only the metadata that sizes the placeholder range.
376 + """
377 + mm_config = self.ctx.model_config.multimodal_config
378 + return mm_config is not None and mm_config.mm_embeds_from_ec_connector
379 +
380 def get_data_parser(self) -> MultiModalDataParser:
381 """
382 Constructs a parser to preprocess multi-modal data items
@@ -362,6 +389,7 @@ class BaseProcessingInfo:
389 """
390 return MultiModalDataParser(
391 expected_hidden_size=self._get_expected_hidden_size(),
392 + embeds_from_ec_connector=self.embeds_from_ec_connector,
393 )
394
395 @cached_property
@@ -372,7 +400,6 @@ class BaseProcessingInfo:
400 def skip_prompt_length_check(self) -> bool:
401 return False
402
375 - @abstractmethod
403 def get_supported_mm_limits(self) -> Mapping[str, int | None]:
404 """
405 Return the maximum supported number of items for each modality.