feat(examples): add remote custom encoder orchestrator

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

@@ -0,0 +1,148 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+import pytest
+import torch
+
+from dynamo.common.external_encoder import (
+    ExternalEncoderResult,
+    decode_request_plane_tensor,
+)
+from dynamo.experimental.llm import LLMUnaryClient
+from dynamo.llm.exceptions import InvalidArgument
+from examples.custom_encoder.remote.orchestrator_worker import (
+    ExternalEncoderOrchestrator,
+    InlineEncoder,
+)
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.pre_merge,
+    pytest.mark.vllm,
+    pytest.mark.gpu_0,
+    pytest.mark.multimodal,
+]
+
+
+class _Encoder:
+    def __init__(self, artifacts: list[Any]) -> None:
+        self.artifacts = artifacts
+        self.raws: list[str] | None = None
+        self.closed = False
+
+    async def encode(self, raws: list[str]) -> list[Any]:
+        self.raws = raws
+        return self.artifacts
+
+    def shutdown(self) -> None:
+        self.closed = True
+
+
+class _Client:
+    def __init__(self, chunks: list[dict[str, Any]]) -> None:
+        self.chunks = chunks
+        self.request: dict[str, Any] | None = None
+        self.context: Any = None
+
+    async def round_robin(
+        self,
+        request: Any,
+        *,
+        annotated: bool,
+        context: Any = None,
+    ) -> AsyncIterator[dict[str, Any]]:
+        assert annotated is False
+        self.request = request
+        self.context = context
+
+        async def stream() -> AsyncIterator[dict[str, Any]]:
+            for chunk in self.chunks:
+                yield chunk
+
+        return stream()
+
+
+def _request() -> dict[str, Any]:
+    return {
+        "token_ids": [1, 99, 2, 99, 3],
+        "multi_modal_data": {
+            "image_url": [
+                {"Url": "https://example.com/one.png"},
+                {"Url": "https://example.com/two.png"},
+            ]
+        },
+        "extra_args": {"mm_kwargs_nixl": {"unused": True}, "keep": 1},
+    }
+
+
+async def test_orchestrator_packages_inline_encoder_result() -> None:
+    first = torch.arange(8, dtype=torch.bfloat16).reshape(2, 4)
+    second = torch.arange(4, dtype=torch.bfloat16).reshape(1, 4)
+    raw_encoder = _Encoder([first, second])
+    encoder = InlineEncoder(raw_encoder, image_token_id=99)
+    raw_client = _Client(
+        [
+            {"token_ids": [7], "index": 0},
+            {"token_ids": [8], "index": 0, "finish_reason": "stop"},
+        ]
+    )
+    context = object()
+    request = _request()
+    orchestrator = ExternalEncoderOrchestrator(
+        encoder,
+        LLMUnaryClient(raw_client),
+        "decoder-model",
+    )
+
+    completion = await orchestrator(request, context=context)
+
+    assert completion["token_ids"] == [7, 8]
+    assert completion["finish_reason"] == "stop"
+    assert raw_encoder.raws == [
+        "https://example.com/one.png",
+        "https://example.com/two.png",
+    ]
+    assert raw_client.context is context
+    assert raw_client.request is not None
+    assert raw_client.request["model"] == "decoder-model"
+    assert "multi_modal_data" not in raw_client.request
+    assert raw_client.request["extra_args"] == {"keep": 1}
+    assert "encoder_result" not in request
+
+    result = ExternalEncoderResult.from_dict(raw_client.request["encoder_result"])
+    assert result.row_splits == (0, 2, 3)
+    torch.testing.assert_close(
+        decode_request_plane_tensor(result.features),
+        torch.cat([first, second]),
+    )
+
+
+async def test_inline_encoder_rejects_mismatched_artifacts() -> None:
+    encoder = InlineEncoder(
+        _Encoder([torch.ones((1, 4)), torch.ones((1, 5))]),
+        image_token_id=99,
+    )
+
+    with pytest.raises(InvalidArgument, match="one hidden size"):
+        await encoder.encode(_request())
+
+
+async def test_inline_encoder_requires_images() -> None:
+    encoder = InlineEncoder(_Encoder([]), image_token_id=99)
+
+    with pytest.raises(InvalidArgument, match="at least one image"):
+        await encoder.encode({"multi_modal_data": {}})
+
+
+def test_inline_encoder_closes_driver() -> None:
+    raw_encoder = _Encoder([])
+    encoder = InlineEncoder(raw_encoder, image_token_id=99)
+
+    encoder.close()
+
+    assert raw_encoder.closed

examples/custom_encoder/remote/README.md

