feat(multimodal): honor optional preprocess in the serial AsyncVisionEncoder

Read preprocess_concurrency from the backend (driver arg overrides for tuning);
0 ⇒ no pool — raws go straight to forward_batch with no A5 barrier (the single
forward is already all-or-nothing for the request). Make the example
HitchhikersVisionEncoder a passthrough (drop its identity preprocess override)
and document the pool opt-in in the Qwen base.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: furionw

components/src/dynamo/vllm/multimodal_utils/async_vision_encoder.py

@@ -47,24 +47,32 @@ class AsyncVisionEncoder(Generic[RawT, ItemT]):
 
     The worker calls ``load`` once at startup and ``await``s ``encode`` per
     request; ``shutdown`` on teardown. All model knowledge lives in ``backend``;
-    this class owns the preprocess pool, the A5 barrier, and the single actor
-    thread that runs ``build`` / ``forward_batch`` / ``close``.
+    this class owns the optional preprocess pool, the A5 barrier, and the single
+    actor thread that runs ``build`` / ``forward_batch`` / ``close``.
     """
 
     def __init__(
         self,
         backend: VisionEncoderBackend[RawT, ItemT],
         *,
-        preprocess_concurrency: int = 4,
+        preprocess_concurrency: int | None = None,
         name: str = "vision-encoder",
     ) -> None:
-        if preprocess_concurrency < 1:
-            raise ValueError("preprocess_concurrency must be >= 1")
+        # The backend declares whether it needs off-loop preprocessing; the
+        # explicit arg is an override for tuning. 0 ⇒ no pool (raws pass straight
+        # to forward_batch).
+        conc = (
+            backend.preprocess_concurrency
+            if preprocess_concurrency is None
+            else preprocess_concurrency
+        )
+        if conc < 0:
+            raise ValueError("preprocess_concurrency must be >= 0")
         self._backend = backend
-        self._preprocess_concurrency = preprocess_concurrency
+        self._preprocess_concurrency = conc
         self._name = name
         self._actor: ThreadPoolExecutor | None = None  # build + every forward
-        self._pool: ThreadPoolExecutor | None = None  # off-loop preprocess
+        self._pool: ThreadPoolExecutor | None = None  # off-loop preprocess (or None)
 
     # ---- lifecycle ---------------------------------------------------------
 
@@ -75,17 +83,22 @@ def load(self, model_id: str) -> None:
         so a misconfigured encoder errors at startup. Single-shot: a second
         ``load()`` raises rather than orphaning the first actor thread and model.
         """
-        if self._actor is not None or self._pool is not None:
+        if self._actor is not None:
             raise RuntimeError("AsyncVisionEncoder.load() called twice")
         try:
             # One actor thread so build + every forward share a thread; a single
             # worker also serializes forwards (FIFO) — no cross-request batching.
             self._actor = ThreadPoolExecutor(
                 max_workers=1, thread_name_prefix=f"{self._name}-actor"
             )
