Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 20 additions & 4 deletions e2e_test/fixtures/setup_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -246,7 +246,9 @@ def setup_backend(request: pytest.FixtureRequest):
trailing dict may add ``prefill_kv`` / ``decode_kv`` lists naming
each worker's KV transfer backend, so one fleet can mix transports,
and ``prefill_env`` / ``decode_env`` dicts of engine environment
for one leg (a deployment-injected ``SMG_PAIRING_PROTOCOL``)
for one leg (a deployment-injected ``SMG_PAIRING_PROTOCOL``) or
``prefill_args`` / ``decode_args`` lists of engine CLI flags for one
leg (a ``--language-model-only`` decode pool)
- ``@pytest.mark.gateway(policy=..., timeout=..., extra_args=...)``: Gateway config

Returns:
Expand Down Expand Up @@ -314,6 +316,9 @@ def setup_backend(request: pytest.FixtureRequest):
for key in ("prefill_env", "decode_env"):
if key in leg_options:
workers_config = {**workers_config, key: dict(leg_options[key])}
for key in ("prefill_args", "decode_args"):
if key in leg_options:
workers_config = {**workers_config, key: list(leg_options[key])}
log_dir = os.environ.get("E2E_LOG_DIR") or gateway_config.get("log_dir")

fail_count = _worker_start_failures.get(engine, 0)
Expand Down Expand Up @@ -445,6 +450,13 @@ def _per_worker_kv(raw, count: int, leg: str) -> list[str] | None:
return backends


def _merge_engine_args(shared: list[str] | None, leg_only: list[str] | None) -> list[str] | None:
"""Combine the class-wide engine flags with one PD leg's own flags."""
if not leg_only:
return shared
return [*(shared or []), *leg_only]


