feat(vllm): add audio transcription serving

components/src/dynamo/vllm/args.py

@@ -325,6 +325,8 @@ def _unsupported_fpm_trace_role(dynamo_config: Config) -> Optional[str]:
         return "embedding"
     if dynamo_config.classify_worker:
         return "classify"
+    if dynamo_config.transcription_worker:
+        return "transcription"
     if dynamo_config.headless:
         return "headless"
     if dynamo_config.disaggregation_mode == DisaggregationMode.ENCODE:
@@ -368,6 +370,14 @@ def update_engine_config_with_dynamo(
                 "Pooling-family worker: defaulting --enable-prefix-caching to False"
             )
             engine_config.enable_prefix_caching = False
+        elif dynamo_config.transcription_worker:
+            # Whisper is an encoder-decoder model. vLLM does not support
+            # prefix caching for cross-attention KV caches, and enabling it
+            # causes the first transcription request to terminate EngineCore.
+            logger.debug(
+                "Transcription worker: defaulting --enable-prefix-caching to False"
+            )
+            engine_config.enable_prefix_caching = False
         else:
             logger.debug(
                 "--enable-prefix-caching or --no-enable-prefix-caching not specified. "

components/src/dynamo/vllm/backend_args.py

@@ -272,6 +272,15 @@ def add_arguments(self, parser) -> None:
             "supported with more than one process.",
         )
 
+        add_negatable_bool_argument(
+            g,
+            flag_name="--transcription-worker",
+            env_var="DYN_VLLM_TRANSCRIPTION_WORKER",
+            default=False,
+            help="Run as an aggregated speech-to-text worker using vLLM's native "
+            "OpenAI transcription serving layer.",
+        )
+
         add_negatable_bool_argument(
             g,
             flag_name="--realtime",
@@ -619,6 +628,7 @@ class DynamoVllmConfig(ConfigBase):
     embedding_worker: bool = False
     embedding_frontend_tokenization: bool = False
     embedding_worker_processes: int = 1
+    transcription_worker: bool = False
     realtime: bool = False
     classify_worker: bool = False
 
@@ -684,6 +694,7 @@ def validate(self) -> None:
         self._validate_embedding_frontend_tokenization()
         self._validate_embedding_worker_exclusivity()
         self._validate_embedding_worker_processes()
+        self._validate_transcription_worker_exclusivity()
         self._validate_realtime_worker_exclusivity()
         self._validate_classify_worker_exclusivity()
         self._validate_custom_encoder()
@@ -1064,6 +1075,46 @@ def _validate_realtime_worker_exclusivity(self) -> None:
         if getattr(getattr(self, "engine_args", None), "enable_lora", False):
             raise ValueError("--realtime cannot be combined with --enable-lora.")
 
+    def _validate_transcription_worker_exclusivity(self) -> None:
+        """Transcription serving uses a dedicated aggregated ASR worker."""
+        if not self.transcription_worker:
+            return
+        if self.disaggregation_mode != DisaggregationMode.AGGREGATED:
+            mode = (
+                self.disaggregation_mode.value
+                if isinstance(self.disaggregation_mode, DisaggregationMode)
+                else self.disaggregation_mode
+            )
+            raise ValueError(
+                "--transcription-worker is only valid with "
+                f"--disaggregation-mode=agg (got {mode})."
+            )
+        for enabled, option in (
+            (self.embedding_worker, "--embedding-worker"),
+            (self.classify_worker, "--classify-worker"),
+            (self.realtime, "--realtime"),
+            (bool(self.custom_encoder_class), "--custom-encoder-class"),
+            (self.gms_shadow_mode, "--gms-shadow-mode"),
+            (self.enable_rl, "--enable-rl"),
+            (self.headless, "--headless"),
+        ):
+            if enabled:
+                raise ValueError(
+                    f"--transcription-worker cannot be combined with {option}."
+                )
+        if self.enable_multimodal:
+            raise ValueError(
+                "--transcription-worker cannot be combined with multimodal worker flags."
+            )
+        if self.benchmark_mode is not None:
+            raise ValueError(
+                "--transcription-worker cannot be combined with --benchmark-mode."
+            )
+        if getattr(getattr(self, "engine_args", None), "enable_lora", False):
+            raise ValueError(
+                "--transcription-worker cannot be combined with --enable-lora."
+            )
+
     def _validate_classify_worker_exclusivity(self) -> None:
         """Classify worker is aggregated-only and exclusive of multimodal /
         embedding roles. Mirrors the embedding-worker constraints — both are

components/src/dynamo/vllm/health_check.py

@@ -7,7 +7,10 @@
 This module defines the default health check payload for vLLM backends.
 """
 
+import base64
+import io
 import logging
+import wave
 from typing import TYPE_CHECKING, Any, Optional
 
 from dynamo.health_check import HEALTH_CHECK_KEY, HealthCheckPayload
@@ -152,6 +155,34 @@ def to_dict(self) -> dict[str, Any]:
         return _layer_probe_marker(super().to_dict())
 
 
+def _silent_wav_b64() -> str:
+    buffer = io.BytesIO()
+    with wave.open(buffer, "wb") as wav:
+        wav.setnchannels(1)
+        wav.setsampwidth(2)
+        wav.setframerate(16_000)
+        wav.writeframes(b"\x00" * 3_200)
+    return base64.b64encode(buffer.getvalue()).decode("ascii")
+
+
+class VllmTranscriptionHealthCheckPayload(HealthCheckPayload):
+    """Real, short ASR probe matching TranscriptionWorkerHandler input."""
+
+    def __init__(self, model_name: Optional[str] = None):
+        self.default_payload: dict[str, Any] = {
+            "audio_b64": _silent_wav_b64(),
+            "filename": "health-check.wav",
+            "language": "en",
+            "response_format": "json",
+        }
+        if model_name is not None:
+            self.default_payload["model"] = model_name
+        super().__init__()
+
+    def to_dict(self) -> dict[str, Any]:
+        return _layer_probe_marker(super().to_dict())
+
+
 class VllmPrefillHealthCheckPayload(HealthCheckPayload):
     """
     vLLM-specific health check payload for prefill workers in disaggregated mode.

components/src/dynamo/vllm/tests/test_vllm_backend_args.py

@@ -49,6 +49,12 @@ def create_config() -> DynamoVllmConfig:
     config.embedding_worker = False
     config.embedding_frontend_tokenization = False
     config.embedding_worker_processes = 1
+    config.transcription_worker = False
+    config.realtime = False
+    config.classify_worker = False
+    config.custom_encoder_class = None
+    config.gms_shadow_mode = False
+    config.enable_rl = False
     config.headless = False
     config.benchmark_mode = None
     config.use_vllm_tokenizer = False
@@ -318,6 +324,38 @@ def test_environment_rejects_invalid_value(self, monkeypatch):
             DynamoVllmArgGroup().add_arguments(parser)
 
 
+class TestTranscriptionWorkerExclusivity:
+    def test_baseline_aggregated_is_accepted(self):
+        config = create_config()
+        config.transcription_worker = True
+        config.disaggregation_mode = DisaggregationMode.AGGREGATED
+        config._validate_transcription_worker_exclusivity()
+
+    def test_non_aggregated_mode_is_rejected(self):
+        config = create_config()
+        config.transcription_worker = True
+        config.disaggregation_mode = DisaggregationMode.DECODE
+        with pytest.raises(ValueError, match="disaggregation-mode=agg"):
+            config._validate_transcription_worker_exclusivity()
+
+    @pytest.mark.parametrize(
+        ("option", "value"),
+        [
+            ("embedding_worker", True),
+            ("classify_worker", True),
+            ("realtime", True),
+            ("headless", True),
+        ],
+    )
+    def test_standalone_role_combinations_are_rejected(self, option, value):
+        config = create_config()
+        config.transcription_worker = True
+        config.disaggregation_mode = DisaggregationMode.AGGREGATED
+        setattr(config, option, value)
+        with pytest.raises(ValueError, match=option.replace("_", "-")):
+            config._validate_transcription_worker_exclusivity()
+
+
 class TestRealtimeWorkerExclusivity:
     def test_baseline_aggregated_is_accepted(self):
         config = create_config()

components/src/dynamo/vllm/tests/test_vllm_health_check_payloads.py

@@ -9,7 +9,10 @@
 parity with trtllm/sglang for any future marker-gated behavior.
 """
 
+import base64
+import io
 import json
+import wave
 
 import pytest
 
@@ -19,11 +22,13 @@
     VllmHealthCheckPayload,
     VllmOmniHealthCheckPayload,
     VllmPrefillHealthCheckPayload,
+    VllmTranscriptionHealthCheckPayload,
 )
 
 pytestmark = [
     pytest.mark.unit,
     pytest.mark.vllm,
+    pytest.mark.core,
     pytest.mark.gpu_0,
     pytest.mark.pre_merge,
 ]