-            self._pool = ThreadPoolExecutor(
-                max_workers=self._preprocess_concurrency,
-                thread_name_prefix=f"{self._name}-pre",
+            # No pool when concurrency is 0 — preprocess is skipped (passthrough).
+            self._pool = (
+                ThreadPoolExecutor(
+                    max_workers=self._preprocess_concurrency,
+                    thread_name_prefix=f"{self._name}-pre",
+                )
+                if self._preprocess_concurrency > 0
+                else None
             )
             self._actor.submit(self._backend.build, model_id).result()
             self.validate()
@@ -110,30 +123,38 @@ def get_image_placeholder_token_id(self) -> int:
     # ---- request path ------------------------------------------------------
 
     async def encode(self, raws: List[RawT]) -> List[torch.Tensor]:
-        """Preprocess (off-loop, A5 barrier) then run a single serial forward.
-
-        Returns one ``(n_visual_tokens, lm_hidden_dim)`` CPU tensor per raw input,
-        in order. Raises if any image's preprocess fails (no GPU work) or if the
-        forward fails.
+        """Optionally preprocess (off-loop, A5 barrier) then run a single serial
+        forward.
+
+        With no preprocess pool (``preprocess_concurrency == 0``) raws go straight
+        to ``forward_batch`` (the backend folds any prep in there). Returns one
+        ``(n_visual_tokens, lm_hidden_dim)`` CPU tensor per raw input, in order.
+        Raises if any image's preprocess fails (no GPU work) or if the forward
+        fails.
         """
-        if self._actor is None or self._pool is None:
+        if self._actor is None:
             raise RuntimeError("AsyncVisionEncoder.encode() called before load()")
         if not raws:
             return []
         loop = asyncio.get_running_loop()
-        # A5 barrier: preprocess all images concurrently, wait for EVERY one to
-        # settle, and run the forward only if all succeeded. return_exceptions=True
-        # makes the gather a true barrier (no short-circuit).
-        tasks = [
-            loop.run_in_executor(self._pool, self._backend.preprocess, raw)
-            for raw in raws
-        ]
-        settled = await asyncio.gather(*tasks, return_exceptions=True)
-        errors = [r for r in settled if isinstance(r, BaseException)]
-        if errors:
-            raise errors[0]
-        preprocessed: List[Preprocessed] = list(settled)  # type: ignore[arg-type]
-        items = [p.item for p in preprocessed]
+        if self._pool is None:
+            # No preprocess phase: raw IS the item. No A5 barrier needed — the
+            # single forward is already all-or-nothing for the request.
+            items: List[ItemT] = list(raws)  # type: ignore[arg-type]  # ItemT==RawT
+        else:
+            # A5 barrier: preprocess all images concurrently, wait for EVERY one to
+            # settle, run the forward only if all succeeded. return_exceptions=True
+            # makes the gather a true barrier (no short-circuit).
+            tasks = [
+                loop.run_in_executor(self._pool, self._backend.preprocess, raw)
+                for raw in raws
+            ]
+            settled = await asyncio.gather(*tasks, return_exceptions=True)
+            errors = [r for r in settled if isinstance(r, BaseException)]
+            if errors:
+                raise errors[0]
+            preprocessed: List[Preprocessed] = list(settled)  # type: ignore[arg-type]
+            items = [p.item for p in preprocessed]
         # Direct, serialized forward on the actor thread (eager; target_bucket
         # defaults to None — there is no graph ladder until CUDA-graph batching
         # is supported).

components/src/dynamo/vllm/tests/multimodal_utils/test_vllm_async_vision_encoder.py

@@ -35,6 +35,9 @@ class _FakeBackend(VisionEncoderBackend):
     """A CPU-only fake backend; records its threads and forward-call boundaries."""
 
     image_token_id = 151655
+    # This fake overrides preprocess, so opt into the off-loop pool (the default
+    # is 0 ⇒ no pool); the passthrough path is covered by its own test below.
+    preprocess_concurrency = 4
 
     def __init__(self, *, fail_on=None):
         self.fail_on = set(fail_on or ())
@@ -185,6 +188,37 @@ def test_shutdown_before_load_is_safe():
     AsyncVisionEncoder(_FakeBackend()).shutdown()  # no-op, no raise
 
 
-def test_preprocess_concurrency_must_be_positive():
+async def test_passthrough_skips_preprocess_when_no_pool():
+    """With no pool (preprocess_concurrency=0) preprocess is skipped and raws go
+    straight to forward_batch — no A5 barrier, no pool thread."""
+
+    class _PassthroughBackend(_FakeBackend):
+        preprocess_concurrency = 0
+
+        def preprocess(self, raw):
+            raise AssertionError("preprocess must not run in passthrough mode")
+
+    be = _PassthroughBackend()
+    enc = AsyncVisionEncoder(be)
+    enc.load("m")
+    try:
+        assert enc._pool is None  # no pool created
+        out = await enc.encode(["a", "bb"])
+        assert len(out) == 2
+        assert be.forward_calls == [["a", "bb"]]  # raws passed straight through
+    finally:
+        enc.shutdown()
+
+
+def test_preprocess_concurrency_zero_disables_pool():
+    enc = AsyncVisionEncoder(_FakeBackend(), preprocess_concurrency=0)
+    enc.load("m")
+    try:
+        assert enc._pool is None
+    finally:
+        enc.shutdown()
+
+
+def test_preprocess_concurrency_rejects_negative():
     with pytest.raises(ValueError, match="preprocess_concurrency"):
-        AsyncVisionEncoder(_FakeBackend(), preprocess_concurrency=0)
+        AsyncVisionEncoder(_FakeBackend(), preprocess_concurrency=-1)

examples/custom_encoder/hitchhikers_vision_encoder.py

@@ -44,7 +44,6 @@
 from safetensors import safe_open
 from transformers.utils import cached_file
 
-from dynamo.vllm.multimodal_utils.vision_encoder_backend import Preprocessed
 from examples.custom_encoder.qwen_vision_encoder import QwenVisionEncoderBackend
 
 logger = logging.getLogger(__name__)
@@ -97,7 +96,9 @@ class HitchhikersVisionEncoder(QwenVisionEncoderBackend):
     so the spliced prompt reads as a coherent sentence.
     """
 
-    # Eager: no graph ladder. Count-based budget (cost == 1 per image).
+    # Eager: no graph ladder. Count-based budget (cost == 1 per image). The URL is
+    # ignored, so there is nothing to preprocess: it inherits the identity-default
+    # preprocess + preprocess_concurrency=0, and raws go straight to forward_batch.
     buckets = None
     max_batch_cost = 8
 
@@ -112,12 +113,6 @@ def build(self, model_id: str) -> None:
             _PHRASE,
         )
 
