diff --git a/docs/features/multimodal_inputs.md b/docs/features/multimodal_inputs.md index e4cd5f56a2fa..6fcf6accc50f 100644 --- a/docs/features/multimodal_inputs.md +++ b/docs/features/multimodal_inputs.md @@ -1416,3 +1416,22 @@ Just like with offline inference, you can skip sending media if you expect cache }, ``` + +#### Deferred Media Fetching + +The HF renderer first attempts to render a conversation without fetching media when +every media item has a UUID and at least one image, audio, or video item is supplied as +a URL or inline data. This is enabled by default. Set `skip_early_mm_lookup=True` to +opt out per request. + +- If every media item is in the processor cache, fetching (downloading and decoding) + is skipped. +- If any item misses, vLLM retries that conversation once, fetching the supplied media + normally. Other conversations in the batch are not retried. + +This optimization requires the processor cache and is disabled for unified vision +chunks and audio extracted from video. An item supplied without a URL must still be +cached for the request to succeed. + +The UUID is authoritative on a cache hit; vLLM does not verify that the URL +still identifies the cached media (see [Multimodal Media UUID Security](../usage/security.md#multimodal-media-uuid-security)). diff --git a/tests/entrypoints/openai/responses/test_protocol.py b/tests/entrypoints/openai/responses/test_protocol.py index 04b07992528b..16a45f6b923f 100644 --- a/tests/entrypoints/openai/responses/test_protocol.py +++ b/tests/entrypoints/openai/responses/test_protocol.py @@ -43,6 +43,15 @@ def test_serialize_messages() -> None: assert serialize_messages([msg, dict_value]) == [msg_value, dict_value] +def test_skip_early_mm_lookup_is_preserved_in_chat_params() -> None: + request = ResponsesRequest(input="hello", skip_early_mm_lookup=True) + + chat_params = request.build_chat_params(None, "auto") + + assert request.skip_early_mm_lookup is True + assert chat_params.skip_early_mm_lookup is True + + def test_custom_tool_grammar_format_rejected() -> None: grammar = {"type": "grammar", "syntax": "lark", "definition": 'start: "pwd"'} with pytest.raises(VLLMValidationError) as exc_info: diff --git a/tests/renderers/chat_utils/test_chat_message.py b/tests/renderers/chat_utils/test_chat_message.py new file mode 100644 index 000000000000..e150db17a998 --- /dev/null +++ b/tests/renderers/chat_utils/test_chat_message.py @@ -0,0 +1,71 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +import pytest + +from vllm.renderers.chat_utils import ( + can_conversation_defer_media_fetching, + has_non_text_content, +) + + +@pytest.mark.parametrize( + ("content", "expected"), + [ + pytest.param("Describe", False, id="text"), + pytest.param( + [{"type": "image_url", "image_url": {"url": "url"}}], True, id="image" + ), + ], +) +def test_has_non_text_content(content, expected): + assert has_non_text_content([{"role": "user", "content": content}]) is expected + + +@pytest.mark.parametrize( + ("content", "expected"), + [ + pytest.param( + [{"type": "image_url", "image_url": {"url": "url"}, "uuid": "image"}], + True, + id="image-url", + ), + pytest.param( + [ + {"type": "image_url", "image_url": {"url": "url"}, "uuid": "image"}, + {"type": "video_url", "video_url": {"url": "url"}}, + ], + False, + id="missing-uuid", + ), + pytest.param( + [{"type": "image_url", "image_url": None, "uuid": "image"}], + False, + id="uuid-without-data", + ), + pytest.param( + [ + { + "type": "input_audio", + "input_audio": {"data": "", "format": "wav"}, + "uuid": "audio", + } + ], + False, + id="input-audio-without-data", + ), + ], +) +def test_can_conversation_defer_media_fetching(content, expected): + """Fetching is deferred only if all media has UUIDs and some has data.""" + messages = [{"role": "user", "content": content}] + + assert can_conversation_defer_media_fetching(messages) is expected + + +def test_can_conversation_defer_media_fetching_does_not_consume_iterators(): + part = {"type": "image_url", "image_url": {"url": "url"}, "uuid": "image"} + content = iter([part]) + + messages = [{"role": "user", "content": content}] + assert not can_conversation_defer_media_fetching(messages) + assert next(content) == part diff --git a/tests/renderers/test_process_multi_modal_uuids.py b/tests/renderers/test_process_multi_modal_uuids.py index bb9a533f0a77..f564a33113c1 100644 --- a/tests/renderers/test_process_multi_modal_uuids.py +++ b/tests/renderers/test_process_multi_modal_uuids.py @@ -1,17 +1,24 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project +import asyncio +from copy import deepcopy from http import HTTPStatus +from unittest.mock import AsyncMock, Mock import pytest +from vllm.assets.audio import AudioAsset from vllm.assets.image import ImageAsset from vllm.assets.video import VideoAsset from vllm.config import CacheConfig, ModelConfig, VllmConfig from vllm.entrypoints.serve import create_error_response from vllm.exceptions import VLLMValidationError +from vllm.multimodal.media import MediaConnector from vllm.multimodal.parse import parse_mm_uuids +from vllm.multimodal.processing.processor import MultiModalProcessorCacheMissError from vllm.renderers.hf import HfRenderer +from vllm.renderers.params import ChatParams from vllm.tokenizers.registry import cached_tokenizer_from_config cherry_pil_image = ImageAsset("cherry_blossom").pil_image @@ -21,13 +28,20 @@ baby_reading_video_asset.np_ndarrays, baby_reading_video_asset.metadata, ) +mary_had_lamb_audio = AudioAsset("mary_had_lamb").audio_and_sample_rate + +QWEN25_VL_MODEL_ID = "Qwen/Qwen2.5-VL-3B-Instruct" +QWEN25_OMNI_MODEL_ID = "Qwen/Qwen2.5-Omni-7B" def _build_renderer( - *, mm_cache_gb: float = 4.0, enable_prefix_caching: bool = True + *, + model: str = QWEN25_VL_MODEL_ID, + mm_cache_gb: float = 4.0, + enable_prefix_caching: bool = True, ) -> HfRenderer: model_config = ModelConfig( - model="Qwen/Qwen2.5-VL-3B-Instruct", + model=model, max_model_len=128, mm_processor_cache_gb=mm_cache_gb, ) @@ -69,6 +83,197 @@ def test_text_only_model_mm_data_maps_to_bad_request(): assert error_response.error.code == HTTPStatus.BAD_REQUEST +def _render_chat(renderer, conversations, params, use_async, **kwargs): + if use_async: + return asyncio.run(renderer.render_chat_async(conversations, params, **kwargs)) + return renderer.render_chat(conversations, params, **kwargs) + + +def _mock_media(monkeypatch, modality, media, use_async): + fetch = AsyncMock(return_value=media) if use_async else Mock(return_value=media) + method = f"fetch_{modality}" + ("_async" if use_async else "") + monkeypatch.setattr(MediaConnector, method, fetch) + return fetch + + +def _media_part(part_type, uuid): + if part_type == "input_audio": + media = {"data": "AAAA", "format": "wav"} + else: + media = {"url": f"https://example.com/{uuid}.{part_type.removesuffix('_url')}"} + return {"type": part_type, part_type: media, "uuid": uuid} + + +def _media_messages(name, *items): + content = [{"type": "text", "text": name}] + content += [_media_part(part_type, uuid) for part_type, uuid in items] + return [{"role": "user", "content": content}] + + +@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize( + ("part_type", "modality", "media", "model"), + [ + pytest.param( + "image_url", "image", cherry_pil_image, QWEN25_VL_MODEL_ID, id="image" + ), + pytest.param( + "video_url", "video", baby_reading_video, QWEN25_VL_MODEL_ID, id="video" + ), + pytest.param( + "audio_url", "audio", mary_had_lamb_audio, QWEN25_OMNI_MODEL_ID, id="audio" + ), + pytest.param( + "input_audio", + "audio", + mary_had_lamb_audio, + QWEN25_OMNI_MODEL_ID, + id="input-audio", + ), + ], +) +def test_cached_uuid_skips_media_fetching( + monkeypatch: pytest.MonkeyPatch, + part_type: str, + modality: str, + media: object, + model: str, + use_async: bool, +): + renderer = _build_renderer(model=model) + fetch_media = _mock_media(monkeypatch, modality, media, use_async) + messages = _media_messages("Describe", (part_type, f"test-{modality}-uuid")) + + _, first_prompts = _render_chat(renderer, [messages], ChatParams(), use_async) + _, second_prompts = _render_chat(renderer, [messages], ChatParams(), use_async) + + first_input = first_prompts[0] + second_input = second_prompts[0] + assert first_input["mm_hashes"] == second_input["mm_hashes"] + assert first_input["prompt_token_ids"] == second_input["prompt_token_ids"] + assert first_input["mm_placeholders"] == second_input["mm_placeholders"] + assert fetch_media.call_count == 1 + + +@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"]) +def test_partial_uuid_hit_retries_only_affected_conversation(monkeypatch, use_async): + """A miss refetches all media in its conversation without rerendering a peer.""" + renderer = _build_renderer() + fetch_image = _mock_media(monkeypatch, "image", cherry_pil_image, use_async) + fetch_video = _mock_media(monkeypatch, "video", baby_reading_video, use_async) + cached = _media_messages("Cached conversation", ("image_url", "cached-image")) + partial = _media_messages("Partial conversation", ("image_url", "partial-image")) + params = ChatParams() + _, warm_prompts = _render_chat(renderer, [cached, partial], params, use_async) + partial[0]["content"].append(_media_part("video_url", "missing-video")) + conversations = [cached, partial] + original = deepcopy(conversations) + fetch_image.reset_mock() + fetch_video.reset_mock() + method = "render_messages_async" if use_async else "render_messages" + mock_type = AsyncMock if use_async else Mock + render_messages = mock_type(wraps=getattr(renderer, method)) + monkeypatch.setattr(renderer, method, render_messages) + + _, prompts = _render_chat(renderer, conversations, params, use_async) + + assert [ + call.args[0][0]["content"][0]["text"] for call in render_messages.call_args_list + ] == ["Partial conversation"] + fetch_image.assert_called_once_with("https://example.com/partial-image.image") + assert fetch_video.call_count == 1 + assert prompts[0]["prompt_token_ids"] == warm_prompts[0]["prompt_token_ids"] + assert prompts[0]["mm_hashes"] == warm_prompts[0]["mm_hashes"] + assert set(prompts[1]["mm_hashes"]) == {"image", "video"} + assert conversations == original + + +def _wrap_process_for_engine(monkeypatch, renderer, use_async): + method = "process_for_engine_async" if use_async else "process_for_engine" + mock_type = AsyncMock if use_async else Mock + process = mock_type(wraps=getattr(renderer, method)) + monkeypatch.setattr(renderer, method, process) + return process + + +@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize( + "case", ["skip-cache", "no-cache", "opt-out", "audio-in-video"] +) +def test_ineligible_request_fetches_media_upfront(monkeypatch, use_async, case): + """Requests that cannot defer media fetching are processed once, not retried.""" + renderer = _build_renderer(mm_cache_gb=0 if case == "no-cache" else 4) + fetch_image = _mock_media(monkeypatch, "image", cherry_pil_image, use_async) + messages = _media_messages("Describe", ("image_url", "image")) + params = ChatParams() + if case == "opt-out": + params = ChatParams(skip_early_mm_lookup=True) + elif case == "audio-in-video": + params = ChatParams(mm_processor_kwargs={"use_audio_in_video": True}) + process = _wrap_process_for_engine(monkeypatch, renderer, use_async) + + _render_chat( + renderer, + [messages], + params, + use_async, + skip_mm_cache=case == "skip-cache", + ) + + assert process.call_count == 1 + assert fetch_image.call_count == 1 + + +@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"]) +def test_uuid_only_cache_miss_is_not_retried(monkeypatch, use_async): + """Without a URL to fall back to, a cache miss fails without a retry.""" + renderer = _build_renderer() + messages = _media_messages("Describe", ("image_url", "image")) + messages[0]["content"][1]["image_url"] = None + process = _wrap_process_for_engine(monkeypatch, renderer, use_async) + + with pytest.raises(MultiModalProcessorCacheMissError): + _render_chat(renderer, [messages], ChatParams(), use_async) + + assert process.call_count == 1 + + +def test_batch_cache_misses_refetch_media_concurrently(monkeypatch): + """Retried conversations fetch media concurrently and keep the batch order.""" + renderer = _build_renderer() + active_fetches = max_active_fetches = 0 + + async def fetch_image(*args, **kwargs): + nonlocal active_fetches, max_active_fetches + active_fetches += 1 + max_active_fetches = max(max_active_fetches, active_fetches) + try: + await asyncio.sleep(0.01) + return cherry_pil_image + finally: + active_fetches -= 1 + + monkeypatch.setattr(MediaConnector, "fetch_image_async", fetch_image) + conversations = [ + _media_messages(f"Conversation {i}", ("image_url", f"image-{i}")) + for i in range(2) + ] + + out_conversations, prompts = _render_chat( + renderer, conversations, ChatParams(), use_async=True + ) + _, cached_prompts = _render_chat( + renderer, conversations, ChatParams(), use_async=True + ) + + assert max_active_fetches == 2 + assert all( + f"Conversation {i}" in str(conversation) + for i, conversation in enumerate(out_conversations) + ) + assert [p["mm_hashes"] for p in prompts] == [p["mm_hashes"] for p in cached_prompts] + + def test_multi_modal_uuids_length_mismatch_raises(): renderer = _build_renderer() diff --git a/vllm/entrypoints/openai/chat_completion/protocol.py b/vllm/entrypoints/openai/chat_completion/protocol.py index 1824af05ffda..6f151cfbce66 100644 --- a/vllm/entrypoints/openai/chat_completion/protocol.py +++ b/vllm/entrypoints/openai/chat_completion/protocol.py @@ -386,6 +386,10 @@ class ChatCompletionRequest(OpenAIBaseModel): default=None, description=("Additional kwargs to pass to the HF processor."), ) + skip_early_mm_lookup: bool = Field( + default=False, + description="Skip the UUID-only rendering attempt for this request.", + ) structured_outputs: StructuredOutputsParams | None = Field( default=None, description="Additional kwargs for structured outputs", @@ -660,6 +664,7 @@ def build_chat_params( extra_kwargs, ), media_io_kwargs=self.media_io_kwargs, + skip_early_mm_lookup=self.skip_early_mm_lookup, # No-tools requests default to tool_choice="none" at the API # layer. Collapse that default before rendering, so K3 emits a # model-visible tool-choice instruction only for requests with a @@ -1188,6 +1193,7 @@ class BatchChatCompletionRequest(OpenAIBaseModel): chat_template_kwargs: dict[str, Any] | None = None media_io_kwargs: dict[str, dict[str, Any]] | None = None mm_processor_kwargs: dict[str, Any] | None = None + skip_early_mm_lookup: bool = False priority: int = Field(default=0, ge=_INT64_MIN, le=_INT64_MAX) cache_salt: str | None = None include_stop_str_in_output: bool = False diff --git a/vllm/entrypoints/openai/responses/protocol.py b/vllm/entrypoints/openai/responses/protocol.py index 00fe1ccdabce..e2613af24910 100644 --- a/vllm/entrypoints/openai/responses/protocol.py +++ b/vllm/entrypoints/openai/responses/protocol.py @@ -376,6 +376,10 @@ class ResponsesRequest(OpenAIBaseModel): default=None, description=("Additional kwargs to pass to the HF processor."), ) + skip_early_mm_lookup: bool = Field( + default=False, + description="Skip the UUID-only rendering attempt for this request.", + ) priority: int = Field( default=0, ge=_INT64_MIN, @@ -511,6 +515,7 @@ def build_chat_params( extra_kwargs, ), media_io_kwargs=self.media_io_kwargs, + skip_early_mm_lookup=self.skip_early_mm_lookup, tool_choice=self.tool_choice if self.tools else None, ) diff --git a/vllm/multimodal/processing/processor.py b/vllm/multimodal/processing/processor.py index 5b46267dff54..140042616b9f 100644 --- a/vllm/multimodal/processing/processor.py +++ b/vllm/multimodal/processing/processor.py @@ -987,6 +987,10 @@ def find_mm_placeholders( return dict(full_groupby_modality(it)) +class MultiModalProcessorCacheMissError(ValueError): + """A processor cache miss for an item whose media data was omitted.""" + + MultiModalIsCached = dict[str, list[bool]] """ A collection of the `is_cached` flag for each item, with a similar structure as @@ -1370,7 +1374,7 @@ def _get_cache_missing_items( for idx in idxs: data = mm_data_items[modality][idx] if data is None: - raise ValueError( + raise MultiModalProcessorCacheMissError( f"Cache miss for {modality} at index {idx} " f"but data is not provided." ) diff --git a/vllm/renderers/base.py b/vllm/renderers/base.py index 395b1ead7acf..5c4f7073030e 100644 --- a/vllm/renderers/base.py +++ b/vllm/renderers/base.py @@ -10,7 +10,7 @@ from dataclasses import replace from functools import cached_property from threading import Lock -from typing import TYPE_CHECKING, Any, Generic, overload +from typing import TYPE_CHECKING, Any, Generic, cast, overload from typing_extensions import TypeVar @@ -46,6 +46,7 @@ ) from vllm.multimodal.processing import BaseMultiModalProcessor from vllm.multimodal.processing import ProcessorInputs as MMProcessorInputs +from vllm.multimodal.processing.processor import MultiModalProcessorCacheMissError from vllm.multimodal.registry import MultiModalTimingRegistry from vllm.tokenizers import TokenizerLike from vllm.utils.async_utils import make_async @@ -53,6 +54,7 @@ from vllm.utils.torch_utils import set_default_torch_num_threads from vllm.v1.metrics.stats import MultiModalCacheStats +from .chat_utils import can_conversation_defer_media_fetching, uses_audio_in_video from .embed_utils import safe_load_prompt_embeds from .inputs import ( DictPrompt, @@ -1339,6 +1341,118 @@ async def render_cmpl_async( ) ) + def _can_defer_uuid_media_fetching(self) -> bool: + """Whether this renderer can render messages without fetching UUID media.""" + return False + + def _render_messages_deferring_uuid_media_fetching( + self, + messages: list["ChatCompletionMessageParam"], + params: ChatParams, + ) -> tuple[list["ConversationMessage"], DictPrompt]: + """Like `render_messages`, but without fetching media that has a UUID.""" + raise NotImplementedError + + async def _render_messages_deferring_uuid_media_fetching_async( + self, + messages: list["ChatCompletionMessageParam"], + params: ChatParams, + ) -> tuple[list["ConversationMessage"], DictPrompt]: + return self._render_messages_deferring_uuid_media_fetching(messages, params) + + def _should_defer_uuid_media_fetching( + self, chat_params: ChatParams, skip_mm_cache: bool + ) -> bool: + return ( + self._can_defer_uuid_media_fetching() + and self._mm_processor_cache is not None + and not skip_mm_cache + and not chat_params.skip_early_mm_lookup + and not uses_audio_in_video(chat_params.mm_processor_kwargs) + ) + + def _render_chat_prompts( + self, + conversations: Sequence[list["ChatCompletionMessageParam"]], + chat_params: ChatParams, + tok_params: TokenizeParams, + prompt_extras: dict[str, Any], + *, + defer_uuid_media_fetching: bool, + ) -> tuple[list[list["ConversationMessage"]], list[TokPrompt], list[bool]]: + """Render and tokenize `conversations`. + + Returns: + The rendered conversations, their prompts, and whether each prompt + had its media fetching deferred. + + """ + + def render(conversation: list["ChatCompletionMessageParam"]): + is_media_fetching_deferred = ( + defer_uuid_media_fetching + and can_conversation_defer_media_fetching(conversation) + ) + render_messages = ( + self._render_messages_deferring_uuid_media_fetching + if is_media_fetching_deferred + else self.render_messages + ) + return ( + *render_messages(conversation, chat_params), + is_media_fetching_deferred, + ) + + rendered = [render(conversation) for conversation in conversations] + + tok_prompts = self.tokenize_prompts( + [prompt for _, prompt, _ in rendered], tok_params + ) + self._apply_prompt_extras(tok_prompts, prompt_extras) + + return ( + [conv for conv, _, _ in rendered], + tok_prompts, + [is_deferred for _, _, is_deferred in rendered], + ) + + async def _render_chat_prompts_async( + self, + conversations: Sequence[list["ChatCompletionMessageParam"]], + chat_params: ChatParams, + tok_params: TokenizeParams, + prompt_extras: dict[str, Any], + *, + defer_uuid_media_fetching: bool, + ) -> tuple[list[list["ConversationMessage"]], list[TokPrompt], list[bool]]: + async def render(conversation: list["ChatCompletionMessageParam"]): + is_media_fetching_deferred = ( + defer_uuid_media_fetching + and can_conversation_defer_media_fetching(conversation) + ) + render_messages = ( + self._render_messages_deferring_uuid_media_fetching_async + if is_media_fetching_deferred + else self.render_messages_async + ) + return ( + *await render_messages(conversation, chat_params), + is_media_fetching_deferred, + ) + + rendered = await asyncio.gather(*(render(c) for c in conversations)) + + tok_prompts = await self.tokenize_prompts_async( + [prompt for _, prompt, _ in rendered], tok_params + ) + self._apply_prompt_extras(tok_prompts, prompt_extras) + + return ( + [conv for conv, _, _ in rendered], + tok_prompts, + [is_deferred for _, _, is_deferred in rendered], + ) + def render_chat( self, conversations: Sequence[list["ChatCompletionMessageParam"]], @@ -1353,31 +1467,51 @@ def render_chat( if tok_params is None: tok_params = self.default_chat_tok_params - rendered = [ - self.render_messages(conversation, chat_params) - for conversation in conversations - ] - - out_conversations = list[list["ConversationMessage"]]() - dict_prompts = list[DictPrompt]() - for conv, prompt in rendered: - out_conversations.append(conv) - dict_prompts.append(prompt) - - tok_prompts = self.tokenize_prompts(dict_prompts, tok_params) - prompt_extras = dict(prompt_extras or {}) prompt_extras["media_io_kwargs"] = chat_params.media_io_kwargs or {} - self._apply_prompt_extras(tok_prompts, prompt_extras) + + rendered = self._render_chat_prompts( + conversations, + chat_params, + tok_params, + prompt_extras, + defer_uuid_media_fetching=self._should_defer_uuid_media_fetching( + chat_params, skip_mm_cache + ), + ) + out_conversations, tok_prompts, media_fetching_deferred = rendered + + def process(prompt: TokPrompt, is_media_fetching_deferred: bool = False): + try: + return self.process_for_engine( + prompt, + arrival_time, + skip_mm_cache=skip_mm_cache, + tok_params=tok_params, + ) + except MultiModalProcessorCacheMissError: + if not is_media_fetching_deferred: + raise + return None eng_prompts = [ - self.process_for_engine( - prompt, arrival_time, skip_mm_cache=skip_mm_cache, tok_params=tok_params - ) - for prompt in tok_prompts + process(p, d) for p, d in zip(tok_prompts, media_fetching_deferred) ] - return out_conversations, eng_prompts + # Retry the deferred prompts that missed the cache with their media fetched. + if missed := [i for i, prompt in enumerate(eng_prompts) if prompt is None]: + retry_conversations, retry_prompts, _ = self._render_chat_prompts( + [conversations[i] for i in missed], + chat_params, + tok_params, + prompt_extras, + defer_uuid_media_fetching=False, + ) + for i, conv, prompt in zip(missed, retry_conversations, retry_prompts): + out_conversations[i] = conv + eng_prompts[i] = process(prompt) + + return out_conversations, cast(list[EngineInput], eng_prompts) async def render_chat_async( self, @@ -1393,30 +1527,54 @@ async def render_chat_async( if tok_params is None: tok_params = self.default_chat_tok_params - rendered = [ - self.render_messages_async(conversation, chat_params) - for conversation in conversations - ] - - out_conversations = list[list["ConversationMessage"]]() - dict_prompts = list[DictPrompt]() - for conv, prompt in await asyncio.gather(*rendered): - out_conversations.append(conv) - dict_prompts.append(prompt) - - tok_prompts = await self.tokenize_prompts_async(dict_prompts, tok_params) - prompt_extras = dict(prompt_extras or {}) prompt_extras["media_io_kwargs"] = chat_params.media_io_kwargs or {} - self._apply_prompt_extras(tok_prompts, prompt_extras) - eng_prompts = await asyncio.gather( - *( - self.process_for_engine_async( - p, arrival_time, skip_mm_cache=skip_mm_cache, tok_params=tok_params + rendered = await self._render_chat_prompts_async( + conversations, + chat_params, + tok_params, + prompt_extras, + defer_uuid_media_fetching=self._should_defer_uuid_media_fetching( + chat_params, skip_mm_cache + ), + ) + out_conversations, tok_prompts, media_fetching_deferred = rendered + + async def process(prompt: TokPrompt, is_media_fetching_deferred: bool = False): + try: + return await self.process_for_engine_async( + prompt, + arrival_time, + skip_mm_cache=skip_mm_cache, + tok_params=tok_params, ) - for p in tok_prompts - ) + except MultiModalProcessorCacheMissError: + if not is_media_fetching_deferred: + raise + return None + + eng_prompts = await asyncio.gather( + *(process(p, d) for p, d in zip(tok_prompts, media_fetching_deferred)) ) - return out_conversations, eng_prompts + # Retry the deferred prompts that missed the cache with their media fetched. + if missed := [i for i, prompt in enumerate(eng_prompts) if prompt is None]: + retried = await self._render_chat_prompts_async( + [conversations[i] for i in missed], + chat_params, + tok_params, + prompt_extras, + defer_uuid_media_fetching=False, + ) + retry_conversations, retry_prompts, _ = retried + retry_eng_prompts = await asyncio.gather( + *(process(prompt) for prompt in retry_prompts) + ) + for i, conv, eng_prompt in zip( + missed, retry_conversations, retry_eng_prompts + ): + out_conversations[i] = conv + eng_prompts[i] = eng_prompt + + return out_conversations, cast(list[EngineInput], eng_prompts) diff --git a/vllm/renderers/chat_utils/__init__.py b/vllm/renderers/chat_utils/__init__.py index dd68b7c620c3..c62081cfb04f 100644 --- a/vllm/renderers/chat_utils/__init__.py +++ b/vllm/renderers/chat_utils/__init__.py @@ -2,6 +2,7 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project from .chat_message import ( + can_conversation_defer_media_fetching, has_non_text_content, parse_chat_message_content, parse_chat_message_content_parts, @@ -13,7 +14,7 @@ load_chat_template, validate_chat_template, ) -from .content_parser import PROMPT_EMBEDS_PLACEHOLDER_TOKEN +from .content_parser import PROMPT_EMBEDS_PLACEHOLDER_TOKEN, uses_audio_in_video from .item_tracker import ( AsyncMultiModalItemTracker, BaseMultiModalItemTracker, @@ -33,10 +34,12 @@ "BaseMultiModalItemTracker", "MultiModalItemTracker", "AsyncMultiModalItemTracker", + "can_conversation_defer_media_fetching", "has_non_text_content", "parse_chat_message_content", "parse_chat_message_content_parts", "parse_chat_messages", "parse_chat_messages_async", "PROMPT_EMBEDS_PLACEHOLDER_TOKEN", + "uses_audio_in_video", ] diff --git a/vllm/renderers/chat_utils/chat_message.py b/vllm/renderers/chat_utils/chat_message.py index aad31fc2c6ee..e37021e0bed9 100644 --- a/vllm/renderers/chat_utils/chat_message.py +++ b/vllm/renderers/chat_utils/chat_message.py @@ -346,6 +346,18 @@ def parse_chat_message_content_mm_part( _TEXT_CONTENT_PART_TYPES = TEXT_PART_TYPES | {"tool_reference"} # Keys that mark a content part as multimodal, whatever its ``type``. _MEDIA_CONTENT_PART_KEYS = frozenset(MM_PARSER_MAP) - _TEXT_CONTENT_PART_TYPES +# Content part types whose media is fetched from a URL or inline data. +_FETCHED_MEDIA_PART_TYPES = frozenset( + {"image_url", "audio_url", "input_audio", "video_url"} +) + + +def _is_non_text_part(part: object) -> bool: + return isinstance(part, dict) and ( + any(key in part for key in _MEDIA_CONTENT_PART_KEYS) + or not isinstance(part_type := part.get("type", "text"), str) + or part_type not in _TEXT_CONTENT_PART_TYPES + ) def has_non_text_content(messages: Any) -> bool: @@ -368,16 +380,39 @@ def has_non_text_content(messages: Any) -> bool: content = msg.get("content") if isinstance(msg, dict) else None if not isinstance(content, list): continue - for part in content: - if isinstance(part, dict) and ( - any(key in part for key in _MEDIA_CONTENT_PART_KEYS) - or not isinstance(part_type := part.get("type", "text"), str) - or part_type not in _TEXT_CONTENT_PART_TYPES - ): - return True + if any(_is_non_text_part(part) for part in content): + return True return False +def can_conversation_defer_media_fetching( + messages: list[ChatCompletionMessageParam], +) -> bool: + """Whether every media part in ``messages`` has a UUID and at least one + has data to fetch. + + Only list content is inspected, so content that is a one-shot iterator is + never consumed; messages with such content cannot defer. + """ + has_data = False + for message in messages: + content = message.get("content") + if content is None or isinstance(content, str): + continue + if not isinstance(content, list): + return False + for part in content: + if not _is_non_text_part(part): + continue + if part.get("uuid") is None: + return False + part_type, data = parse_chat_message_content_mm_part(part) + if part_type in _FETCHED_MEDIA_PART_TYPES: + # `input_audio` carries its data inside a dict. + has_data |= bool(data.get("data") if isinstance(data, dict) else data) + return has_data + + def parse_chat_message_content_parts( role: str, parts: Iterable[ChatCompletionContentPartParam], @@ -777,6 +812,7 @@ def parse_chat_messages( content_format: ChatTemplateContentFormat, media_io_kwargs: dict[str, dict[str, Any]] | None = None, mm_processor_kwargs: dict[str, Any] | None = None, + defer_uuid_media_fetching: bool = False, ) -> tuple[ list[ConversationMessage], MultiModalDataDict | None, @@ -786,6 +822,7 @@ def parse_chat_messages( mm_tracker = MultiModalItemTracker( model_config, media_io_kwargs=media_io_kwargs, + defer_uuid_media_fetching=defer_uuid_media_fetching, ) for msg in messages: @@ -816,6 +853,7 @@ async def parse_chat_messages_async( content_format: ChatTemplateContentFormat, media_io_kwargs: dict[str, dict[str, Any]] | None = None, mm_processor_kwargs: dict[str, Any] | None = None, + defer_uuid_media_fetching: bool = False, ) -> tuple[ list[ConversationMessage], MultiModalDataDict | None, @@ -825,6 +863,7 @@ async def parse_chat_messages_async( mm_tracker = AsyncMultiModalItemTracker( model_config, media_io_kwargs=media_io_kwargs, + defer_uuid_media_fetching=defer_uuid_media_fetching, ) for msg in messages: diff --git a/vllm/renderers/chat_utils/content_parser.py b/vllm/renderers/chat_utils/content_parser.py index b84efa03706b..4df4916e5ef8 100644 --- a/vllm/renderers/chat_utils/content_parser.py +++ b/vllm/renderers/chat_utils/content_parser.py @@ -54,6 +54,13 @@ def make_prompt_embeds_placeholder(index: int) -> str: ) +def uses_audio_in_video(mm_processor_kwargs: dict[str, Any] | None) -> bool: + """Whether audio is extracted from video content parts.""" + return bool( + mm_processor_kwargs and mm_processor_kwargs.get("use_audio_in_video", False) + ) + + class BaseMultiModalContentParser(ABC): def __init__(self) -> None: super().__init__() @@ -158,6 +165,10 @@ def _connector(self) -> MediaConnector: download_budget=self._tracker.download_budget, ) + @cached_property + def _use_audio_in_video(self) -> bool: + return uses_audio_in_video(self._mm_processor_kwargs) + @property def model_config(self) -> ModelConfig: return self._tracker.model_config @@ -180,6 +191,7 @@ def parse_prompt_embeds(self, data: str) -> None: self._add_placeholder("prompt_embeds", PROMPT_EMBEDS_PLACEHOLDER_TOKEN) def parse_image(self, image_url: str | None, uuid: str | None = None) -> None: + image_url = self._tracker.url_to_fetch(image_url, uuid) image = self._connector.fetch_image(image_url) if image_url else None placeholder = self._tracker.add("image", (image, uuid)) @@ -250,6 +262,7 @@ def parse_image_pil( self._add_placeholder("image", placeholder) def parse_audio(self, audio_url: str | None, uuid: str | None = None) -> None: + audio_url = self._tracker.url_to_fetch(audio_url, uuid) audio = self._connector.fetch_audio(audio_url) if audio_url else None placeholder = self._tracker.add("audio", (audio, uuid)) @@ -272,6 +285,7 @@ def parse_input_audio( return self.parse_audio(audio_url, uuid) def parse_video(self, video_url: str | None, uuid: str | None = None) -> None: + video_url = self._tracker.url_to_fetch(video_url, uuid) video = ( self._connector.fetch_video( video_url=video_url, @@ -285,11 +299,7 @@ def parse_video(self, video_url: str | None, uuid: str | None = None) -> None: self._add_placeholder("video", placeholder) # Extract audio from video if use_audio_in_video is True - if ( - video_url - and self._mm_processor_kwargs - and self._mm_processor_kwargs.get("use_audio_in_video", False) - ): + if video_url and self._use_audio_in_video: audio = self._connector.fetch_audio(video_url) if video_url else None audio_placeholder = self._tracker.add("audio", (audio, uuid)) self._add_placeholder("audio", audio_placeholder) @@ -348,6 +358,10 @@ def _connector(self) -> MediaConnector: download_budget=self._tracker.download_budget, ) + @cached_property + def _use_audio_in_video(self) -> bool: + return uses_audio_in_video(self._mm_processor_kwargs) + @property def model_config(self) -> ModelConfig: return self._tracker.model_config @@ -388,6 +402,7 @@ async def _image_with_uuid_async(self, image_url: str | None, uuid: str | None): return image, uuid def parse_image(self, image_url: str | None, uuid: str | None = None) -> None: + image_url = self._tracker.url_to_fetch(image_url, uuid) placeholder = self._tracker.add( "image", partial(self._image_with_uuid_async, image_url, uuid) ) @@ -480,6 +495,7 @@ async def _audio_with_uuid_async(self, audio_url: str | None, uuid: str | None): return audio, uuid def parse_audio(self, audio_url: str | None, uuid: str | None = None) -> None: + audio_url = self._tracker.url_to_fetch(audio_url, uuid) placeholder = self._tracker.add( "audio", partial(self._audio_with_uuid_async, audio_url, uuid) ) @@ -513,17 +529,14 @@ async def _video_with_uuid_async(self, video_url: str | None, uuid: str | None): return video, uuid def parse_video(self, video_url: str | None, uuid: str | None = None) -> None: + video_url = self._tracker.url_to_fetch(video_url, uuid) placeholder = self._tracker.add( "video", partial(self._video_with_uuid_async, video_url, uuid) ) self._add_placeholder("video", placeholder) # Extract audio from video if use_audio_in_video is True - if ( - video_url - and self._mm_processor_kwargs - and self._mm_processor_kwargs.get("use_audio_in_video", False) - ): + if video_url and self._use_audio_in_video: audio_placeholder = self._tracker.add( "audio", partial(self._audio_with_uuid_async, video_url, uuid) ) diff --git a/vllm/renderers/chat_utils/item_tracker.py b/vllm/renderers/chat_utils/item_tracker.py index c2872e6852b3..d7f4a16458e0 100644 --- a/vllm/renderers/chat_utils/item_tracker.py +++ b/vllm/renderers/chat_utils/item_tracker.py @@ -207,11 +207,13 @@ def __init__( self, model_config: ModelConfig, media_io_kwargs: dict[str, dict[str, Any]] | None = None, + defer_uuid_media_fetching: bool = False, ): super().__init__() self._model_config = model_config self._media_io_kwargs = media_io_kwargs + self._defer_uuid_media_fetching = defer_uuid_media_fetching self._items_by_modality = defaultdict[str, list[_T]](list) # Track original modality for each vision_chunk item (image or video) @@ -266,6 +268,10 @@ def mm_processor(self): def video_processor_name(self) -> str | None: return get_video_processor_cls_name(self.model_config) + def url_to_fetch(self, url: str | None, uuid: str | None) -> str | None: + """Return `url`, or `None` if fetching media with a UUID is deferred.""" + return None if self._defer_uuid_media_fetching and uuid is not None else url + def add(self, modality: ModalityStr, item: _T) -> str | None: """Add a multi-modal item to the current prompt and returns the placeholder string to use, if any. diff --git a/vllm/renderers/hf.py b/vllm/renderers/hf.py index fa8a1db71689..d0dae24a8e31 100644 --- a/vllm/renderers/hf.py +++ b/vllm/renderers/hf.py @@ -1101,10 +1101,31 @@ def _can_produce_offsets(self) -> bool: # Only fast tokenizers expose offset_mapping. return self.tokenizer is not None and self.tokenizer.is_fast + @override + def _can_defer_uuid_media_fetching(self) -> bool: + return not self.use_unified_vision_chunk + def render_messages( self, messages: list[ChatCompletionMessageParam], params: ChatParams, + ) -> tuple[list[ConversationMessage], DictPrompt]: + return self._render_messages(messages, params, defer_uuid_media_fetching=False) + + @override + def _render_messages_deferring_uuid_media_fetching( + self, + messages: list[ChatCompletionMessageParam], + params: ChatParams, + ) -> tuple[list[ConversationMessage], DictPrompt]: + return self._render_messages(messages, params, defer_uuid_media_fetching=True) + + def _render_messages( + self, + messages: list[ChatCompletionMessageParam], + params: ChatParams, + *, + defer_uuid_media_fetching: bool, ) -> tuple[list[ConversationMessage], DictPrompt]: model_config = self.model_config tokenizer = self.get_tokenizer() @@ -1127,6 +1148,7 @@ def render_messages( ), media_io_kwargs=params.media_io_kwargs, mm_processor_kwargs=params.mm_processor_kwargs, + defer_uuid_media_fetching=defer_uuid_media_fetching, ) # prompt_embeds tensors are carried by the tracker through mm_data, @@ -1228,6 +1250,27 @@ async def render_messages_async( self, messages: list[ChatCompletionMessageParam], params: ChatParams, + ) -> tuple[list[ConversationMessage], DictPrompt]: + return await self._render_messages_async( + messages, params, defer_uuid_media_fetching=False + ) + + @override + async def _render_messages_deferring_uuid_media_fetching_async( + self, + messages: list[ChatCompletionMessageParam], + params: ChatParams, + ) -> tuple[list[ConversationMessage], DictPrompt]: + return await self._render_messages_async( + messages, params, defer_uuid_media_fetching=True + ) + + async def _render_messages_async( + self, + messages: list[ChatCompletionMessageParam], + params: ChatParams, + *, + defer_uuid_media_fetching: bool, ) -> tuple[list[ConversationMessage], DictPrompt]: model_config = self.model_config tokenizer = self.get_tokenizer() @@ -1250,6 +1293,7 @@ async def render_messages_async( ), media_io_kwargs=params.media_io_kwargs, mm_processor_kwargs=params.mm_processor_kwargs, + defer_uuid_media_fetching=defer_uuid_media_fetching, ) prompt_embeds_tensors: list[torch.Tensor] | None = None diff --git a/vllm/renderers/params.py b/vllm/renderers/params.py index cea7eb3d03b1..46abc758a33f 100644 --- a/vllm/renderers/params.py +++ b/vllm/renderers/params.py @@ -95,6 +95,9 @@ class ChatParams: mm_processor_kwargs: dict[str, Any] | None = None """The kwargs to pass to the multi-modal processor.""" + skip_early_mm_lookup: bool = False + """Fetch media with UUIDs upfront instead of first trying the processor cache.""" + tool_choice: Any | None = None """Request-level tool choice for renderers that need API metadata.""" @@ -129,6 +132,7 @@ def with_defaults( default_mm_processor_kwargs, self.mm_processor_kwargs, ), + skip_early_mm_lookup=self.skip_early_mm_lookup, tool_choice=self.tool_choice, response_format=self.response_format, )