From e8ea00b1cfe69673524688d2d1447cc145f3de3e Mon Sep 17 00:00:00 2001 From: Fang Han Date: Tue, 3 Nov 2026 11:48:47 -0800 Subject: [PATCH] [Feature][Multimodal] Continue multimodal requests from existing KV with features.requires_kv Let a decode instance continue a multimodal request from decoder KV that already exists, locally or through a KV connector, without the media and without encoder capacity. The Generate API input carries the prefill-side token ids plus item hashes and placeholder positions, and `features.requires_kv` states that every item must be covered by KV. Engine contract: - `MultiModalFeatures.requires_kv` (omitted from the wire format when false) rejects serialized payloads and empty hashes; `engine_input_from_features` builds the engine input and `serve_tokens` uses it. `MultiModalInput.mm_requires_kv` and `MultiModalFeatureSpec.requires_kv` carry the flag through IPC. - Admission exempts KV-only inputs from the encoder-cache capacity check and validates them instead: nonempty hash and no payload per item, spans within the prompt and ending before the final prompt token (which is always recomputed); prompt logprobs, pooling, M-RoPE and encoder-decoder models are rejected. - The scheduler caches `Request.mm_required_kv_tokens` and fails the request instead of recomputing a span from placeholder ids whenever `num_computed_tokens` is below it, including after preemption and failed KV loads; streaming sessions extend the boundary. - The receiver cache skips `requires_kv` features and the renderers refuse to render such inputs again. Because the decode input carries prefill's hashes and positions, decode computes the same block hashes as prefill, so prefix caching and hash-keyed KV stores see the same blocks on both sides. Compared with the previous revision, the Python export/restore helpers, the kv_handoff descriptor and model fingerprint, the vllm/inputs/mm_features package move and the is_embed round-trip are dropped; the latter will be a separate PR. Closes #60521 Co-authored-by: Codex Co-Authored-By: Claude Fable 5.1 Signed-off-by: Fang Han --- docs/features/disagg_prefill.md | 8 + .../test_generate_stream.py | 47 +++ .../token_in_token_out/test_mm_serde.py | 76 ++++- tests/multimodal/test_mm_requires_kv.py | 316 ++++++++++++++++++ tests/v1/core/test_scheduler.py | 225 +++++++++++++ tests/v1/core/utils.py | 4 +- .../token_in_token_out/mm_features.py | 31 ++ .../scale_out/token_in_token_out/protocol.py | 26 ++ .../scale_out/token_in_token_out/serving.py | 41 +-- vllm/inputs/engine.py | 8 + vllm/multimodal/cache/base.py | 6 + vllm/multimodal/inputs.py | 8 + vllm/renderers/base.py | 12 + vllm/v1/core/sched/scheduler.py | 22 ++ vllm/v1/engine/input_processor.py | 106 +++++- vllm/v1/request.py | 10 + 16 files changed, 900 insertions(+), 46 deletions(-) create mode 100644 tests/multimodal/test_mm_requires_kv.py diff --git a/docs/features/disagg_prefill.md b/docs/features/disagg_prefill.md index 8fd67ca4c20d..de746ca9fa62 100644 --- a/docs/features/disagg_prefill.md +++ b/docs/features/disagg_prefill.md @@ -79,6 +79,14 @@ decode = client.chat.completions.create( If `messages` has non-text content or `echo` is set, the ids are ignored and `messages` is rendered instead, so it must match the prefill request. Otherwise `kv_transfer_params["prompt_token_ids"]` must be a non-empty list of non-negative integers, or the request fails with HTTP 400, as it always does on `/v1/chat/completions/batch`. +## Continuing a multimodal request from existing KV + +A decode instance can continue a multimodal request without the media and without encoder capacity when the prompt's KV already exists, locally or through a KV connector. The [Generate API](../serving/online_serving/token_in_token_out.md) input carries the prefill-side `token_ids` and a `features` object with the item hashes and placeholder ranges, and `features.requires_kv: true` states that every item must be covered by decoder KV. The same hashes and positions make the decode instance compute the same block hashes as prefill, so prefix caching and hash-keyed KV stores see the same blocks on both sides. `--limit-mm-per-prompt` may be 0 on the decode instance: such inputs are exempt from the encoder-cache capacity check. + +The flag is a requirement, not proof that the KV exists. Before scheduling any computation, the engine checks that the computed tokens cover every required span, including after a preemption or a failed KV load, and fails the request (`finish_reason="error"`) rather than recompute an image from placeholder token ids. Items must carry a nonempty hash and no media payload; the prompt must end with at least one token after the last multimodal span, because the final prompt token is always recomputed; prompt logprobs, pooling, M-RoPE models and encoder-decoder models are not supported. Local prefix-cache hits are whole blocks and never include the block holding the final prompt token, so without a connector the spans must end within the first `floor((N - 1) / block_size) * block_size` tokens of an `N`-token prompt. + +Python callers submit the same input directly to `AsyncLLM.generate()` or `LLMEngine.add_request()` as a `MultiModalInput` with `mm_kwargs` set to `None` per item and `mm_requires_kv=True`. `LLM.generate()` and the renderers reject such inputs, because rendering them again would drop the requirement and the media identities. + ## Generate API output modes When the prefill and decode stages use the [Generate API](../serving/online_serving/token_in_token_out.md) (`/inference/v1/generate`), only the decode response reaches the client, so set `output_mode` on the decode request only. A proxy that reuses the client's request body for the prefill request must reset `output_mode` to `tokens` there. A prefill instance started with `--tokens-only` has no tokenizer and rejects `output_mode: "text"` with a 400. diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py b/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py index 4ac47ef11125..64bb462c86ba 100644 --- a/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py +++ b/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py @@ -1878,3 +1878,50 @@ async def mock_generate(*args, **kwargs): .delta.content == "a" ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("remote_kv", [False, True]) +async def test_serve_tokens_requires_kv_features_skip_rendering_and_media(remote_kv): + engine = _mock_engine() + payload: dict[str, Any] = { + "token_ids": [1, 2, 2, 2, 3], + "features": { + "mm_hashes": {"image": ["image-a"]}, + "mm_placeholders": {"image": [{"offset": 1, "length": 3}]}, + "requires_kv": True, + }, + "cache_salt": "tenant-a", + "sampling_params": {"max_tokens": 1}, + "return_token_ids": True, + } + if remote_kv: + payload["kv_transfer_params"] = {"do_remote_prefill": True} + request = GenerateRequest.model_validate_json(json.dumps(payload)) + + async def mock_generate(prompt, sampling_params, *args, **kwargs): + assert prompt["prompt_token_ids"] == payload["token_ids"] + assert prompt["mm_hashes"] == {"image": ["image-a"]} + assert prompt["mm_kwargs"] == {"image": [None]} + assert prompt["mm_requires_kv"] is True + assert prompt["cache_salt"] == "tenant-a" + (placeholder,) = prompt["mm_placeholders"]["image"] + assert (placeholder.offset, placeholder.length) == (1, 3) + if remote_kv: + assert sampling_params.extra_args["kv_transfer_params"] == { + "do_remote_prefill": True + } + yield _make_request_output( + "req-1", token_ids=[10], finish_reason="stop", finished=True + ) + + engine.generate = MagicMock(side_effect=mock_generate) + serving = _build_serving_tokens(engine) + engine.renderer.render_cmpl_async = AsyncMock( + side_effect=AssertionError("processed features must not be rendered again") + ) + response = await serving.serve_tokens(request) + assert isinstance(response, GenerateTokensResponse) + engine.generate.assert_called_once() + serving.online_renderer.preprocess_completion.assert_not_awaited() + engine.renderer.render_cmpl_async.assert_not_awaited() 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 da994d7501ef..f9318276525b 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 @@ -13,6 +13,7 @@ from vllm.entrypoints.openai.completion.protocol import CompletionRequest from vllm.entrypoints.openai.responses.protocol import ResponsesRequest from vllm.entrypoints.scale_out.token_in_token_out.mm_features import ( + engine_input_from_features, extract_mm_features, merge_mm_kwargs_items, mm_kwargs_from_features, @@ -26,7 +27,7 @@ MultiModalFeatures, PlaceholderRangeInfo, ) -from vllm.inputs import mm_input +from vllm.inputs import MultiModalInput, mm_input from vllm.multimodal.inputs import ( MultiModalBatchedField, MultiModalFieldElem, @@ -131,7 +132,7 @@ def _image_engine_input( *, pixel_keep_on_cpu: bool = False, grid_keep_on_cpu: bool = True, -) -> tuple[object, MultiModalFieldElem, MultiModalFieldElem]: +) -> tuple[MultiModalInput, MultiModalFieldElem, MultiModalFieldElem]: pixel_values = MultiModalFieldElem( data=torch.randn(5, 3, dtype=torch.float32), field=MultiModalBatchedField(keep_on_cpu=pixel_keep_on_cpu), @@ -422,3 +423,74 @@ def test_metadata_can_replace_or_extend_full_mm_data(): merged = merge_mm_kwargs_items(full_item, metadata_item) assert merged is not None assert set(merged) == {"pixel_values", "image_grid_thw"} + + +def test_requires_kv_is_omitted_from_wire_format_when_false(): + engine_input, _, _ = _image_engine_input() + features = extract_mm_features(engine_input) + assert features is not None + assert features.requires_kv is False + assert "requires_kv" not in features.model_dump() + assert "requires_kv" not in features.model_dump_json() + restored = MultiModalFeatures.model_validate_json(features.model_dump_json()) + assert restored.requires_kv is False + + +@pytest.mark.parametrize("has_transfer", [False, True]) +def test_requires_kv_restores_media_free_engine_input(has_transfer): + engine_input, _, _ = _image_engine_input() + features = extract_mm_features(engine_input).model_dump() + features["kwargs_data"] = None + features["mm_metadata"] = None + features["requires_kv"] = True + request = GenerateRequest.model_validate( + { + "token_ids": engine_input["prompt_token_ids"], + "features": features, + "sampling_params": {}, + "cache_salt": "tenant-a", + "kv_transfer_params": {"remote": True} if has_transfer else None, + } + ) + assert request.features is not None + assert request.features.requires_kv is True + restored = engine_input_from_features( + request.token_ids, request.features, cache_salt=request.cache_salt + ) + assert restored["mm_requires_kv"] is True + assert restored["cache_salt"] == "tenant-a" + assert restored["mm_hashes"] == engine_input["mm_hashes"] + assert restored["mm_kwargs"]["image"][0] is None + assert [(p.offset, p.length) for p in restored["mm_placeholders"]["image"]] == [ + (p.offset, p.length) for p in engine_input["mm_placeholders"]["image"] + ] + + +def test_features_without_kv_requirement_restore_their_payloads(): + engine_input, _, _ = _image_engine_input() + features = extract_mm_features(engine_input) + restored = engine_input_from_features(engine_input["prompt_token_ids"], features) + assert restored.get("mm_requires_kv", False) is False + assert "pixel_values" in restored["mm_kwargs"]["image"][0] + + +@pytest.mark.parametrize( + ("change", "error"), + [ + ("payload", "serialized media payloads"), + ("metadata", "serialized media payloads"), + ("hash", "nonempty hash"), + ], +) +def test_requires_kv_rejects_media_payloads_and_empty_hashes(change, error): + engine_input, _, _ = _image_engine_input() + features = extract_mm_features(engine_input).model_dump() + if change == "metadata": + features["mm_metadata"] = features["kwargs_data"] + if change != "payload": + features["kwargs_data"] = None + if change == "hash": + features["mm_hashes"]["image"][0] = "" + features["requires_kv"] = True + with pytest.raises(ValidationError, match=error): + MultiModalFeatures.model_validate(features) diff --git a/tests/multimodal/test_mm_requires_kv.py b/tests/multimodal/test_mm_requires_kv.py new file mode 100644 index 000000000000..493afabf5256 --- /dev/null +++ b/tests/multimodal/test_mm_requires_kv.py @@ -0,0 +1,316 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Engine-side contract for multimodal inputs that require existing KV. + +A decode instance without media or encoder capacity continues a multimodal +request purely from decoder KV. Such an input carries hashes and placeholder +positions but no media and sets ``mm_requires_kv``: admission accepts it with a +zero encoder budget, its block hashes equal the prefill engine's, the receiver +cache skips it, and renderers refuse to render it again. The scheduler side +(fail the request instead of computing when KV coverage is missing) is covered +in tests/v1/core/test_scheduler.py. +""" + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +import torch + +from vllm.entrypoints.common.offline import OfflineInferenceMixin +from vllm.exceptions import VLLMValidationError +from vllm.inputs import MultiModalInput, mm_input +from vllm.multimodal.cache import ( + LruKeyReplicatedReceiverCache, + MultiModalCacheMissError, +) +from vllm.multimodal.inputs import ( + MultiModalFeatureSpec, + MultiModalKwargsItem, + MultiModalKwargsItems, + PlaceholderRange, +) +from vllm.pooling_params import PoolingParams +from vllm.renderers.base import BaseRenderer +from vllm.renderers.params import TokenizeParams +from vllm.sampling_params import SamplingParams +from vllm.utils.hashing import sha256 +from vllm.v1.core.kv_cache_utils import get_request_block_hasher, init_none_hash +from vllm.v1.engine.input_processor import InputProcessor +from vllm.v1.request import Request +from vllm.v1.serial_utils import MsgpackDecoder, MsgpackEncoder + +PROMPT_TOKEN_IDS = [1, 99, 99, 99, 2, 99, 99, 3] + + +def rendered_input(image_hash: str = "a") -> MultiModalInput: + """A processed prompt as the renderer hands it to the prefill engine.""" + return mm_input( + prompt_token_ids=list(PROMPT_TOKEN_IDS), + mm_kwargs={"image": [None, None]}, + mm_hashes={"image": [image_hash, "second-image"]}, + mm_placeholders={ + "image": [ + PlaceholderRange(offset=1, length=3), + PlaceholderRange(offset=5, length=2), + ] + }, + cache_salt="tenant-a", + ) + + +def kv_only_input(image_hash: str = "a") -> MultiModalInput: + """The same prompt as a decode instance receives it: no media, KV required.""" + prompt = rendered_input(image_hash) + prompt["mm_requires_kv"] = True + return prompt + + +def without_text_suffix(prompt: MultiModalInput) -> MultiModalInput: + # Extend the last image span onto the final prompt token. + prompt["mm_placeholders"] = { + "image": [ + PlaceholderRange(offset=1, length=3), + PlaceholderRange(offset=5, length=3), + ] + } + return prompt + + +def test_block_hashes_match_prefill_and_differ_across_images(): + init_none_hash(sha256) + + def block_hashes(prompt: MultiModalInput): + features = [ + MultiModalFeatureSpec( + data=None, + modality="image", + identifier=mm_hash, + mm_hash=mm_hash, + mm_position=position, + requires_kv=prompt.get("mm_requires_kv", False), + ) + for mm_hash, position in zip( + prompt["mm_hashes"]["image"], prompt["mm_placeholders"]["image"] + ) + ] + return Request( + request_id="any-request", + prompt_token_ids=prompt["prompt_token_ids"], + sampling_params=SamplingParams(max_tokens=1), + pooling_params=None, + mm_features=features, + cache_salt=prompt["cache_salt"], + block_hasher=get_request_block_hasher(2, sha256), + ).block_hashes + + prefill = block_hashes(rendered_input("a")) + decode = block_hashes(kv_only_input("a")) + other = block_hashes(kv_only_input("b")) + # The continuation reproduces prefill's block hashes exactly, so the local + # prefix cache and every hash-keyed KV consumer see the same blocks on both + # sides, while a different image still changes every hash. + assert prefill == decode + assert all(a != b for a, b in zip(decode, other)) + + +def test_receiver_cache_skips_only_kv_only_features(): + cache = LruKeyReplicatedReceiverCache( + SimpleNamespace( + get_multimodal_config=lambda: SimpleNamespace(mm_processor_cache_gb=0.01) + ) + ) + feature = MultiModalFeatureSpec( + None, + "image", + "image-a", + PlaceholderRange(1, 3), + mm_hash="image-a", + requires_kv=True, + ) + assert cache.get_and_update_features([feature])[0].data is None + feature.requires_kv = False + with pytest.raises(MultiModalCacheMissError): + cache.get_and_update_features([feature]) + + +def test_kv_requirement_survives_engine_ipc_serialization(): + feature = MultiModalFeatureSpec( + None, + "image", + "a", + PlaceholderRange(1, 3, torch.tensor([True, False, True])), + mm_hash="a", + requires_kv=True, + ) + restored = MsgpackDecoder(MultiModalFeatureSpec).decode( + MsgpackEncoder().encode(feature) + ) + assert restored.requires_kv is True + assert restored.data is None + assert restored.identifier == feature.identifier + assert torch.equal(restored.mm_position.is_embed, feature.mm_position.is_embed) + + +@pytest.fixture +def admission_processor(monkeypatch): + """A bare InputProcessor for a decode instance with no encoder capacity.""" + processor = object.__new__(InputProcessor) + processor.model_config = SimpleNamespace( + max_model_len=64, + runner_type="generate", + is_encoder_decoder=False, + uses_mrope=False, + ) + processor.vllm_config = SimpleNamespace( + model_config=processor.model_config, + lora_config=None, + speculative_config=None, + ec_transfer_config=None, + kv_transfer_config=None, + watermark_config=None, + parallel_config=SimpleNamespace( + data_parallel_size=1, data_parallel_size_local=1, local_engines_only=False + ), + cache_config=SimpleNamespace(kv_sharing_fast_prefill=False), + ) + processor.lora_config = None + processor.renderer = SimpleNamespace( + tokenizer=None, get_eos_token_id=lambda: 2, validate_token_ids=Mock() + ) + processor.generation_config_fields = {} + processor.skip_prompt_length_check = False + processor.supports_mm_inputs = True + processor.mm_encoder_cache_size = 0 + monkeypatch.setattr(processor, "_validate_params", Mock(return_value=False)) + return processor + + +@pytest.mark.parametrize("encoder_cache_size", [0, 16]) +def test_kv_only_input_is_admitted_without_encoder_capacity( + admission_processor, encoder_cache_size +): + admission_processor.mm_encoder_cache_size = encoder_cache_size + request = admission_processor.process_inputs( + "decode", kv_only_input(), SamplingParams(max_tokens=1), ("generate",) + ) + assert request.prompt_token_ids == PROMPT_TOKEN_IDS + assert request.cache_salt == "tenant-a" + assert request.mm_features is not None + assert [f.identifier for f in request.mm_features] == ["a", "second-image"] + assert all(f.requires_kv and f.data is None for f in request.mm_features) + + +@pytest.mark.parametrize( + ("invalid", "error"), + [ + ("suffix", "after the last multimodal span"), + ("media", "no media payload"), + ("hash", "nonempty hash"), + ("bounds", "exceeds the prompt length"), + ("prompt_logprobs", "cannot be rescored"), + ("pooling", "cannot be rescored"), + ("mrope", "M-RoPE"), + ], +) +def test_kv_only_admission_rejects_invalid_contracts( + admission_processor, invalid, error +): + prompt = kv_only_input() + params: SamplingParams | PoolingParams = SamplingParams(max_tokens=1) + if invalid == "suffix": + without_text_suffix(prompt) + elif invalid == "media": + prompt["mm_kwargs"] = MultiModalKwargsItems( + {"image": [MultiModalKwargsItem({}), None]} + ) + elif invalid == "hash": + prompt["mm_hashes"] = {"image": ["", "second-image"]} + elif invalid == "bounds": + prompt["mm_placeholders"] = { + "image": [ + PlaceholderRange(offset=1, length=3), + PlaceholderRange(offset=5, length=20), + ] + } + elif invalid == "prompt_logprobs": + params = SamplingParams(max_tokens=1, prompt_logprobs=1) + elif invalid == "pooling": + params = PoolingParams() + else: + admission_processor.model_config.uses_mrope = True + with pytest.raises(VLLMValidationError, match=error): + admission_processor.process_inputs("decode", prompt, params, ("generate",)) + + +@pytest.mark.parametrize("has_media", [False, True]) +def test_ordinary_multimodal_input_still_requires_encoder_capacity( + admission_processor, has_media +): + prompt = rendered_input() + if has_media: + prompt["mm_kwargs"] = MultiModalKwargsItems( + {"image": [MultiModalKwargsItem({}), None]} + ) + with pytest.raises(VLLMValidationError, match="pre-allocated encoder cache size 0"): + admission_processor.process_inputs( + "prefill", prompt, SamplingParams(max_tokens=1), ("generate",) + ) + + +@pytest.mark.parametrize("invalid", ["length", "token_ids"]) +def test_kv_only_admission_still_validates_prompt(admission_processor, invalid): + prompt = kv_only_input() + if invalid == "length": + admission_processor.model_config.max_model_len = 4 + error = "maximum model length" + else: + error = "invalid token ID" + admission_processor.renderer.validate_token_ids.side_effect = ( + VLLMValidationError(error) + ) + with pytest.raises(VLLMValidationError, match=error): + admission_processor.process_inputs( + "decode", prompt, SamplingParams(max_tokens=1), ("generate",) + ) + + +@pytest.fixture +def prompt_renderer(): + class TestRenderer(BaseRenderer): + def __init__(self): + self.tokenizer = None + self.mm_processor = None + self.model_config = SimpleNamespace( + max_model_len=64, encoder_config=None, is_encoder_decoder=False + ) + + def render_messages(self, *args, **kwargs): + raise NotImplementedError + + return TestRenderer() + + +def test_offline_preprocessing_rejects_only_kv_only_inputs(prompt_renderer): + # Exercise the real LLM.generate preprocessing path, including its renderer. + llm = OfflineInferenceMixin() + llm.renderer = prompt_renderer + llm.model_config = prompt_renderer.model_config + with pytest.raises(VLLMValidationError, match="AsyncLLM.generate"): + llm._preprocess_cmpl([kv_only_input()]) # type: ignore[list-item] + # Pre-rendered inputs without the KV requirement keep their existing path + # (tests/entrypoints/multimodal/llm/test_mm_cache_external_injection.py). + prerendered = rendered_input() + (kept,) = llm._preprocess_cmpl([prerendered]) # type: ignore[list-item] + assert kept.get("prompt_token_ids") == prerendered["prompt_token_ids"] + (ordinary,) = llm._preprocess_cmpl([{"prompt_token_ids": [1, 2, 3]}]) + assert ordinary["type"] == "token" + assert ordinary["prompt_token_ids"] == [1, 2, 3] + + +@pytest.mark.asyncio +async def test_async_renderer_rejects_kv_only_input(prompt_renderer): + with pytest.raises(VLLMValidationError, match="AsyncLLM.generate"): + await prompt_renderer.render_cmpl_async( + [kv_only_input()], TokenizeParams(max_total_tokens=64) + ) diff --git a/tests/v1/core/test_scheduler.py b/tests/v1/core/test_scheduler.py index 4bb9fc337ac7..0ade40d64ab3 100644 --- a/tests/v1/core/test_scheduler.py +++ b/tests/v1/core/test_scheduler.py @@ -5147,6 +5147,7 @@ def test_abort_request_when_structured_output_fsm_cannot_advance(): scheduler.aux_output_connector = None scheduler.grammar_compile_error_reqs = {} scheduler.encoder_cache_mismatch_reqs = set() + scheduler.missing_multimodal_kv_reqs = set() scheduler.vllm_config = Mock() scheduler.return_sampling_mask = False scheduler.recompute_kv_load_failures = False @@ -8877,3 +8878,227 @@ def test_cpu_encoder_input_updates_preserve_shm_handles(use_v2_model_runner): {request.request_id: {1: None}} if use_v2_model_runner else {} ) assert request.mm_features[0].data is handle + + +@pytest.mark.parametrize("computed", [0, 49, 50, 149]) +def test_media_free_request_without_complete_kv_fails_without_encoder_work(computed): + scheduler = create_scheduler(model="llava-hf/llava-1.5-7b-hf") + request = create_requests( + num_requests=1, + num_tokens=160, + mm_positions=[[PlaceholderRange(offset=50, length=100)]], + mm_requires_kv=True, + )[0] + scheduler.add_request(request) + request.num_computed_tokens = computed + output = scheduler.schedule() + assert request.request_id not in output.num_scheduled_tokens + assert not output.scheduled_encoder_inputs + result = scheduler.update_from_output( + output, ModelRunnerOutput(req_ids=[], req_id_to_index={}) + ) + assert request.status == RequestStatus.FINISHED_ERROR + assert result[request.client_index].outputs[0].finish_reason == FinishReason.ERROR + assert request.request_id not in scheduler.requests + + +def test_media_free_coverage_is_rechecked_after_preemption(): + scheduler = create_scheduler(model="llava-hf/llava-1.5-7b-hf") + request = create_requests( + num_requests=1, + num_tokens=160, + mm_positions=[[PlaceholderRange(offset=50, length=100)]], + mm_requires_kv=True, + )[0] + # Imported/local KV covers the image, so text continuation needs no encoder. + assert scheduler._try_schedule_encoder_inputs(request, 150, 10, 100) == ( + [], + 10, + 100, + [], + [], + ) + request.status = RequestStatus.RUNNING + scheduler._preempt_request(request, 0) + assert request.num_computed_tokens == 0 + assert scheduler._try_schedule_encoder_inputs(request, 0, 160, 100) == ( + [], + 0, + 100, + [], + [], + ) + assert request.request_id in scheduler.missing_multimodal_kv_reqs + + +def test_media_free_async_load_waits_without_claiming_coverage(): + scheduler = create_scheduler(model="llava-hf/llava-1.5-7b-hf") + request = create_requests( + num_requests=1, + num_tokens=160, + mm_positions=[[PlaceholderRange(offset=50, length=100)]], + mm_requires_kv=True, + )[0] + assert scheduler._try_schedule_encoder_inputs(request, 0, 0, 100) == ( + [], + 0, + 100, + [], + [], + ) + assert not scheduler.missing_multimodal_kv_reqs + + +@pytest.mark.parametrize("outcome", ["success", "failure", "cancel"]) +@pytest.mark.parametrize("recompute", [False, True]) +def test_media_free_remote_kv_lifecycle(outcome, recompute): + scheduler = create_scheduler( + model="llava-hf/llava-1.5-7b-hf", + enable_prefix_caching=True, + use_kv_connector=mock_kv(matched_tokens=160, is_async=True), + ) + scheduler.recompute_kv_load_failures = recompute + request = create_requests( + num_requests=1, + num_tokens=160, + mm_positions=[[PlaceholderRange(offset=50, length=100)]], + mm_requires_kv=True, + )[0] + scheduler.add_request(request) + loading = scheduler.schedule() + assert request.status == RequestStatus.WAITING_FOR_REMOTE_KVS + assert not loading.scheduled_encoder_inputs + if outcome == "cancel": + scheduler.finish_requests([request.request_id], RequestStatus.FINISHED_ABORTED) + received = KVConnectorOutput(finished_recving={request.request_id}) + if outcome == "failure": + received.failed_recving = {request.request_id} + outputs = scheduler.update_from_output( + loading, + ModelRunnerOutput(req_ids=[], req_id_to_index={}, kv_connector_output=received), + ) + if outcome == "success": + resumed = scheduler.schedule() + assert request.request_id in resumed.num_scheduled_tokens + assert not resumed.scheduled_encoder_inputs + assert not scheduler.missing_multimodal_kv_reqs + elif outcome == "failure": + if recompute: + # The connector cannot supply the missing KV on retry either. + scheduler.connector.config = dataclasses.replace( + scheduler.connector.config, matched_tokens=0, is_async=False + ) + resumed = scheduler.schedule() + assert request.request_id not in resumed.num_scheduled_tokens + assert not resumed.scheduled_encoder_inputs + outputs = scheduler.update_from_output( + resumed, ModelRunnerOutput(req_ids=[], req_id_to_index={}) + ) + assert request.status == RequestStatus.FINISHED_ERROR + assert ( + outputs[request.client_index].outputs[0].finish_reason == FinishReason.ERROR + ) + else: + assert request.status == RequestStatus.FINISHED_ABORTED + + +def test_media_free_sync_load_failure_cannot_recompute_missing_media(): + scheduler = create_scheduler( + model="llava-hf/llava-1.5-7b-hf", + use_kv_connector=mock_kv(matched_tokens=160, is_async=False), + ) + scheduler.recompute_kv_load_failures = True + request = create_requests( + num_requests=1, + num_tokens=176, + mm_positions=[[PlaceholderRange(offset=50, length=100)]], + mm_requires_kv=True, + )[0] + scheduler.add_request(request) + scheduled = scheduler.schedule() + invalid_block = scheduled.scheduled_new_reqs[0].block_ids[0][5] + outputs = scheduler.update_from_output( + scheduled, + ModelRunnerOutput( + req_ids=[request.request_id], + req_id_to_index={request.request_id: 0}, + sampled_token_ids=[[1]], + kv_connector_output=KVConnectorOutput(invalid_block_ids={invalid_block}), + ), + ) + assert not any(output.outputs for output in outputs.values()) + resumed = scheduler.schedule() + assert not resumed.num_scheduled_tokens + assert not resumed.scheduled_encoder_inputs + scheduler.update_from_output( + resumed, ModelRunnerOutput(req_ids=[], req_id_to_index={}) + ) + assert request.status == RequestStatus.FINISHED_ERROR + + +@pytest.mark.parametrize("image_end", [150, 170]) +def test_media_free_local_prefix_cache_coverage_without_connector(image_end): + scheduler = create_scheduler( + model="llava-hf/llava-1.5-7b-hf", enable_prefix_caching=True + ) + assert scheduler.connector is None + kwargs = dict( + num_requests=1, + num_tokens=176, + mm_positions=[[PlaceholderRange(offset=50, length=image_end - 50)]], + mm_hashes_list=[["same-image"]], + max_tokens=1, + ) + prefill = create_requests(**kwargs, req_ids=["prefill"])[0] + scheduler.add_request(prefill) + scheduled = scheduler.schedule() + scheduler.update_from_output( + scheduled, + ModelRunnerOutput( + req_ids=[prefill.request_id], + req_id_to_index={prefill.request_id: 0}, + sampled_token_ids=[[1]], + ), + ) + assert prefill.is_finished() + decode = create_requests(**kwargs, req_ids=["decode"], mm_requires_kv=True)[0] + scheduler.add_request(decode) + resumed = scheduler.schedule() + # A full prompt hit replays its final block, leaving 160 cached tokens. + assert not resumed.scheduled_encoder_inputs + if image_end <= 160: + assert resumed.num_scheduled_tokens[decode.request_id] == 16 + assert not scheduler.missing_multimodal_kv_reqs + else: + assert decode.request_id not in resumed.num_scheduled_tokens + scheduler.update_from_output( + resumed, ModelRunnerOutput(req_ids=[], req_id_to_index={}) + ) + assert decode.status == RequestStatus.FINISHED_ERROR + + +def test_media_free_streaming_update_extends_required_kv_boundary(): + from vllm.v1.request import StreamingUpdate + + scheduler = create_scheduler(model="llava-hf/llava-1.5-7b-hf") + session = create_requests(num_requests=1, num_tokens=10)[0] + session.append_output_token_ids(0) # Final sampled token of the previous chunk. + session.num_computed_tokens = 10 + continuation = create_requests( + num_requests=1, + num_tokens=10, + mm_positions=[[PlaceholderRange(offset=3, length=4)]], + mm_requires_kv=True, + )[0] + continuation.resumable = True + update = StreamingUpdate.from_request(continuation) + assert update is not None + scheduler._update_request_as_session(session, update) + assert session.mm_required_kv_tokens == 17 + assert scheduler._try_schedule_encoder_inputs(session, 10, 10, 100) == ( + [], + 0, + 100, + [], + [], + ) diff --git a/tests/v1/core/utils.py b/tests/v1/core/utils.py index 7eb26ae7f996..307a255090f8 100644 --- a/tests/v1/core/utils.py +++ b/tests/v1/core/utils.py @@ -270,6 +270,7 @@ def create_requests( block_size: int = 16, req_ids: list[str] | None = None, sampling_params: SamplingParams | None = None, + mm_requires_kv: bool = False, ) -> list[Request]: global _none_hash_initialized if not _none_hash_initialized: @@ -328,10 +329,11 @@ def create_requests( # Unique dummy hash for each mm item identifier = f"hash{i}_{j}" mm_feature = MultiModalFeatureSpec( - data=MultiModalKwargsItem.dummy(), + data=None if mm_requires_kv else MultiModalKwargsItem.dummy(), mm_position=position, identifier=identifier, modality="image", + requires_kv=mm_requires_kv, ) mm_features.append(mm_feature) 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 index 124d60286797..d86887a75623 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/mm_features.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/mm_features.py @@ -21,10 +21,13 @@ MultiModalHashes, MultiModalInput, MultiModalPlaceholders, + mm_input, ) from vllm.multimodal.inputs import ( MultiModalKwargsItem, + MultiModalKwargsItems, MultiModalKwargsOptionalItems, + PlaceholderRange, ) @@ -118,6 +121,34 @@ def mm_kwargs_from_features( return mm_kwargs +def engine_input_from_features( + token_ids: list[int], + features: MultiModalFeatures, + *, + cache_salt: str | None = None, +) -> MultiModalInput: + """Build the processed engine input described by wire-format features. + + Items without inline data stay `None` and are resolved from the processor + cache on admission, unless `features.requires_kv` is set, in which case the + engine requires existing decoder KV for every item instead. + """ + mm_placeholders: dict[str, list[PlaceholderRange]] = { + modality: [PlaceholderRange(offset=p.offset, length=p.length) for p in ranges] + for modality, ranges in features.mm_placeholders.items() + } + engine_input = mm_input( + prompt_token_ids=list(token_ids), + mm_kwargs=MultiModalKwargsItems(mm_kwargs_from_features(features)), + mm_hashes=features.mm_hashes, + mm_placeholders=mm_placeholders, + cache_salt=cache_salt, + ) + if features.requires_kv: + engine_input["mm_requires_kv"] = True + return engine_input + + def placeholder_ranges_from_engine_input( engine_input: EngineInput, ) -> dict[str, list[PlaceholderRangeInfo]] | None: 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 188009f8ac59..9f8fc3294793 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py @@ -160,6 +160,32 @@ class MultiModalFeatures(BaseModel): `pixel_values`. """ + requires_kv: bool = Field(default=False, exclude_if=lambda value: not value) + """All items must already be covered by decoder KV; none is computed here. + + This is a requirement, not proof that the KV exists: the engine checks the + actual coverage when scheduling, including after preemption and failed KV + loads, and fails the request rather than recompute the media. It holds for + local prefix KV as much as for KV loaded through a connector, so it does + not need `kv_transfer_params`. Items carry hashes and placeholder positions + only: `kwargs_data` and `mm_metadata` must not carry payloads. + """ + + @model_validator(mode="after") + def _validate_kv_requirement(self) -> "MultiModalFeatures": + if not self.requires_kv: + return self + if not self.mm_hashes or any( + not hashes or any(not h for h in hashes) + for hashes in self.mm_hashes.values() + ): + raise ValueError("requires_kv requires a nonempty hash per item") + if _has_serialized_mm_items(self.kwargs_data) or _has_serialized_mm_items( + self.mm_metadata + ): + raise ValueError("requires_kv does not accept serialized media payloads") + return self + @model_validator(mode="after") def _validate_parallel_fields(self) -> "MultiModalFeatures": modalities = set(self.mm_hashes) 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 1f4af4ad9b83..3bb92e2b46ad 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/serving.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/serving.py @@ -6,7 +6,7 @@ import itertools import math import time -from collections.abc import AsyncGenerator, Callable +from collections.abc import AsyncGenerator, Callable, Mapping from collections.abc import Sequence as GenericSequence from typing import Any @@ -44,14 +44,10 @@ from vllm.entrypoints.serve.utils.api_utils import get_max_tokens, should_include_usage from vllm.entrypoints.serve.utils.request_logger import RequestLogger from vllm.exceptions import GenerationError, VLLMValidationError -from vllm.inputs import EngineInput, TokensPrompt, mm_input +from vllm.inputs import EngineInput, TokensPrompt from vllm.logger import init_logger from vllm.logprobs import Logprob -from vllm.multimodal.inputs import ( - MultiModalKwargsItem, - MultiModalKwargsItems, - PlaceholderRange, -) +from vllm.multimodal.inputs import MultiModalKwargsItem from vllm.outputs import RequestOutput from vllm.renderers.chat_utils import AsyncMultiModalItemTracker from vllm.renderers.online_derenderer import OnlineDerenderer @@ -62,7 +58,7 @@ from vllm.utils.serial_utils import numpy2base64 from .mm_features import ( - mm_kwargs_from_features, + engine_input_from_features, placeholder_ranges_from_engine_input, ) from .protocol import ( @@ -194,7 +190,7 @@ def __init__( def _validate_mm_cache_handles( self, - mm_kwargs: dict[str, list[MultiModalKwargsItem | None]], + mm_kwargs: Mapping[str, GenericSequence[MultiModalKwargsItem | None]], mm_hashes: dict[str, list[str]], ) -> ErrorResponse | None: cache = self.online_renderer.renderer.mm_processor_cache @@ -333,27 +329,18 @@ async def serve_tokens( [prompt] ) elif features := request.features: - # Convert PlaceholderRangeInfo → PlaceholderRange per modality. - mm_placeholders: dict[str, list[PlaceholderRange]] = { - modality: [ - PlaceholderRange(offset=p.offset, length=p.length) for p in ranges - ] - for modality, ranges in features.mm_placeholders.items() - } - # 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) - if error := self._validate_mm_cache_handles(mm_kwargs, features.mm_hashes): - return error - - engine_input = mm_input( - prompt_token_ids=request.token_ids, - mm_kwargs=MultiModalKwargsItems(mm_kwargs), - mm_hashes=features.mm_hashes, - mm_placeholders=mm_placeholders, + # Metadata-only items are valid when ec_transfer_params is set; + # requires_kv items carry no payload and skip the cache check. + engine_input = engine_input_from_features( + request.token_ids, + features, cache_salt=request.cache_salt, ) + if error := self._validate_mm_cache_handles( + engine_input["mm_kwargs"], features.mm_hashes + ): + return error else: (engine_input,) = await self.online_renderer.preprocess_completion( request, diff --git a/vllm/inputs/engine.py b/vllm/inputs/engine.py index aa8c9054335e..1f28db78e69e 100644 --- a/vllm/inputs/engine.py +++ b/vllm/inputs/engine.py @@ -144,6 +144,14 @@ class MultiModalInput(_InputOptions): mm_hashes: MultiModalHashes """The hashes of the multi-modal data.""" + mm_requires_kv: NotRequired[bool] + """Every multi-modal item must already be covered by decoder KV. + + The items carry hashes and placeholder positions but no media, so the + engine cannot compute them. Missing KV coverage fails the request instead + of being treated as a cache miss. + """ + mm_placeholders: MultiModalPlaceholders """ For each modality, information about the placeholder tokens in diff --git a/vllm/multimodal/cache/base.py b/vllm/multimodal/cache/base.py index 287cf7d5823d..6a8a871679a7 100644 --- a/vllm/multimodal/cache/base.py +++ b/vllm/multimodal/cache/base.py @@ -474,11 +474,17 @@ def get_and_update_features( identifier for backward compatibility). """ for feature in mm_features: + if feature.requires_kv: + # data=None is a deliberate absence (decoder KV supplies the + # media), not a cache reference; a lookup would report a miss. + continue cache_key = feature.mm_hash or feature.identifier self.touch_receiver_cache_item(cache_key, feature.data) missing_mm_hashes: list[str] = [] for feature in mm_features: + if feature.requires_kv: + continue cache_key = feature.mm_hash or feature.identifier try: feature.data = self.get_and_update_item(feature.data, cache_key) diff --git a/vllm/multimodal/inputs.py b/vllm/multimodal/inputs.py index 91fa34e86422..950761812cca 100644 --- a/vllm/multimodal/inputs.py +++ b/vllm/multimodal/inputs.py @@ -372,6 +372,14 @@ class MultiModalFeatureSpec: mm_hash: str | None = None """The hash for caching processor outputs (without LoRA prefix).""" + requires_kv: bool = False + """No media payload exists: this feature must be covered by decoder KV. + + Unlike a receiver-cache reference (`data=None`), this bypasses the media + cache. Scheduling any computation before the end of this feature fails + the request, including after preemption or a partial KV load. + """ + @staticmethod def gather_kwargs(features: list["MultiModalFeatureSpec"], keys: set[str]): kwargs = defaultdict[str, list[NestedTensors]](list) diff --git a/vllm/renderers/base.py b/vllm/renderers/base.py index 395b1ead7acf..eb999726edc7 100644 --- a/vllm/renderers/base.py +++ b/vllm/renderers/base.py @@ -694,6 +694,16 @@ async def _detokenize_prompt_async(self, prompt: TokensPrompt) -> TokensPrompt: return prompt + @staticmethod + def _reject_kv_only_prompt(prompt: SingletonDictPrompt) -> None: + if prompt.get("mm_requires_kv"): # type: ignore[typeddict-item] + raise VLLMValidationError( + "Inputs that require existing multimodal KV coverage cannot be " + "rendered again: rendering would drop the KV requirement and the " + "media identities. Submit them directly to AsyncLLM.generate() " + "or LLMEngine.add_request()." + ) + @overload def _tokenize_singleton_prompt( self, @@ -713,6 +723,7 @@ def _tokenize_singleton_prompt( prompt: SingletonDictPrompt, params: TokenizeParams, ) -> SingletonTokPrompt: + self._reject_kv_only_prompt(prompt) if "prompt_token_ids" not in prompt and "prompt_embeds" not in prompt: if not isinstance(prompt.get("prompt"), str): raise TypeError( @@ -757,6 +768,7 @@ async def _tokenize_singleton_prompt_async( prompt: SingletonDictPrompt, params: TokenizeParams, ) -> SingletonTokPrompt: + self._reject_kv_only_prompt(prompt) if "prompt_token_ids" not in prompt and "prompt_embeds" not in prompt: if not isinstance(prompt.get("prompt"), str): raise TypeError( diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py index fdd7e88b61a3..30eda8ef1dfb 100644 --- a/vllm/v1/core/sched/scheduler.py +++ b/vllm/v1/core/sched/scheduler.py @@ -259,6 +259,7 @@ def __init__( # update_from_output. self.grammar_compile_error_reqs: dict[str, RequestError] = {} self.encoder_cache_mismatch_reqs: set[str] = set() + self.missing_multimodal_kv_reqs: set[str] = set() # Encoder-related. # Calculate encoder cache size if applicable @@ -1901,6 +1902,11 @@ def _update_request_as_session( mm_feature.mm_position = replace( mm_feature.mm_position, offset=mm_feature.mm_position.offset + base ) + if mm_feature.requires_kv: + session.mm_required_kv_tokens = max( + session.mm_required_kv_tokens, + mm_feature.mm_position.offset + mm_feature.mm_position.length, + ) session.mm_features.extend(update.mm_features) session._all_token_ids.extend(update.prompt_token_ids or ()) @@ -2044,6 +2050,20 @@ def _try_schedule_encoder_inputs( """ if num_new_tokens == 0 or not request.has_encoder_inputs: return [], num_new_tokens, encoder_compute_budget, [], [] + if num_computed_tokens < request.mm_required_kv_tokens: + # The request carries no media for these spans, so they cannot be + # recomputed here. Fail the request instead of running the decoder + # over placeholder ids. After preemption or a failed KV load, + # num_computed_tokens is re-derived, so the check re-applies. + logger.error( + "Request %s cannot recompute multimodal input without media; " + "KV covers %d tokens, requires %d. Retry through prefill.", + request.request_id, + num_computed_tokens, + request.mm_required_kv_tokens, + ) + self.missing_multimodal_kv_reqs.add(request.request_id) + return [], 0, encoder_compute_budget, [], [] encoder_inputs_to_schedule: list[int] = [] mm_features = request.mm_features assert mm_features is not None @@ -2561,6 +2581,8 @@ def update_from_output( error_req_ids = set(request_errors) error_req_ids.update(self.encoder_cache_mismatch_reqs) self.encoder_cache_mismatch_reqs.clear() + error_req_ids.update(self.missing_multimodal_kv_reqs) + self.missing_multimodal_kv_reqs.clear() if failed_kv_load_req_ids and not self.recompute_kv_load_failures: error_req_ids.update(failed_kv_load_req_ids) if self.ec_connector is not None: diff --git a/vllm/v1/engine/input_processor.py b/vllm/v1/engine/input_processor.py index 2ddfe5457f93..c1b6757f0752 100644 --- a/vllm/v1/engine/input_processor.py +++ b/vllm/v1/engine/input_processor.py @@ -11,6 +11,7 @@ from vllm.exceptions import VLLMValidationError from vllm.inputs import ( EngineInput, + MultiModalInput, PromptType, SingletonInput, split_enc_dec_input, @@ -439,7 +440,7 @@ def process_inputs( current_platform.validate_request(engine_input, params) encoder_input, decoder_input = split_enc_dec_input(engine_input) - self._validate_model_inputs(encoder_input, decoder_input) + self._validate_model_inputs(encoder_input, decoder_input, params) # Mypy can be conservative for TypedDict unions; normalize access. if decoder_input["type"] == "embeds": @@ -517,6 +518,7 @@ def process_inputs( mm_features: list[MultiModalFeatureSpec] | None = None if decoder_input["type"] == "multimodal": + requires_kv = decoder_input.get("mm_requires_kv", False) decoder_mm_inputs = decoder_input["mm_kwargs"] decoder_mm_positions = decoder_input["mm_placeholders"] decoder_mm_hashes = decoder_input["mm_hashes"] @@ -548,6 +550,7 @@ def process_inputs( ), mm_position=decoder_mm_positions[modality][idx], mm_hash=base_mm_hash, + requires_kv=requires_kv, ) ) @@ -621,6 +624,7 @@ def _validate_model_input( self, prompt_input: SingletonInput, prompt_type: Literal["encoder", "decoder"], + params: SamplingParams | PoolingParams, ) -> None: prompt_ids = ( None @@ -644,19 +648,25 @@ def _validate_model_input( ) if prompt_input["type"] == "multimodal": - decoder_mm_positions = prompt_input["mm_placeholders"] - for modality, mm_positions in decoder_mm_positions.items(): - for mm_position in mm_positions: - num_embeds = mm_position.get_num_embeds() - if num_embeds > self.mm_encoder_cache_size: - raise VLLMValidationError( - f"The {prompt_type} prompt contains a(n) {modality} item " - f"with {num_embeds} embedding tokens, which exceeds the " - f"pre-allocated encoder cache size " - f"{self.mm_encoder_cache_size}. Please reduce the input " - f"size or increase the encoder cache size " - f"by setting --limit-mm-per-prompt at startup." - ) + if prompt_input.get("mm_requires_kv", False): + # KV-only inputs carry no media, so encoder capacity does not + # apply; the scheduler enforces the actual KV coverage. + self._validate_kv_only_input(prompt_input, params) + else: + decoder_mm_positions = prompt_input["mm_placeholders"] + for modality, mm_positions in decoder_mm_positions.items(): + for mm_position in mm_positions: + num_embeds = mm_position.get_num_embeds() + if num_embeds > self.mm_encoder_cache_size: + raise VLLMValidationError( + f"The {prompt_type} prompt contains a(n) " + f"{modality} item with {num_embeds} embedding " + "tokens, which exceeds the pre-allocated encoder " + f"cache size {self.mm_encoder_cache_size}. Please " + "reduce the input size or increase the encoder " + "cache size by setting --limit-mm-per-prompt at " + "startup." + ) # Shared by generate, embedding and pooling requests. if prompt_ids: @@ -666,8 +676,72 @@ def _validate_model_inputs( self, encoder_input: SingletonInput | None, decoder_input: SingletonInput, + params: SamplingParams | PoolingParams, ): if encoder_input is not None: - self._validate_model_input(encoder_input, prompt_type="encoder") + self._validate_model_input( + encoder_input, prompt_type="encoder", params=params + ) + + self._validate_model_input(decoder_input, prompt_type="decoder", params=params) - self._validate_model_input(decoder_input, prompt_type="decoder") + def _validate_kv_only_input( + self, + decoder_input: MultiModalInput, + params: SamplingParams | PoolingParams, + ) -> None: + """Validate a media-free input whose features require decoder KV. + + The input carries hashes and placeholder positions but no media, so + nothing about the items can be derived here. Reject up front what the + scheduler could otherwise only fail after the KV has been transferred. + """ + if self.model_config.uses_mrope or self.model_config.is_encoder_decoder: + # M-RoPE positions and encoder-decoder cross-attention are derived + # from the media itself, which a KV-only input does not carry. + raise VLLMValidationError( + "mm_requires_kv is not supported for M-RoPE or encoder-decoder models." + ) + if ( + not isinstance(params, SamplingParams) + or params.prompt_logprobs is not None + or params.prompt_logprob_token_ids is not None + ): + raise VLLMValidationError( + "mm_requires_kv inputs cannot be rescored without media; pooling, " + "prompt_logprobs and prompt_logprob_token_ids are unsupported." + ) + num_prompt_tokens = len(decoder_input["prompt_token_ids"]) + mm_kwargs = decoder_input["mm_kwargs"] + last_span_end = 0 + for modality, hashes in decoder_input["mm_hashes"].items(): + positions = decoder_input["mm_placeholders"].get(modality, []) + items = mm_kwargs.get(modality) + if items is None: + items = [None] * len(hashes) + if not hashes or not (len(hashes) == len(positions) == len(items)): + raise VLLMValidationError( + "mm_requires_kv inputs need one hash and one placeholder per " + f"{modality} item." + ) + for mm_hash, position, item in zip(hashes, positions, items): + if not mm_hash or item is not None: + raise VLLMValidationError( + "mm_requires_kv inputs must carry a nonempty hash and no " + f"media payload per item; got {modality} hash={mm_hash!r} " + f"with payload={'present' if item is not None else 'absent'}." + ) + span_end = position.offset + position.length + if span_end > num_prompt_tokens: + raise VLLMValidationError( + f"{modality} placeholder [{position.offset}, {span_end}) " + f"exceeds the prompt length {num_prompt_tokens}." + ) + last_span_end = max(last_span_end, span_end) + if last_span_end >= num_prompt_tokens: + # The final prompt token is always recomputed to produce logits, so + # a span that reaches it could never be covered by KV alone. + raise VLLMValidationError( + "mm_requires_kv inputs need at least one prompt token after the " + "last multimodal span." + ) diff --git a/vllm/v1/request.py b/vllm/v1/request.py index 78bd38430e33..27c8cc657409 100644 --- a/vllm/v1/request.py +++ b/vllm/v1/request.py @@ -187,6 +187,16 @@ def __init__( # Multi-modal related self.mm_features = mm_features or [] + # Furthest prompt position that decoder KV must already cover because + # the media behind it is not available on this instance; 0 otherwise. + self.mm_required_kv_tokens = max( + ( + feature.mm_position.offset + feature.mm_position.length + for feature in self.mm_features + if feature.requires_kv + ), + default=0, + ) # Read-only views # Prevent directly appending to these lists since