-    def preprocess(self, image_url: str) -> Preprocessed[str]:
-        """Pass-through: the URL is ignored, so there is nothing to fetch/decode.
-
-        cost=1 (count-based)."""
-        return Preprocessed(item=image_url, cost=1)
-
     def forward_batch(
         self, items: List[str], target_bucket: Optional[int] = None
     ) -> List[torch.Tensor]:

examples/custom_encoder/qwen_vision_encoder.py

@@ -5,9 +5,11 @@
 
 Hardcodes the Qwen ``<|image_pad|>`` placeholder id and loads the model tokenizer
 (handy for subclasses that tokenize text). A concrete Qwen-family encoder
-subclasses this and implements only ``preprocess`` + ``forward_batch``.
+subclasses this and implements ``forward_batch`` — plus ``preprocess`` (and
+``preprocess_concurrency > 0``) only when it needs off-loop fetch/resize.
 
     class MyQwenEncoder(QwenVisionEncoderBackend):
+        preprocess_concurrency = 4              # enable the off-loop pool
         def build(self, model_id):
             super().build(model_id)             # loads self.tokenizer
             # ... load ViT + projector (pick the device yourself) ...
@@ -30,8 +32,9 @@ class QwenVisionEncoderBackend(VisionEncoderBackend):
 
     Hardcodes ``image_token_id`` to Qwen3-VL's ``<|image_pad|>`` (151655) — override
     it for other versions (e.g. 248056 for Qwen3.5). ``build`` loads the model
-    tokenizer; ``preprocess`` and ``forward_batch`` stay abstract, so this class
-    cannot be instantiated directly — subclass it and implement them.
+    tokenizer; ``forward_batch`` stays abstract, so this class cannot be
+    instantiated directly — subclass it and implement ``forward_batch`` (and
+    ``preprocess`` only if it needs off-loop prep).
     """
 
     # Qwen3-VL <|image_pad|>; override for other Qwen versions.