diff --git a/docs/serving/online_serving/renderer.md b/docs/serving/online_serving/renderer.md index 5469490953..66a19775e0 100644 --- a/docs/serving/online_serving/renderer.md +++ b/docs/serving/online_serving/renderer.md @@ -27,3 +27,147 @@ VLLM_ENABLE_SCALE_OUT_ENDPOINTS=1 vllm serve - Render chat completions For the post processing counterpart that turns generated token IDs back into OpenAI compatible responses, see the [Derenderer APIs](derenderer.md). + +## Multimodal Render Features + +Multimodal render responses include a `features` object with per-modality +hashes, placeholder ranges, and serialized processor data. When the model +exposes placeholder-metadata or `keep_on_cpu` fields (for example +`image_grid_thw`), the response also includes `mm_metadata`. Each +`mm_metadata` entry is a base64-encoded `MultiModalKwargsItem` containing +only those fields, not encoder inputs such as `pixel_values`. + +The arrays in `mm_hashes`, `mm_placeholders`, `kwargs_data`, and +`mm_metadata` use the same per-modality item order. Downstream workers +should split those fields: + +- Encode requests keep `kwargs_data`. +- Prefill requests may omit `kwargs_data` and send `mm_metadata` only when + `ec_transfer_params` is also set, so embeddings are loaded by the EC + connector. Omitting `kwargs_data` without `ec_transfer_params` is rejected. +- Legacy clients that ignore `mm_metadata` and keep sending `kwargs_data` + continue to work. + +## Example + +The example below shows how a disaggregated encode / prefill coordinator can +split a multimodal render response. The render step returns both +`kwargs_data` (encoder tensors plus metadata) and `mm_metadata` (metadata +only). Encode keeps the full payload; prefill drops `kwargs_data` after the +EC connector has published embeddings. + +```python +import httpx + +MODEL = "Qwen/Qwen3-VL-2B-Instruct" +RENDER = "http://localhost:8100" # vllm launch render ... +ENCODE = "http://localhost:8200" # encode worker +PREFILL = "http://localhost:8300" # prefill worker + +chat_request = { + "model": MODEL, + "messages": [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": ""}}, + {"type": "text", "text": "Describe this image."}, + ], + } + ], +} + +with httpx.Client(timeout=120.0) as client: + # 1. Render: preprocess into token IDs and multimodal features. + render_response = client.post( + f"{RENDER}/v1/chat/completions/render", json=chat_request + ).json() + + features = render_response["features"] + # features["kwargs_data"]["image"][0] -> pixel_values + image_grid_thw + # features["mm_metadata"]["image"][0] -> image_grid_thw only + + # 2. Encode: send full kwargs_data so the encoder can run vision towers. + encode_response = client.post( + f"{ENCODE}/inference/v1/generate", + json={ + "token_ids": render_response["token_ids"], + "features": { + "mm_hashes": features["mm_hashes"], + "mm_placeholders": features["mm_placeholders"], + "kwargs_data": features["kwargs_data"], + }, + "sampling_params": {"max_tokens": 1}, + }, + ).json() + ec_transfer_params = encode_response["ec_transfer_params"] + + # 3. Prefill: omit kwargs_data; load embeddings via EC connector. + prefill_response = client.post( + f"{PREFILL}/inference/v1/generate", + json={ + "token_ids": render_response["token_ids"], + "features": { + "mm_hashes": features["mm_hashes"], + "mm_placeholders": features["mm_placeholders"], + "mm_metadata": features["mm_metadata"], + }, + "ec_transfer_params": ec_transfer_params, + "sampling_params": {"max_tokens": 64}, + }, + ).json() + +print(prefill_response["choices"][0]["token_ids"]) +``` + +Single-process clients can keep passing the full render response to +`/inference/v1/generate` unchanged; `mm_metadata` is optional and ignored +when `kwargs_data` is present. + +### Payload shape + +#### Render response + +`/v1/chat/completions/render` returns both `kwargs_data` and `mm_metadata`. +The arrays share the same per-modality item order. Base64 blobs are truncated +below for readability. + +```json +{ + "token_ids": [151644, 872], + "features": { + "mm_hashes": {"image": ["abc123..."]}, + "mm_placeholders": {"image": [{"offset": 0, "length": 256}]}, + "kwargs_data": { + "image": [""] + }, + "mm_metadata": { + "image": [""] + } + } +} +``` + +Forward `kwargs_data` to the encode worker. Keep `mm_metadata` for prefill. + +#### Prefill request + +Prefill omits `kwargs_data` and sends `mm_metadata` with +`ec_transfer_params` from the encode response: + +```json +{ + "token_ids": [151644, 872], + "features": { + "mm_hashes": {"image": ["abc123..."]}, + "mm_placeholders": {"image": [{"offset": 0, "length": 256}]}, + "mm_metadata": { + "image": [""] + } + }, + "ec_transfer_params": { + "ec_items": [{"mm_hash": "abc123...", "peer_host": "10.0.0.1"}] + }, + "sampling_params": {"max_tokens": 64} +} +``` diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_mm_serde.py b/tests/entrypoints/scale_out/token_in_token_out/test_mm_serde.py index d24436bbd4..8bd1bb9091 100644 --- a/tests/entrypoints/scale_out/token_in_token_out/test_mm_serde.py +++ b/tests/entrypoints/scale_out/token_in_token_out/test_mm_serde.py @@ -6,21 +6,30 @@ token_in_token_out generate endpoint. """ import torch +from pydantic import ValidationError +from vllm.entrypoints.scale_out.token_in_token_out.mm_features import ( + extract_mm_features, + merge_mm_kwargs_items, + mm_kwargs_from_features, +) from vllm.entrypoints.scale_out.token_in_token_out.mm_serde import ( decode_mm_kwargs_item, encode_mm_kwargs_item, ) from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( + GenerateRequest, MultiModalFeatures, PlaceholderRangeInfo, ) +from vllm.inputs import mm_input from vllm.multimodal.inputs import ( MultiModalBatchedField, MultiModalFieldElem, MultiModalFlatField, MultiModalKwargsItem, MultiModalSharedField, + PlaceholderRange, ) @@ -112,3 +121,208 @@ def test_mm_features_with_kwargs_data(): decoded = decode_mm_kwargs_item(features2.kwargs_data["image"][0]) assert torch.equal(elem.data, decoded["pixel_values"].data) + + +def _image_engine_input( + *, + pixel_keep_on_cpu: bool = False, + grid_keep_on_cpu: bool = True, +) -> tuple[object, MultiModalFieldElem, MultiModalFieldElem]: + pixel_values = MultiModalFieldElem( + data=torch.randn(5, 3, dtype=torch.float32), + field=MultiModalBatchedField(keep_on_cpu=pixel_keep_on_cpu), + ) + image_grid_thw = MultiModalFieldElem( + data=torch.tensor([[1, 24, 24]], dtype=torch.int32), + field=MultiModalBatchedField(keep_on_cpu=grid_keep_on_cpu), + ) + item = MultiModalKwargsItem( + {"pixel_values": pixel_values, "image_grid_thw": image_grid_thw} + ) + engine_input = mm_input( + prompt_token_ids=[1, 2, 3], + mm_kwargs={"image": [item]}, + mm_hashes={"image": ["abc123"]}, + mm_placeholders={"image": [PlaceholderRange(offset=0, length=1)]}, + ) + return engine_input, pixel_values, image_grid_thw + + +def test_render_extracts_metadata_fields_separately(): + """Render metadata contains CPU fields but excludes encoder inputs.""" + engine_input, pixel_values, image_grid_thw = _image_engine_input() + + features = extract_mm_features(engine_input) + + assert features is not None + assert features.mm_metadata is not None + features = MultiModalFeatures.model_validate_json(features.model_dump_json()) + assert features.mm_metadata is not None + metadata = decode_mm_kwargs_item(features.mm_metadata["image"][0]) + assert set(metadata) == {"image_grid_thw"} + assert torch.equal(metadata["image_grid_thw"].data, image_grid_thw.data) + + assert features.kwargs_data is not None + full_data = decode_mm_kwargs_item(features.kwargs_data["image"][0]) + assert set(full_data) == {"pixel_values", "image_grid_thw"} + assert torch.equal(full_data["pixel_values"].data, pixel_values.data) + + +def test_extract_includes_declared_placeholder_metadata_fields(): + """EC placeholder metadata is kept even when keep_on_cpu is unset.""" + engine_input, _pixel_values, image_grid_thw = _image_engine_input( + grid_keep_on_cpu=False + ) + + features = extract_mm_features( + engine_input, + metadata_fields_for=lambda modality: ( + {"image_grid_thw"} if modality == "image" else set() + ), + ) + + assert features is not None + assert features.mm_metadata is not None + metadata = decode_mm_kwargs_item(features.mm_metadata["image"][0]) + assert set(metadata) == {"image_grid_thw"} + assert torch.equal(metadata["image_grid_thw"].data, image_grid_thw.data) + + +def test_legacy_kwargs_only_generate_keeps_full_payload(): + """Old clients that omit mm_metadata still reconstruct full kwargs.""" + engine_input, pixel_values, image_grid_thw = _image_engine_input() + rendered = extract_mm_features(engine_input) + assert rendered is not None + assert rendered.kwargs_data is not None + + features = MultiModalFeatures( + mm_hashes=rendered.mm_hashes, + mm_placeholders=rendered.mm_placeholders, + kwargs_data=rendered.kwargs_data, + ) + merged = mm_kwargs_from_features(features) + item = merged["image"][0] + assert item is not None + assert set(item) == {"pixel_values", "image_grid_thw"} + assert torch.equal(item["pixel_values"].data, pixel_values.data) + assert torch.equal(item["image_grid_thw"].data, image_grid_thw.data) + + +def test_metadata_only_generate_requires_ec_transfer_params(): + engine_input, _, _ = _image_engine_input() + rendered = extract_mm_features(engine_input) + assert rendered is not None + payload = { + "token_ids": [1, 2, 3], + "sampling_params": {"max_tokens": 1}, + "features": { + "mm_hashes": rendered.mm_hashes, + "mm_placeholders": { + modality: [p.model_dump() for p in ranges] + for modality, ranges in rendered.mm_placeholders.items() + }, + "mm_metadata": rendered.mm_metadata, + }, + } + + try: + GenerateRequest.model_validate(payload) + raise AssertionError("expected metadata-only request without EC to fail") + except ValidationError as exc: + assert "ec_transfer_params" in str(exc) + + payload["ec_transfer_params"] = {"image-hash": {"peer_host": "10.0.0.1"}} + request = GenerateRequest.model_validate(payload) + assert request.features is not None + assert request.features.kwargs_data is None + merged = mm_kwargs_from_features(request.features) + item = merged["image"][0] + assert item is not None + assert set(item) == {"image_grid_thw"} + + +def test_kwargs_and_metadata_generate_does_not_require_ec(): + engine_input, pixel_values, image_grid_thw = _image_engine_input() + rendered = extract_mm_features(engine_input) + assert rendered is not None + request = GenerateRequest.model_validate( + { + "token_ids": [1, 2, 3], + "sampling_params": {"max_tokens": 1}, + "features": rendered.model_dump(), + } + ) + assert request.features is not None + merged = mm_kwargs_from_features(request.features) + item = merged["image"][0] + assert item is not None + assert set(item) == {"pixel_values", "image_grid_thw"} + assert torch.equal(item["pixel_values"].data, pixel_values.data) + assert torch.equal(item["image_grid_thw"].data, image_grid_thw.data) + + +def test_cache_hit_null_kwargs_keeps_metadata_slot(): + pixel_values = MultiModalFieldElem( + data=torch.randn(5, 3, dtype=torch.float32), + field=MultiModalBatchedField(), + ) + image_grid_thw = MultiModalFieldElem( + data=torch.tensor([[1, 24, 24]], dtype=torch.int32), + field=MultiModalBatchedField(keep_on_cpu=True), + ) + item = MultiModalKwargsItem( + {"pixel_values": pixel_values, "image_grid_thw": image_grid_thw} + ) + engine_input = mm_input( + prompt_token_ids=[1, 2, 3, 4], + mm_kwargs={"image": [item, None]}, + mm_hashes={"image": ["hit", "cached"]}, + mm_placeholders={ + "image": [ + PlaceholderRange(offset=0, length=1), + PlaceholderRange(offset=2, length=1), + ] + }, + ) + + features = extract_mm_features(engine_input) + assert features is not None + assert features.kwargs_data is not None + assert features.mm_metadata is not None + assert features.kwargs_data["image"][1] is None + assert features.mm_metadata["image"][1] is None + assert len(features.mm_metadata["image"]) == 2 + metadata = decode_mm_kwargs_item(features.mm_metadata["image"][0]) + assert torch.equal(metadata["image_grid_thw"].data, image_grid_thw.data) + + +def test_features_alignment_rejects_length_mismatch(): + try: + MultiModalFeatures( + mm_hashes={"image": ["a", "b"]}, + mm_placeholders={"image": [PlaceholderRangeInfo(offset=0, length=1)]}, + ) + except ValidationError: + return + raise AssertionError("expected placeholder length mismatch to fail") + + +def test_metadata_can_replace_or_extend_full_mm_data(): + """Metadata-only requests retain the fields required by the model.""" + metadata_elem = MultiModalFieldElem( + data=torch.tensor([[1, 24, 24]], dtype=torch.int32), + field=MultiModalBatchedField(keep_on_cpu=True), + ) + metadata_item = MultiModalKwargsItem({"image_grid_thw": metadata_elem}) + pixel_elem = MultiModalFieldElem( + data=torch.randn(5, 3), + field=MultiModalBatchedField(), + ) + full_item = MultiModalKwargsItem( + {"pixel_values": pixel_elem, "image_grid_thw": metadata_elem} + ) + + assert merge_mm_kwargs_items(None, metadata_item) == metadata_item + merged = merge_mm_kwargs_items(full_item, metadata_item) + assert merged is not None + assert set(merged) == {"pixel_values", "image_grid_thw"} diff --git a/vllm/entrypoints/scale_out/derender/serving.py b/vllm/entrypoints/scale_out/derender/serving.py index 4e4024a909..81e919d840 100644 --- a/vllm/entrypoints/scale_out/derender/serving.py +++ b/vllm/entrypoints/scale_out/derender/serving.py @@ -1,7 +1,6 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import time -from typing import cast import vllm.envs as envs from vllm.entrypoints.openai.chat_completion.protocol import ( @@ -19,16 +18,11 @@ from vllm.entrypoints.openai.models.serving import ( from vllm.entrypoints.serve.engine.protocol import ErrorResponse, UsageInfo from vllm.entrypoints.serve.engine.serving import BaseServing from vllm.entrypoints.serve.utils.request_logger import RequestLogger -from vllm.inputs import ( - EngineInput, - MultiModalHashes, - MultiModalInput, - MultiModalPlaceholders, -) +from vllm.inputs import EngineInput from vllm.logger import init_logger from vllm.renderers.online_derenderer import OnlineDerenderer -from ..token_in_token_out.mm_serde import encode_mm_kwargs_item +from ..token_in_token_out.mm_features import extract_mm_features from ..token_in_token_out.protocol import ( DerenderChatRequest, DerenderChatStreamRequest, @@ -37,7 +31,6 @@ from ..token_in_token_out.protocol import ( DerenderStreamState, GenerateResponse, MultiModalFeatures, - PlaceholderRangeInfo, ) logger = init_logger(__name__) @@ -337,37 +330,4 @@ class ServingDerender(BaseServing): def _extract_mm_features( engine_input: EngineInput, ) -> MultiModalFeatures | None: - """Extract multimodal metadata from a rendered engine prompt. - - Returns ``None`` for text-only prompts. - """ - if engine_input.get("type") != "multimodal": - return None - - # At this point engine_input is a MultiModalInput TypedDict. - mm_engine_input = cast(MultiModalInput, engine_input) - mm_hashes: MultiModalHashes = mm_engine_input["mm_hashes"] - raw_placeholders: MultiModalPlaceholders = mm_engine_input["mm_placeholders"] - - mm_placeholders = { - modality: [ - PlaceholderRangeInfo(offset=p.offset, length=p.length) for p in ranges - ] - for modality, ranges in raw_placeholders.items() - } - - # Serialize tensor data per modality. - kwargs_data: dict[str, list[str | None]] | None = None - if raw_mm_kwargs := mm_engine_input.get("mm_kwargs"): - kwargs_data = {} - for modality, items in raw_mm_kwargs.items(): - kwargs_data[modality] = [ - encode_mm_kwargs_item(item) if item is not None else None - for item in items - ] - - return MultiModalFeatures( - mm_hashes=mm_hashes, - mm_placeholders=mm_placeholders, - kwargs_data=kwargs_data, - ) + return extract_mm_features(engine_input) diff --git a/vllm/entrypoints/scale_out/render/serving.py b/vllm/entrypoints/scale_out/render/serving.py index e800dfebe5..f36b1a3363 100644 --- a/vllm/entrypoints/scale_out/render/serving.py +++ b/vllm/entrypoints/scale_out/render/serving.py @@ -1,7 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project -from typing import cast - from vllm.entrypoints.anthropic.protocol import AnthropicMessagesRequest from vllm.entrypoints.anthropic.serving import AnthropicServingMessages from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest @@ -10,23 +8,20 @@ from vllm.entrypoints.openai.models.serving import ( OpenAIModelRegistry, OpenAIServingModels, ) -from vllm.entrypoints.scale_out.token_in_token_out.mm_serde import encode_mm_kwargs_item +from vllm.entrypoints.scale_out.token_in_token_out.mm_features import ( + extract_mm_features, +) from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( GenerateRequest, MultiModalFeatures, - PlaceholderRangeInfo, ) from vllm.entrypoints.serve.engine.protocol import ErrorResponse from vllm.entrypoints.serve.engine.serving import BaseServing from vllm.entrypoints.serve.utils.api_utils import get_max_tokens from vllm.entrypoints.serve.utils.request_logger import RequestLogger -from vllm.inputs import ( - EngineInput, - MultiModalHashes, - MultiModalInput, - MultiModalPlaceholders, -) +from vllm.inputs import EngineInput from vllm.logger import init_logger +from vllm.multimodal.parse import MultiModalDataParser from vllm.renderers.inputs.preprocess import ( extract_prompt_components, extract_prompt_len, @@ -59,6 +54,9 @@ class ServingRender(BaseServing): ) ) + self._placeholder_metadata_parser: MultiModalDataParser | None = None + self._placeholder_metadata_parser_failed = False + self.default_sampling_params = ( online_renderer.model_config.get_diff_sampling_param() ) @@ -231,41 +229,35 @@ class ServingRender(BaseServing): return generate_requests - @staticmethod + def _placeholder_metadata_fields(self, modality: str) -> set[str]: + """Fields the EC consumer still requires after embeddings transfer.""" + if self._placeholder_metadata_parser_failed: + return set() + if self._placeholder_metadata_parser is None: + try: + from vllm.multimodal import MULTIMODAL_REGISTRY + + self._placeholder_metadata_parser = ( + MULTIMODAL_REGISTRY.create_processor( + self.model_config + ).info.data_parser + ) + except Exception: + logger.debug( + "Could not load placeholder metadata fields; " + "mm_metadata will use keep_on_cpu fields only." + ) + self._placeholder_metadata_parser_failed = True + return set() + return set( + self._placeholder_metadata_parser.placeholder_metadata_fields(modality) + ) + def _extract_mm_features( + self, engine_input: EngineInput, ) -> MultiModalFeatures | None: - """Extract multimodal metadata from a rendered engine prompt. - - Returns ``None`` for text-only prompts. - """ - if engine_input.get("type") != "multimodal": - return None - - # At this point engine_input is a MultiModalInput TypedDict. - mm_engine_input = cast(MultiModalInput, engine_input) - mm_hashes: MultiModalHashes = mm_engine_input["mm_hashes"] - raw_placeholders: MultiModalPlaceholders = mm_engine_input["mm_placeholders"] - - mm_placeholders = { - modality: [ - PlaceholderRangeInfo(offset=p.offset, length=p.length) for p in ranges - ] - for modality, ranges in raw_placeholders.items() - } - - # Serialize tensor data per modality. - kwargs_data: dict[str, list[str | None]] | None = None - if raw_mm_kwargs := mm_engine_input.get("mm_kwargs"): - kwargs_data = {} - for modality, items in raw_mm_kwargs.items(): - kwargs_data[modality] = [ - encode_mm_kwargs_item(item) if item is not None else None - for item in items - ] - - return MultiModalFeatures( - mm_hashes=mm_hashes, - mm_placeholders=mm_placeholders, - kwargs_data=kwargs_data, + return extract_mm_features( + engine_input, + metadata_fields_for=self._placeholder_metadata_fields, ) diff --git a/vllm/entrypoints/scale_out/token_in_token_out/mm_features.py b/vllm/entrypoints/scale_out/token_in_token_out/mm_features.py new file mode 100644 index 0000000000..d7f89e895e --- /dev/null +++ b/vllm/entrypoints/scale_out/token_in_token_out/mm_features.py @@ -0,0 +1,159 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Helpers for render-time multimodal feature extraction and generate input.""" + +from __future__ import annotations + +from collections.abc import Callable, Collection, Sequence +from typing import cast + +from vllm.entrypoints.scale_out.token_in_token_out.mm_serde import ( + decode_mm_kwargs_item, + encode_mm_kwargs_item, +) +from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( + MultiModalFeatures, + PlaceholderRangeInfo, +) +from vllm.inputs import ( + EngineInput, + MultiModalHashes, + MultiModalInput, + MultiModalPlaceholders, +) +from vllm.multimodal.inputs import ( + MultiModalKwargsItem, + MultiModalKwargsOptionalItems, +) + + +def merge_mm_kwargs_items( + kwargs_item: MultiModalKwargsItem | None, + metadata_item: MultiModalKwargsItem | None, +) -> MultiModalKwargsItem | None: + """Merge full multimodal data with its metadata-only counterpart.""" + if kwargs_item is None: + return metadata_item + if metadata_item is None: + return kwargs_item + return MultiModalKwargsItem({**kwargs_item, **metadata_item}) + + +def _encode_metadata_items( + items: Sequence[MultiModalKwargsItem | None], + *, + declared: Collection[str], +) -> list[str | None]: + """Serialize placeholder-metadata and ``keep_on_cpu`` fields per item.""" + metadata_items: list[str | None] = [] + for item in items: + if item is None: + metadata_items.append(None) + continue + + metadata_item = MultiModalKwargsItem( + { + key: elem + for key, elem in item.items() + if elem.field.keep_on_cpu or key in declared + } + ) + metadata_items.append( + encode_mm_kwargs_item(metadata_item) if metadata_item else None + ) + return metadata_items + + +def _encode_mm_kwargs_with_metadata( + raw_mm_kwargs: MultiModalKwargsOptionalItems, + *, + metadata_fields_for: Callable[[str], Collection[str]] | None = None, +) -> tuple[dict[str, list[str | None]], dict[str, list[str | None]] | None]: + """Serialize full kwargs and their metadata-only subsets per modality.""" + kwargs_data: dict[str, list[str | None]] = {} + metadata_by_modality: dict[str, list[str | None]] = {} + + for modality, items in raw_mm_kwargs.items(): + kwargs_data[modality] = [ + encode_mm_kwargs_item(item) if item is not None else None for item in items + ] + + declared = ( + set(metadata_fields_for(modality)) + if metadata_fields_for is not None + else set() + ) + metadata_items = _encode_metadata_items(items, declared=declared) + if any(item is not None for item in metadata_items): + metadata_by_modality[modality] = metadata_items + + mm_metadata = metadata_by_modality or None + return kwargs_data, mm_metadata + + +def mm_kwargs_from_features( + features: MultiModalFeatures, +) -> dict[str, list[MultiModalKwargsItem | None]]: + """Deserialize ``features`` into per-modality kwargs for ``mm_input``.""" + mm_kwargs: dict[str, list[MultiModalKwargsItem | None]] = {} + kwargs_data = features.kwargs_data or {} + mm_metadata = features.mm_metadata or {} + for modality, hashes in features.mm_hashes.items(): + n = len(hashes) + kwargs_items = [ + decode_mm_kwargs_item(item) if item is not None else None + for item in kwargs_data.get(modality, [None] * n) + ] + metadata_items = [ + decode_mm_kwargs_item(item) if item is not None else None + for item in mm_metadata.get(modality, [None] * n) + ] + mm_kwargs[modality] = [ + merge_mm_kwargs_items(kwargs_item, metadata_item) + for kwargs_item, metadata_item in zip( + kwargs_items, metadata_items, strict=True + ) + ] + return mm_kwargs + + +def extract_mm_features( + engine_input: EngineInput, + *, + metadata_fields_for: Callable[[str], Collection[str]] | None = None, +) -> MultiModalFeatures | None: + """Extract multimodal features from a rendered engine prompt. + + Returns ``None`` for text-only prompts. ``mm_metadata`` keeps the + intersection of processed kwargs that prefill needs after EC transfer: + fields declared as embedding metadata, plus fields marked + ``keep_on_cpu`` (for example M-RoPE grid dims). + """ + if engine_input.get("type") != "multimodal": + return None + + mm_engine_input = cast(MultiModalInput, engine_input) + mm_hashes: MultiModalHashes = mm_engine_input["mm_hashes"] + raw_placeholders: MultiModalPlaceholders = mm_engine_input["mm_placeholders"] + + mm_placeholders = { + modality: [ + PlaceholderRangeInfo(offset=p.offset, length=p.length) for p in ranges + ] + for modality, ranges in raw_placeholders.items() + } + + kwargs_data: dict[str, list[str | None]] | None = None + mm_metadata: dict[str, list[str | None]] | None = None + if raw_mm_kwargs := mm_engine_input.get("mm_kwargs"): + kwargs_data, mm_metadata = _encode_mm_kwargs_with_metadata( + raw_mm_kwargs, + metadata_fields_for=metadata_fields_for, + ) + + return MultiModalFeatures( + mm_hashes=mm_hashes, + mm_placeholders=mm_placeholders, + kwargs_data=kwargs_data, + mm_metadata=mm_metadata, + ) diff --git a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py index ecf75184e7..65643b9e99 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py @@ -44,6 +44,14 @@ class PlaceholderRangeInfo(BaseModel): # placeholder masks that cannot be recomputed from offset+length alone. +def _has_serialized_mm_items( + payload: dict[str, list[str | None]] | None, +) -> bool: + if not payload: + return False + return any(item is not None for items in payload.values() for item in items) + + class MultiModalFeatures(BaseModel): """Lightweight multimodal metadata produced by the render step. @@ -67,6 +75,18 @@ class MultiModalFeatures(BaseModel): ``None`` for metadata-only (cache-hit) responses. """ + mm_metadata: dict[str, list[str | None]] | None = None + """Per-modality serialized metadata for disaggregated prefill. + + Each value is a list parallel to ``mm_hashes[modality]``. A ``str`` + entry is a base64-encoded ``MultiModalKwargsItem`` containing only + placeholder-metadata and ``keep_on_cpu`` fields. ``None`` means that + the metadata is unavailable for that item. Prefill can use this + instead of ``kwargs_data`` only when ``ec_transfer_params`` is also + set, so embeddings arrive through the EC connector rather than from + ``pixel_values``. + """ + @model_validator(mode="after") def _validate_parallel_fields(self) -> "MultiModalFeatures": modalities = set(self.mm_hashes) @@ -76,6 +96,8 @@ class MultiModalFeatures(BaseModel): ) if self.kwargs_data is not None and set(self.kwargs_data) != modalities: raise ValueError("kwargs_data must use the same modalities as mm_hashes") + if self.mm_metadata is not None and set(self.mm_metadata) != modalities: + raise ValueError("mm_metadata must use the same modalities as mm_hashes") flattened_ranges: list[tuple[int, int]] = [] for modality in modalities: @@ -93,6 +115,13 @@ class MultiModalFeatures(BaseModel): raise ValueError( f"{modality} kwargs_data and mm_hashes must have the same length" ) + if ( + self.mm_metadata is not None + and len(self.mm_metadata[modality]) != num_hashes + ): + raise ValueError( + f"{modality} mm_metadata and mm_hashes must have the same length" + ) flattened_ranges.extend( (placeholder.offset, placeholder.offset + placeholder.length) for placeholder in self.mm_placeholders[modality] @@ -156,6 +185,22 @@ class GenerateRequest(BaseModel): raise ValueError("content_parts and features are mutually exclusive") return self + @model_validator(mode="after") + def _require_ec_for_metadata_only(self) -> "GenerateRequest": + features = self.features + if features is None: + return self + if not _has_serialized_mm_items(features.mm_metadata): + return self + if _has_serialized_mm_items(features.kwargs_data): + return self + if self.ec_transfer_params: + return self + raise ValueError( + "features.mm_metadata without features.kwargs_data requires " + "ec_transfer_params so embeddings can be loaded by the EC connector" + ) + sampling_params: SamplingParams """The sampling parameters for the model.""" diff --git a/vllm/entrypoints/scale_out/token_in_token_out/serving.py b/vllm/entrypoints/scale_out/token_in_token_out/serving.py index 7fd8865ade..d3725e3cf7 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/serving.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/serving.py @@ -35,7 +35,6 @@ from vllm.inputs import EngineInput, TokensPrompt, mm_input from vllm.logger import init_logger from vllm.logprobs import Logprob from vllm.multimodal.inputs import ( - MultiModalKwargsItem, MultiModalKwargsItems, PlaceholderRange, ) @@ -45,7 +44,7 @@ from vllm.sampling_params import RequestOutputKind, SamplingParams from vllm.utils.collection_utils import as_list from vllm.utils.serial_utils import numpy2base64 -from .mm_serde import decode_mm_kwargs_item +from .mm_features import mm_kwargs_from_features from .protocol import ( GenerateRequest, GenerateResponse, @@ -177,17 +176,9 @@ class ServingTokens(GenerateBaseServing): for modality, ranges in features.mm_placeholders.items() } - # Deserialize tensor data when present; None → cache hit. - mm_kwargs: dict[str, list[MultiModalKwargsItem | None]] = {} - if features.kwargs_data is not None: - for modality, items in features.kwargs_data.items(): - mm_kwargs[modality] = [ - decode_mm_kwargs_item(item) if item is not None else None - for item in items - ] - else: - for modality, hashes in features.mm_hashes.items(): - mm_kwargs[modality] = [None] * len(hashes) + # Deserialize full tensor data and optional metadata-only data. + # Metadata-only items are valid when ec_transfer_params is set. + mm_kwargs = mm_kwargs_from_features(features) engine_input = mm_input( prompt_token_ids=request.token_ids,