[Frontend] Expose multimodal metadata for disaggregated prefill (#54659)

Signed-off-by: zhouyou9505 <zhouyou9505@gmail.com>
Co-authored-by: Cursor Grok 4.6 <noreply@cursor.com>

docs/serving/online_serving/renderer.md

@@ -27,3 +27,147 @@ VLLM_ENABLE_SCALE_OUT_ENDPOINTS=1 vllm serve <model>
     - 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": "<data-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": ["<base64 MultiModalKwargsItem: pixel_values + image_grid_thw>"]
+    },
+    "mm_metadata": {
+      "image": ["<base64 MultiModalKwargsItem: image_grid_thw only>"]
+    }
+  }
+}
+```
+
+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": ["<base64 MultiModalKwargsItem: image_grid_thw only>"]
+    }
+  },
+  "ec_transfer_params": {
+    "ec_items": [{"mm_hash": "abc123...", "peer_host": "10.0.0.1"}]
+  },
+  "sampling_params": {"max_tokens": 64}
+}
+```

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"}

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)

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,
         )

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,
+    )

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."""
 

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,