feat(multimodal): support externally processed mm_kwargs with cache injection (#39502)

Signed-off-by: Krish Hung <krishung5@gmail.com>
Signed-off-by: krishung5 <krish@nvidia.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

tests/entrypoints/llm/test_mm_cache_external_injection.py

+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+Test that ``InputProcessor.inject_into_mm_cache()`` correctly injects
+pre-processed mm_kwargs into the processor cache and reports MM cache
+hit rate metrics accurately.
+
+This is used by frameworks like Dynamo that run the HF processor on a
+frontend and transfer pre-processed mm_kwargs to the backend, avoiding
+redundant processing.
+"""
+
+import logging
+
+import pytest
+import regex as re
+
+from tests.entrypoints.openai.chat_completion.test_vision import TEST_IMAGE_ASSETS
+from vllm import LLM, SamplingParams
+from vllm.renderers.params import ChatParams
+from vllm.v1.metrics import loggers as stat_loggers
+from vllm.v1.metrics.reader import Counter, Metric
+
+
+def _make_messages(image_url: str):
+    return [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": "image_url",
+                    "image_url": {"url": image_url},
+                },
+            ],
+        }
+    ]
+
+
+def _get_counter_value(metrics: list[Metric], name: str):
+    metric = next(m for m in metrics if m.name == name)
+    assert isinstance(metric, Counter)
+    return metric.value
+
+
+def _get_mm_cache_stats(metrics: list[Metric]):
+    mm_cache_queries = _get_counter_value(metrics, "vllm:mm_cache_queries")
+    mm_cache_hits = _get_counter_value(metrics, "vllm:mm_cache_hits")
+    return mm_cache_queries, mm_cache_hits
+
+
+def _get_mm_cache_log(llm: LLM, caplog_vllm: pytest.LogCaptureFixture) -> float:
+    caplog_vllm.clear()
+    with caplog_vllm.at_level(logging.INFO, logger=stat_loggers.__name__):
+        llm.llm_engine.do_log_stats()
+
+    assert len(caplog_vllm.records) == 1
+    msg = caplog_vllm.records[0].getMessage()
+
+    assert "MM cache hit rate" in msg
+    match = re.search(r"MM cache hit rate: ([0-9.]+)%", msg)
+    assert match is not None
+    return float(match.group(1))
+
+
+@pytest.mark.parametrize("image_urls", [TEST_IMAGE_ASSETS[:2]], indirect=True)
+@pytest.mark.parametrize("mm_processor_cache_type", ["lru", "shm"])
+def test_inject_into_mm_cache(
+    num_gpus_available,
+    image_urls,
+    mm_processor_cache_type,
+    caplog_vllm,
+):
+    """Test that inject_into_mm_cache() injects pre-processed mm_kwargs into
+    the processor cache and MM cache hit metrics are updated correctly.
+
+    Steps:
+    1. Two normal requests (same image) -> cache miss then hit (baseline)
+    2. Extract cached kwargs, call inject_into_mm_cache with a new hash,
+       then generate with a pre-rendered input -> verifies injection works
+    """
+    llm = LLM(
+        model="llava-hf/llava-1.5-7b-hf",
+        max_model_len=4096,
+        max_num_seqs=5,
+        enforce_eager=True,
+        disable_log_stats=False,
+        limit_mm_per_prompt={"image": 2},
+        mm_processor_cache_type=mm_processor_cache_type,
+    )
+
+    # Step 1: Normal requests to populate the cache
+    llm.chat(_make_messages(image_urls[0]))
+    assert _get_mm_cache_stats(llm.get_metrics()) == (1, 0)
+
+    llm.chat(_make_messages(image_urls[0]))
+    assert _get_mm_cache_stats(llm.get_metrics()) == (2, 1)
+    assert _get_mm_cache_log(llm, caplog_vllm) == pytest.approx(50.0)
+
+    # Step 2: Use a second image to get valid expanded tokens and
+    # placeholder positions via the renderer.
+    llm.chat(_make_messages(image_urls[1]))
+    queries_before = _get_mm_cache_stats(llm.get_metrics())[0]  # 3
+
+    renderer = llm.llm_engine.renderer
+    cache = renderer.mm_processor_cache
+    assert cache is not None, "Processor cache should be enabled"
+
+    _, eng_prompts = renderer.render_chat(
+        [_make_messages(image_urls[1])],
+        ChatParams(),
+    )
+    eng_input = eng_prompts[0]
+
+    # Inject pre-processed mm_kwargs with a NEW hash via public API
+    new_mm_hash = "deadbeef" * 8
+    mm_hashes = {"image": [new_mm_hash]}
+    mm_kwargs = eng_input["mm_kwargs"]
+
+    llm.llm_engine.input_processor.inject_into_mm_cache(mm_hashes, mm_kwargs)
+
+    # Build pre-rendered input (no externally_processed flag needed)
+    pre_rendered_input = {
+        "type": "multimodal",
+        "prompt_token_ids": eng_input["prompt_token_ids"],
+        "mm_kwargs": mm_kwargs,
+        "mm_hashes": mm_hashes,
+        "mm_placeholders": eng_input["mm_placeholders"],
+    }
+
+    llm.generate(
+        pre_rendered_input,
+        sampling_params=SamplingParams(max_tokens=1),
+    )
+
+    # Verify cache was queried and injection happened
+    queries_after = _get_mm_cache_stats(llm.get_metrics())[0]
+    assert queries_after > queries_before, (
+        "Cache should have been queried for the injected item"
+    )
+    mm_rate = _get_mm_cache_log(llm, caplog_vllm)
+    assert mm_rate >= 0.0, "MM cache hit rate should be reported"
+
+
+@pytest.mark.parametrize("image_urls", [TEST_IMAGE_ASSETS[:1]], indirect=True)
+def test_inject_into_mm_cache_without_cache(
+    num_gpus_available,
+    image_urls,
+):
+    """Test that inject_into_mm_cache works gracefully when processor cache
+    is disabled (mm_processor_cache_gb=0). Should not crash.
+    """
+    llm = LLM(
+        model="llava-hf/llava-1.5-7b-hf",
+        max_model_len=4096,
+        max_num_seqs=5,
+        enforce_eager=True,
+        disable_log_stats=False,
+        limit_mm_per_prompt={"image": 2},
+        mm_processor_cache_gb=0,
+    )
+
+    # Run a normal chat request first to warm up the model.
+    llm.chat(_make_messages(image_urls[0]))
+
+    # Use the renderer to get a proper EngineInput with expanded tokens
+    renderer = llm.llm_engine.renderer
+    _, eng_prompts = renderer.render_chat(
+        [_make_messages(image_urls[0])],
+        ChatParams(),
+    )
+    eng_input = eng_prompts[0]
+
+    mm_hashes = {"image": ["abcd1234" * 8]}
+    mm_kwargs = eng_input["mm_kwargs"]
+
+    # inject_into_mm_cache should not crash even without cache
+    llm.llm_engine.input_processor.inject_into_mm_cache(mm_hashes, mm_kwargs)
+
+    # Build and generate with pre-rendered input
+    pre_rendered_input = {
+        "type": "multimodal",
+        "prompt_token_ids": eng_input["prompt_token_ids"],
+        "mm_kwargs": mm_kwargs,
+        "mm_hashes": mm_hashes,
+        "mm_placeholders": eng_input["mm_placeholders"],
+    }
+
+    result = llm.generate(
+        pre_rendered_input,
+        sampling_params=SamplingParams(max_tokens=1),
+    )
+    assert len(result) == 1, "Should produce one output"
+    assert len(result[0].outputs) >= 1, "Should have at least one output sequence"

vllm/v1/engine/input_processor.py

@@ -172,6 +172,45 @@ class InputProcessor:
             return mm_hash
         return f"{lora_request.lora_name}:{mm_hash}"
 
+    def inject_into_mm_cache(
+        self,
+        mm_hashes: dict[str, list[str]],
+        mm_kwargs: dict[str, list],
+    ) -> None:
+        """Inject pre-processed mm_kwargs into the processor cache.
+
+        Call this when mm_kwargs have already been through the HF processor
+        externally (e.g. by a frontend that transfers pre-processed tensors
+        to the backend).  This ensures MM cache hit rate metrics are reported
+        accurately and avoids redundant processing on subsequent requests
+        with the same images.
+
+        Uses ``get_and_update_item()`` with an empty prompt_updates list,
+        since token expansion has already been handled externally.
+        """
+        cache = self.renderer.mm_processor_cache
+        if cache is None:
+            return
+        try:
+            for modality, hashes in mm_hashes.items():
+                items = mm_kwargs.get(modality, [])
+                for i, mm_hash in enumerate(hashes):
+                    if i < len(items) and items[i] is not None:
+                        # Insert into cache via get_and_update_item.
+                        # Use the returned item (may be an address for SHM
+                        # cache or the original item for LRU cache).
+                        items[i], _ = cache.get_and_update_item(
+                            (items[i], []),
+                            mm_hash,
+                        )
+            # Update cache stats to reflect the externally processed items
+            self.renderer.update_mm_cache_stats()
+        except Exception:
+            logger.warning(
+                "Failed to inject mm_kwargs into processor cache",
+                exc_info=True,
+            )
+
     @staticmethod
     def assign_request_id(request: EngineCoreRequest):
         """Replace the externally supplied request ID with an internal request ID