@@ -0,0 +1,60 @@
+# Remote Custom Encoder
+
+This deployment keeps an application-owned custom encoder inside a bespoke orchestrator worker and calls a stock aggregated `dynamo.vllm` worker for generation.
+
+```text
+OpenAI client
+     │
+     ▼
+generic Dynamo frontend
+     │
+     ▼
+bespoke orchestrator worker
+     ├── inline VisionEncoderBackend
+     ├── CPU/MsgPack encoder_result
+     └── LLMUnaryClient
+                 │
+                 ▼
+       stock aggregated dynamo.vllm
+```
+
+The encoder remains transport-agnostic. The orchestrator concatenates its ordered CPU tensors, calls `encode_request_plane_tensor`, adds the versioned result to `GenerateRequest.encoder_result`, and removes the raw media fields before invoking vLLM. The remote vLLM endpoint streams token chunks internally; `LLMUnaryClient.complete()` folds them into the single terminal result returned by the orchestrator.
+
+The initial handoff supports contiguous two-dimensional CPU `bfloat16`, `float16`, and `float32` linear embeddings. It requires a text-only aggregated decoder configured with `--enable-prompt-embeds`. The payload travels through the ordinary MsgPack request plane; it does not use NIXL.
+
+## Run
+
+From the repository root:
+
+```bash
+./examples/custom_encoder/remote/launch.sh
+```
+
+The default `HitchhikersVisionEncoder` ignores the image contents and substitutes the embeddings for a known phrase, making the prompt-splicing path easy to inspect. Replace it with another `VisionEncoderBackend` using:
+
+```bash
+DYN_ENCODER_CLASS=your_package.YourVisionEncoder \
+./examples/custom_encoder/remote/launch.sh
+```
+
+Send an OpenAI-compatible request to the public orchestrator model:
+
+```bash
+curl http://localhost:8000/v1/chat/completions \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "model": "remote-custom-encoder",
+    "messages": [{
+      "role": "user",
+      "content": [
+        {"type": "text", "text": "Based on The Hitchhiker’s Guide to the Galaxy, The Answer to"},
+        {"type": "image_url", "image_url": {"url": "https://example.com/ignored.png"}},
+        {"type": "text", "text": " is?"}
+      ]
+    }],
+    "max_tokens": 24,
+    "temperature": 0
+  }'
+```
+
+The default backend is expected to steer the model toward `42`; it is a semantic smoke test rather than a real image encoder.

examples/custom_encoder/remote/__init__.py

@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Application-owned remote-decoder custom encoder."""

examples/custom_encoder/remote/launch.sh

@@ -0,0 +1,55 @@
+#!/bin/bash
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(readlink -f "$SCRIPT_DIR/../../..")"
+source "$REPO_ROOT/examples/common/gpu_utils.sh"
+source "$REPO_ROOT/examples/common/launch_utils.sh"
+trap dynamo_exit_trap EXIT
+
+MODEL="${DYN_MODEL:-Qwen/Qwen2.5-1.5B-Instruct}"
+PUBLIC_MODEL_NAME="${DYN_SERVED_MODEL_NAME:-remote-custom-encoder}"
+DECODER_MODEL_NAME="${DYN_DECODER_MODEL_NAME:-remote-custom-encoder-decoder}"
+NAMESPACE="${DYN_NAMESPACE:-remote-custom-encoder}"
+GENERATOR_ENDPOINT="$NAMESPACE.generator.generate"
+HTTP_PORT="${DYN_HTTP_PORT:-8000}"
+DECODER_GPU="${DYN_DECODER_GPU:-${CUDA_VISIBLE_DEVICES:-0}}"
+ENCODER_GPU="${DYN_ENCODER_GPU:-$DECODER_GPU}"
+MAX_MODEL_LEN="${DYN_MAX_MODEL_LEN:-4096}"
+GPU_MEM_ARGS=$(build_vllm_gpu_mem_args)
+[[ -z "$GPU_MEM_ARGS" ]] && GPU_MEM_ARGS="--gpu-memory-utilization 0.8"
+
+export DYN_MODEL="$MODEL"
+export DYN_SERVED_MODEL_NAME="$PUBLIC_MODEL_NAME"
+export DYN_DECODER_MODEL_NAME="$DECODER_MODEL_NAME"
+export DYN_NAMESPACE="$NAMESPACE"
+export DYN_REQUEST_PLANE=tcp
+export DYN_REQUEST_PLANE_CODEC=msgpack
+export DYN_TCP_MAX_MESSAGE_SIZE=209715200
+export DYN_HTTP_BODY_LIMIT_MB=200
+
+print_launch_banner --no-curl "Remote Custom Encoder" "$MODEL" "$HTTP_PORT" \
+    "Inline encoder: ${DYN_ENCODER_CLASS:-HitchhikersVisionEncoder}" \
+    "Remote decoder: dyn://$GENERATOR_ENDPOINT"
+
+python -m dynamo.frontend --http-port "$HTTP_PORT" &
+
+CUDA_VISIBLE_DEVICES="$DECODER_GPU" \
+DYN_SYSTEM_PORT="${DYN_GENERATOR_SYSTEM_PORT:-8081}" \
+python -m dynamo.vllm \
+    --model "$MODEL" \
+    --served-model-name "$DECODER_MODEL_NAME" \
+    --endpoint "dyn://$GENERATOR_ENDPOINT" \
+    --enable-prompt-embeds \
+    --max-model-len "$MAX_MODEL_LEN" \
+    $GPU_MEM_ARGS \
+    "$@" &
+
+CUDA_VISIBLE_DEVICES="$ENCODER_GPU" \
+DYN_SYSTEM_PORT="${DYN_ORCHESTRATOR_SYSTEM_PORT:-8082}" \
+python -m examples.custom_encoder.remote.orchestrator_worker &
+
+wait_any_exit