@@ -32,6 +37,10 @@
     pytest.param(VllmHealthCheckPayload, id="VllmHealthCheckPayload"),
     pytest.param(VllmPrefillHealthCheckPayload, id="VllmPrefillHealthCheckPayload"),
     pytest.param(VllmOmniHealthCheckPayload, id="VllmOmniHealthCheckPayload"),
+    pytest.param(
+        VllmTranscriptionHealthCheckPayload,
+        id="VllmTranscriptionHealthCheckPayload",
+    ),
     pytest.param(
         VllmEmbeddingHealthCheckPayload,
         id="VllmEmbeddingHealthCheckPayload",
@@ -89,3 +98,19 @@ def test_embedding_payload_omits_model_when_no_name():
     assert "model" not in payload
     assert payload.get("input") == "probe"
     assert payload.get(HEALTH_CHECK_KEY) is True
+
+
+def test_transcription_payload_contains_valid_short_wav():
+    payload = VllmTranscriptionHealthCheckPayload(model_name="whisper").to_dict()
+
+    assert payload.get("model") == "whisper"
+    assert payload.get("language") == "en"
+    assert payload.get("response_format") == "json"
+    assert payload.get(HEALTH_CHECK_KEY) is True
+
+    audio = base64.b64decode(payload["audio_b64"], validate=True)
+    with wave.open(io.BytesIO(audio), "rb") as wav:
+        assert wav.getnchannels() == 1
+        assert wav.getsampwidth() == 2
+        assert wav.getframerate() == 16_000
+        assert wav.getnframes() == 1_600