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
12 changes: 7 additions & 5 deletions components/src/dynamo/common/backend/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,11 +44,13 @@ class GenerateRequest(TypedDict, total=False):

Multimodal keys (``multi_modal_data``, ``mm_processor_kwargs``,
``mm_routing_info``) are populated by the frontend preprocessor when
the request carries media. ``encoder_result`` is set by the
frontend when forwarding a request from an Encode worker
to a downstream Prefill/Aggregated peer; engines read it via
:func:`dynamo.common.backend.multimodal.require_encoder_result`. All
four are object-shaped (``dict``) by contract.
the request carries media. ``encoder_result`` may be set by the frontend
when forwarding a request from an Encode worker or by an application
orchestrator before calling an Aggregated peer; engines read it via
:func:`dynamo.common.backend.multimodal.require_encoder_result`.
``encoder_result`` is deliberately opaque at this common layer; its
producer and consuming engine own the concrete schema. All four are
object-shaped (``dict``) by contract.

``model`` carries the requested model name (set by the Rust
preprocessor). Engines that support dynamic LoRA read it to route a
Expand Down
81 changes: 76 additions & 5 deletions components/src/dynamo/vllm/handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,10 @@
VisionEncoderBackend,
create_custom_encoder_adapter,
)
from .multimodal_utils.custom_encoder.handoff import external_encoder_request_conflicts
from .multimodal_utils.custom_encoder.handoff_consumer import (
ExternalEncoderHandoffConsumer,
)
from .multimodal_utils.prefill_worker_utils import MultiModalEmbeddingLoader
from .multimodal_utils.request_processor import (
IMAGE_URL_KEY,
Expand Down Expand Up @@ -3424,6 +3428,9 @@ def __init__(
encode_worker_client=encode_worker_client,
)
self._first_token_source = first_token_source
self._external_encoder_handoff_consumer: Optional[
ExternalEncoderHandoffConsumer
] = None

async def generate(self, request, context):
# Use context ID for request tracking and correlation
Expand All @@ -3438,6 +3445,16 @@ async def generate(self, request, context):
with time_and_log_code_section(
f"[DECODE] request: {request_id} generate"
) as decode_timer:
if self.use_vllm_tokenizer and request.get("encoder_result") is not None:
yield {
"finish_reason": (
"error: external encoder results require token-in/token-out "
"mode"
),
"index": 0,
"token_ids": [],
}
return
if self.use_vllm_tokenizer:
# Text-in-text-out mode: use InputParamManager and OpenAI-compatible format
generator = self._generate_text_mode(request, context, request_id)
Expand Down Expand Up @@ -3557,6 +3574,41 @@ async def _assemble_custom_encoder_prompt(
)
return prepared

async def _assemble_external_encoder_prompt(
self,
request: Dict[str, Any],
request_id: str,
) -> EmbedsPrompt:
"""Decode an external encoder result and prepare the vLLM prompt."""

conflicts = external_encoder_request_conflicts(request)

if conflicts:
raise InvalidArgument(
"encoder_result is authoritative and cannot be combined with "
"multimodal inputs, transfer data, or routing metadata: "
f"{sorted(conflicts)}"
)
encoder_result = request.get("encoder_result")
if not isinstance(encoder_result, Mapping):
raise InvalidArgument("encoder_result must be an object")
token_ids = request.get("token_ids")
if not isinstance(token_ids, list):
raise InvalidArgument("external encoder results require token_ids")
if self._external_encoder_handoff_consumer is None:
self._external_encoder_handoff_consumer = ExternalEncoderHandoffConsumer(
self.model_config,
self.config.engine_args,
)
prompt = await asyncio.to_thread(
self._external_encoder_handoff_consumer.prepare_prompt,
encoder_result,
token_ids,
)

logger.debug("Request %s: prepared external encoder prompt", request_id)
return prompt

async def _generate_token_mode(self, request, context, request_id):
"""Generate tokens using internal protocol format (token-in-token-out)."""
# Firstly extract disaggregated params from prefill result if available
Expand All @@ -3582,10 +3634,29 @@ async def _generate_token_mode(self, request, context, request_id):
)
is_decode_only = False
mode = DisaggregationMode.AGGREGATED

has_external_encoder_result = request.get("encoder_result") is not None
if has_external_encoder_result and mode != DisaggregationMode.AGGREGATED:
yield {
"finish_reason": (
"error: external encoder results currently require an "
"aggregated vLLM worker"
),
"index": 0,
"token_ids": [],
}
return
has_mm_data = request.get("multi_modal_data") is not None
custom_prompt: EmbedsPrompt | TokensPrompt | None = None
assembled_prompt: EmbedsPrompt | TokensPrompt | None = None