def _start_pd_leg(
*,
model_id: str,
Expand Down Expand Up @@ -532,8 +544,12 @@ def _setup_pd(

parallel_start = bool(workers_config.get("parallel_start"))
# The class marker's engine flags (a decode window, a context length)
# reach every leg, as they do for regular workers.
# reach every leg, as they do for regular workers. The ``prefill_args`` /
# ``decode_args`` leg options add flags to one leg only (a
# ``--language-model-only`` decode pool, say).
extra_engine_args = workers_config.get("extra_engine_args")
prefill_engine_args = _merge_engine_args(extra_engine_args, workers_config.get("prefill_args"))
decode_engine_args = _merge_engine_args(extra_engine_args, workers_config.get("decode_args"))
all_workers: list = []
try:
prefill_workers = _start_pd_leg(
Expand All @@ -547,7 +563,7 @@ def _setup_pd(
wait_ready=not parallel_start,
tp=prefill_tp,
kv_backends=prefill_kv,
extra_engine_args=extra_engine_args,
extra_engine_args=prefill_engine_args,
extra_env=workers_config.get("prefill_env"),
)
all_workers.extend(prefill_workers)
Expand All @@ -565,7 +581,7 @@ def _setup_pd(
wait_ready=not parallel_start,
tp=decode_tp,
kv_backends=decode_kv,
extra_engine_args=extra_engine_args,
extra_engine_args=decode_engine_args,
extra_env=workers_config.get("decode_env"),
)
all_workers.extend(decode_workers)
Expand Down
83 changes: 83 additions & 0 deletions e2e_test/router/test_pd_multimodal.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,11 @@
path the gateway took: Qwen3-VL forwards references, Phi-3.5 stays on the
router path because its placeholder is not one a vLLM worker expands.

The language-model-only lanes run the decode pool with
``--language-model-only`` (no vision encoder, encoder-cache budget 0): the
grid-less model takes the fully stripped decode leg, while the M-RoPE model
cannot pair with such a pool and must be rejected with a clear 400.

Usage:
E2E_RUNTIME=vllm pytest e2e_test/router/test_pd_multimodal.py -v
"""
Expand All @@ -22,6 +27,7 @@
import logging
from pathlib import Path

import httpx
import pytest
from infra import assert_mm_processing
from PIL import Image
Expand Down Expand Up @@ -150,3 +156,80 @@ def test_parallel_sampling_keeps_vision_on_decode(self, setup_backend):
answer = _first_word(choice.message.content)
logger.info("PD multimodal n=2 answer (pug in blanket): %s", answer)
assert answer == "yes", f"Expected 'yes' from each sample, got: {answer}"


def _post_blanket(gateway, model: str, image_url: str, **sampling) -> httpx.Response:
return httpx.post(
f"{gateway.base_url}/v1/chat/completions",
json={
"model": model,
"messages": _blanket_messages(image_url),
**sampling,
},
timeout=120.0,
)


@pytest.mark.engine("vllm")
@pytest.mark.gpu(2)
@pytest.mark.model("microsoft/Phi-3.5-vision-instruct")
@pytest.mark.e2e
@pytest.mark.parametrize(
"setup_backend",
[("pd_grpc", (1, 1, {"decode_args": ["--language-model-only"]}))],
indirect=True,
)
class TestPDMultimodalLanguageModelOnlyDecode:
"""Language-model-only decode pool (the production P/D shape).

The decode workers run with no vision encoder and an encoder-cache
budget of 0; the gateway strips the decode leg down to the expanded
token ids, the KV handoff and the per-image content hashes, so the
decode engine sees a pure-text prompt and never schedules its encoder
cache. Grid-less (standard-RoPE) model: isolation still comes from the
hash-derived cache_salt.
"""

def test_different_images_same_prefix_do_not_alias(self, setup_backend):
backend, model, client, gateway = setup_backend
_assert_no_cross_image_aliasing(client, model)
# A language-model-only decode worker cannot expand media references,
# so the media stay router-processed.
assert_mm_processing(gateway, worker_expandable=False)

def test_parallel_sampling_is_rejected_clearly(self, setup_backend):
# n>1 has no KV handoff; a pixel-less decode worker cannot recompute
# a multimodal prompt locally, so the gateway must say so in a 400
# rather than fail deep in the engine.
backend, model, client, gateway = setup_backend
resp = _post_blanket(gateway, model, _PUG_IMAGE, temperature=0.1, max_tokens=16, n=2)
logger.info("n>1 on language-model-only decode: %s %s", resp.status_code, resp.text[:300])
assert resp.status_code == 400, f"expected 400, got {resp.status_code}: {resp.text[:300]}"
assert "language-model-only" in resp.text


@pytest.mark.engine("vllm")
@pytest.mark.gpu(2)
@pytest.mark.model("Qwen/Qwen3-VL-8B-Instruct")
@pytest.mark.e2e
@pytest.mark.parametrize(
"setup_backend",
[("pd_grpc", (1, 1, {"decode_args": ["--language-model-only"]}))],
indirect=True,
)
class TestPDMultimodalMropeLanguageModelOnlyDecode:
"""M-RoPE models cannot pair with a language-model-only decode pool.

The decode leg needs the grid tensors for positions; the pixel-less
decode worker cannot accept them. The gateway rejects the pairing with a
clear 400 instead of silently mis-rotating positions.
"""

def test_mrope_request_is_rejected_clearly(self, setup_backend):
backend, model, client, gateway = setup_backend
resp = _post_blanket(gateway, model, _PUG_IMAGE, temperature=0, max_tokens=16)
logger.info(
"M-RoPE on language-model-only decode: %s %s", resp.status_code, resp.text[:300]
)
assert resp.status_code == 400, f"expected 400, got {resp.status_code}: {resp.text[:300]}"
assert "language-model-only" in resp.text
75 changes: 75 additions & 0 deletions grpc_servicer/smg_grpc_servicer/vllm/mm_salt.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,82 @@
"""Engine-free multimodal helpers for tensor-stripped (PD decode) legs."""

import logging
from collections.abc import Sequence

logger = logging.getLogger(__name__)


def engine_accepts_mm_inputs(model_config) -> bool:
"""Whether the engine runs a vision encoder for pixel payloads.

``is_multimodal_model`` only reports the architecture: it stays true
under ``--language-model-only``, which zeroes every modality limit and
drops the vision encoder (encoder-cache budget 0), and it is true for
``enable_mm_embeds``-only models that ingest pre-computed embeddings but
cannot encode pixels. The router reads this through the worker's
``supports_vision`` label to decide whether a decode leg may carry an mm
payload, so the answer must come from vLLM's own check. That check moved
between releases:

- vLLM main: ``ModelConfig.supports_multimodal_inputs`` property;
- vLLM 0.19-0.20 (the servicer's supported range): the equivalent
``MULTIMODAL_REGISTRY.supports_multimodal_inputs(model_config)``.
"""
supports = getattr(model_config, "supports_multimodal_inputs", None)
if supports is None:
supports = _registry_supports_multimodal_inputs(model_config)
if not supports:
return False
# vLLM counts enable_mm_embeds-only engines as accepting multimodal
# inputs (they ingest embeddings), but they have no encoder for the pixel
# payloads supports_vision speaks for.
return not _mm_embeds_only(model_config)


def _registry_supports_multimodal_inputs(model_config) -> bool:
"""vLLM 0.19-0.20 fallback: the check lives on the multimodal registry."""
if not model_config.is_multimodal_model:
return False
try:
from vllm.multimodal import MULTIMODAL_REGISTRY
except ImportError:
# Engine-free context (unit tests): the architecture is all we have.
return True
try:
return bool(MULTIMODAL_REGISTRY.supports_multimodal_inputs(model_config))
except ValueError:
# No registered processor for this architecture: text-only.
return False
except Exception:
# An unexpected probe failure keeps the previous
# architecture-based answer rather than disabling a healthy
# full-vision worker.
logger.warning(
"supports_multimodal_inputs probe failed; reporting the architecture's "
"multimodal capability",
exc_info=True,
)
return True


def _mm_embeds_only(model_config) -> bool:
"""Whether the engine ingests only pre-computed embeddings (no encoder).

True when ``enable_mm_embeds`` is on and every explicitly listed modality
limit is 0 — including the ``--language-model-only`` case, where
``get_limit_per_prompt`` reads 0 for every modality.
"""
mm_config = getattr(model_config, "multimodal_config", None)
if mm_config is None or not getattr(mm_config, "enable_mm_embeds", False):
return False
limits = getattr(mm_config, "limit_per_prompt", None) or {}
get_limit = getattr(mm_config, "get_limit_per_prompt", None)
if get_limit is None:
return False
if getattr(mm_config, "language_model_only", False):
return True
return bool(limits) and all(get_limit(modality) == 0 for modality in limits)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Nit: _mm_embeds_only iterates only the explicitly configured limit_per_prompt keys, while vLLM's own embeds-only check iterates the model's supported modalities (info.supported_mm_limits). The two diverge when enable_mm_embeds=True and the operator zeroes a subset of modalities, e.g. --limit-mm-per-prompt image=0 on an image+video model: every explicit limit is 0, so this returns True and the worker advertises supports_vision=false, even though the video encoder is still live (default non-zero limit). That misclassifies a healthy encoder-bearing worker — the same "don't blind a healthy worker" direction the except Exception handler in _registry_supports_multimodal_inputs deliberately guards against — and for an M-RoPE model would turn a valid P/D pairing into a hard pd_decode_language_model_only_mrope 400. Niche config, so not blocking, but if the model's supported-modality set (or a fully populated limit_per_prompt) is reachable here, iterating that instead would close the gap.



def has_preprocessed_mm_payload(mm_inputs) -> bool:
"""True when the payload carries tensors the preprocessed path can use.
Expand Down
13 changes: 10 additions & 3 deletions grpc_servicer/smg_grpc_servicer/vllm/servicer.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,11 @@
build_mm_processor,
env_int,
)
from smg_grpc_servicer.vllm.mm_salt import has_preprocessed_mm_payload, mm_identity_cache_salt
from smg_grpc_servicer.vllm.mm_salt import (
engine_accepts_mm_inputs,
has_preprocessed_mm_payload,
mm_identity_cache_salt,
)
from smg_grpc_servicer.vllm.mm_tensors import tensor_from_proto

from ..pd_pairing import pairing_protocol_from_env
Expand Down Expand Up @@ -630,7 +634,7 @@ async def GetModelInfo(
is_generation=model_config.runner_type == "generate",
max_context_length=model_config.max_model_len,
vocab_size=model_config.get_vocab_size(),
supports_vision=model_config.is_multimodal_model,
supports_vision=engine_accepts_mm_inputs(model_config),
served_model_name=model_config.served_model_name or model_config.model,
tokenizer_path=model_config.tokenizer or "",
model_type=getattr(hf_config, "model_type", "") or "",
Expand Down Expand Up @@ -673,9 +677,12 @@ async def GetServerInfo(

mm_processor = ""
mm_media_ref_schemes = ""
# A --language-model-only engine accepts no multimodal inputs, so it
# must not advertise worker-side media processing either: the router
# would send media references this worker cannot expand.
if (
self._mm_processor is not None
and self.engine.model_config.is_multimodal_model
and engine_accepts_mm_inputs(self.engine.model_config)
and await self._mm_processor.probe()
):
mm_processor = self._mm_processor.name
Expand Down
Loading
Loading