if (
if has_external_encoder_result:
assembled_prompt = await self._assemble_external_encoder_prompt(
request, request_id
)
multi_modal_data = None
mm_processor_kwargs = None
pre_rendered = None
elif (
mode == DisaggregationMode.AGGREGATED
and self._custom_encoder is not None
and has_mm_data
Expand All @@ -3596,7 +3667,7 @@ async def _generate_token_mode(self, request, context, request_id):
# Failures propagate as exceptions; the bindings map the type to a
# typed backend error, so an input fault answers 400 with its
# message and an engine fault stays a retryable 5xx.
custom_prompt = await self._assemble_custom_encoder_prompt(
assembled_prompt = await self._assemble_custom_encoder_prompt(
request,
request_id,
)
Expand Down Expand Up @@ -3631,8 +3702,8 @@ async def _generate_token_mode(self, request, context, request_id):
# branches without spelling out the full union.
prompt: Any
with _nvtx.annotate("mm_backend:build_prompt", color="yellow"):
if custom_prompt is not None:
prompt = custom_prompt
if assembled_prompt is not None:
prompt = assembled_prompt
elif pre_rendered is not None:
# pre_rendered is a MultiModalInput dict with "type": "multimodal".
# The engine's InputProcessor.process_inputs() will see the "type"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,15 @@
RawT,
VisionEncoderBackend,
)
from dynamo.vllm.multimodal_utils.custom_encoder.handoff import ExternalEncoderHandoff

__all__ = [
"AsyncVisionEncoder",
"ArtifactT",
"build_mixed_embeds",
"CustomEncoderAdapter",
"create_custom_encoder_adapter",
"ExternalEncoderHandoff",
"ItemT",
"Preprocessed",
"Qwen3VLImageEncoding",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -162,33 +162,22 @@ def _emit_text(text_ids: list[int]) -> None:
return prompt_embeds, out_token_ids, is_token_ids


class LinearEmbedsAdapter(CustomEncoderAdapter[torch.Tensor]):
"""Build mixed ``EmbedsPrompt`` inputs for a text-only decoder."""
class LinearEmbedsPromptBuilder:
"""Validate linear encoder artifacts and build a mixed vLLM prompt."""

def __init__(
self,
backend: VisionEncoderBackend[Any, Any, torch.Tensor],
model_config: Any,
engine_args: Any,
) -> None:
def __init__(self, model_config: Any, engine_args: Any) -> None:
if model_config is None:
raise ValueError("CustomEncoder requires the resolved vLLM ModelConfig")
if _is_multimodal_model(model_config):
raise ValueError(
"CustomEncoder does not yet support this multimodal decoder; "
"the linear EmbedsPrompt adapter is only valid for text-only models"
"the linear EmbedsPrompt adapter is only valid for text-only decoders"
)
if not getattr(engine_args, "enable_prompt_embeds", False):
raise ValueError(
"text-only CustomEncoder output requires --enable-prompt-embeds"
)
image_token_id = getattr(backend, "image_token_id", None)
if not isinstance(image_token_id, int) or isinstance(image_token_id, bool):
raise ValueError(
"text-only CustomEncoder output requires an integer image_token_id"
)

self._image_token_id = image_token_id
self._hidden_size = _hidden_size(model_config)
model_dtype = getattr(model_config, "dtype", None)
self._dtype = model_dtype if isinstance(model_dtype, torch.dtype) else None
Expand All @@ -197,14 +186,15 @@ def prepare_prompt(
self,
token_ids: list[int],
artifacts: Sequence[torch.Tensor],
) -> EmbedsPrompt | TokensPrompt:
"""Build a mixed prompt from per-image visual embedding tensors.
*,
image_token_id: int,
) -> EmbedsPrompt:
"""Build a mixed prompt from ordered per-image embedding tensors."""

Each artifact must be a CPU tensor shaped
``(n_visual_tokens, decoder_hidden_size)`` with the decoder's dtype.
Artifacts must appear in the same order as the image placeholders in
``token_ids``.
"""
if not isinstance(image_token_id, int) or isinstance(image_token_id, bool):
raise ValueError(
"text-only CustomEncoder output requires an integer image_token_id"
)
rows = list(artifacts)
for index, tensor in enumerate(rows):
if not isinstance(tensor, torch.Tensor):
Expand All @@ -224,10 +214,41 @@ def prepare_prompt(
)

prompt_embeds, prompt_token_ids, prompt_is_token_ids = build_mixed_embeds(
token_ids, rows, self._image_token_id
token_ids, rows, image_token_id
)
return EmbedsPrompt(
prompt_embeds=prompt_embeds,
prompt_token_ids=prompt_token_ids,
prompt_is_token_ids=prompt_is_token_ids,
)


class LinearEmbedsAdapter(CustomEncoderAdapter[torch.Tensor]):
"""Build mixed ``EmbedsPrompt`` inputs for a text-only decoder."""

def __init__(
self,
backend: VisionEncoderBackend[Any, Any, torch.Tensor],
model_config: Any,
engine_args: Any,
) -> None:
image_token_id = getattr(backend, "image_token_id", None)
if not isinstance(image_token_id, int) or isinstance(image_token_id, bool):
raise ValueError(
"text-only CustomEncoder output requires an integer image_token_id"
)
self._image_token_id = image_token_id
self._prompt_builder = LinearEmbedsPromptBuilder(model_config, engine_args)

def prepare_prompt(
self,
token_ids: list[int],
artifacts: Sequence[torch.Tensor],
) -> EmbedsPrompt | TokensPrompt:
"""Build a mixed prompt from per-image visual embedding tensors."""

return self._prompt_builder.prepare_prompt(
token_ids,
artifacts,
image_token_id=self._image_token_id,
)
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,13 @@
(the generic cross-request batcher) and ``AsyncVisionEncoder`` (the async
request-API glue). This module defines only the contract those drivers call.

The encoder runs in the **same process** as the aggregated vLLM worker (no
separate encode worker, no NIXL transfer): it turns image inputs into ordered,
producer-defined artifacts. The resolved downstream decoder selects an adapter
that validates those artifacts and constructs the final engine prompt.
The backend runs in the same process as Dynamo's encoder driver and turns image
inputs into ordered, producer-defined artifacts. Inline mode hosts that driver
inside the aggregated vLLM worker. ``ExternalEncoderHandoff`` hosts the same
driver in an application worker, packages linear tensor artifacts for the
request plane, and lets stock aggregated vLLM reconstruct the same prompt. The
resolved downstream decoder adapter validates the artifacts and constructs the
final engine prompt in either placement.

Division of labour (author vs. Dynamo):

Expand Down
Loading
Loading