dynamo.common

Abstract base class for a receiver of precomputed embeddings from the encode worker.

1
from dynamo.common.multimodal import AbstractEmbeddingReceiver

components/src/dynamo/common/multimodal/embedding_transfer.py#L83

Public methods

receive_embeddings

1
receive_embeddings(request: TransferRequest) -> tuple[int, torch.Tensor]

Abstract method to receive precomputed embeddings for a given request ID.

Parameters

Returns

• int — A tuple containing the tensor ID and the received embeddings as a torch.Tensor.
• torch.Tensor — Caller should invoke release_tensor(tensor_id) when the tensor is no longer needed to free up resources.

source

release_tensor

1
release_tensor(tensor_id: int) -> None

Abstract method to indicate that the tensor associated with the ID is no longer in use. Args: tensor_id: The ID of the tensor to release.

source

Abstract base class for a sender of precomputed embeddings to the downstream worker.

1
from dynamo.common.multimodal import AbstractEmbeddingSender

components/src/dynamo/common/multimodal/embedding_transfer.py#L114

Public methods

send_embeddings

1
send_embeddings(embeddings: torch.Tensor, stage_embeddings: bool = False) -> tuple[TransferRequest, Awaitable[None]]

Abstract method to send precomputed embeddings for a given request ID.

Parameters

Returns: A tuple containing the TransferRequest object and an awaitable that can be awaited to indicate the send is completed.

source

aiohttp-backed concrete client.

1
from dynamo.common.http import AiohttpClient

1
AiohttpClient(config = None) -> None

components/src/dynamo/common/http/aiohttp_client.py#L28

Public methods

init

1
__init__(config = None) -> None

No summary available.

source

close

1
close() -> None

No summary available.

source

Async wrapper with request coalescing over MultimodalEmbeddingCacheManager.

1
from dynamo.common.multimodal import AsyncEncoderCache

1
AsyncEncoderCache(cache: MultimodalEmbeddingCacheManager)

Provides async get_or_compute that deduplicates concurrent requests for the same key, ensuring only one encoding runs at a time per key.

Thread Safety: This class is NOT thread-safe. It is designed to run within a single asyncio event loop. All access must be from the same thread.

components/src/dynamo/common/multimodal/async_encoder_cache.py#L45

Public methods

init

1
__init__(cache: MultimodalEmbeddingCacheManager)

Initialize the async encoder cache.

Parameters

source

get

1
get(key: str) -> Optional[CachedEmbedding]

Synchronous get from underlying cache.

Parameters

Returns

• Optional[CachedEmbedding] — Cached embedding or None if not found.

source

get_or_compute

1
get_or_compute(key: str, compute_fn: Callable[[], Awaitable[CachedEmbedding]]) -> CachedEmbedding

Get from cache or compute with request coalescing.

If the key is in cache, returns immediately. If another coroutine is already computing this key, waits for that result. Otherwise, computes and caches the result.

Parameters

Returns

• CachedEmbedding — The cached or computed embedding.

Raises

• Exception — Re-raises any exception from compute_fn.

source

Async audio loader for multimodal pipelines.

1
from dynamo.common.multimodal import AudioLoader

1
AudioLoader(http_timeout: float = 30.0, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None) -> None

Delegates URL fetching and decoding to vLLM’s MediaConnector + AudioMediaIO so that the exact same loading logic runs whether the request arrives via vllm serve or through Dynamo. Returns (waveform, sample_rate) tuples at the native sample rate — vLLM’s model-specific MultiModalDataParser handles resampling and channel normalization downstream.

Also supports the NIXL decoded variant for frontend-decoded audio transferred via RDMA.

components/src/dynamo/common/multimodal/audio_loader.py#L74

Public methods

init

1
__init__(http_timeout: float = 30.0, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None) -> None

No summary available.

source

load_audio

1
load_audio(audio_url: str) -> tuple[np.ndarray, float]

Load audio from a URL and return a (waveform, sample_rate) tuple.

Supports http(s), data: URIs, file:// paths, and bare filesystem paths. Audio is loaded at the native sample rate — no resampling is performed.

source

load_audio_batch

1
load_audio_batch(audio_mm_items: List[Dict[str, Any]]) -> List[tuple[np.ndarray, float]]

Load a batch of audio files from multimodal data items.

Supports two paths:

1. Url variant: Download and decode audio via vLLM’s MediaConnector
2. Decoded variant: Read pre-decoded audio via NIXL RDMA (requires enable_frontend_decoding=True)

Returns

• List[tuple[np.ndarray, float]] — List of (waveform, sample_rate) tuples.

source

Abstract base for all engines — the modality-agnostic lifecycle.

1
from dynamo.common.backend import BaseEngine

Worker drives every engine through the same lifecycle regardless of modality; only the request/response shape of generate differs. That method is therefore declared on the modality-specific subclasses (LLMEngine for token-based inference, RawEngine for raw non-token media generation), not here.

Lifecycle:

1. from_args(argv) — parse CLI args, return (engine, WorkerConfig)
2. start() — start the engine, return EngineConfig metadata. After start() returns, generate() MUST be ready to accept calls. Worker begins serving immediately after start().
3. generate() — called for each request (concurrent calls expected)
4. abort() — called when a request is cancelled (optional, default no-op)
5. cleanup() — called once on shutdown, release all resources

components/src/dynamo/common/backend/engine.py#L154

Public methods

from_args

1
from_args(cls, argv: list[str] | None = None) -> tuple[BaseEngine, WorkerConfig]

Parse CLI args and construct the engine (not yet started).

Parameters

Returns

• tuple[BaseEngine, WorkerConfig] — A (engine, worker_config) pair.

source

start

1
start(worker_id: int) -> EngineConfig

Start the engine and return registration metadata.

After this returns the engine MUST be ready to accept generate() calls. Worker will register the model and begin serving immediately.

worker_id is an opaque, runtime-allocated unique identifier for this worker. It is stable from start() onward for the worker’s lifetime and unique across replicas in the cluster. Engines that need a per-worker key for cluster-wide bookkeeping should derive it from this value rather than hashing host/pid or asking operators for a CLI override. The internal mechanism (discovery instance ID) is not part of the contract — engines should treat it as opaque.

source

abort

1
abort(context: Context) -> None

Abort an in-flight request (optional, default no-op).

Called by Worker when the client disconnects or the request is cancelled. Override to release engine resources (KV cache, scheduler slots, etc.).

context.metadata in this callback reflects the original propagated request metadata snapshot. Mutations made to context.metadata during generate are not visible here.

source

is_quiescent

1
is_quiescent() -> Optional[bool]

Whether in-flight KV transfers are done, so cleanup may release GPU memory. The Rust Worker polls this on prefill workers between the grace period and cleanup:

• True — quiescent; exit the drain loop now.
• False — busy; poll again next tick.
• None — no introspection (default); poll until the drain budget (DYN_PREFILL_DRAIN_TIMEOUT_S) expires. Never frees KV early.

Aggregated/decode workers are never polled. Override only if the engine can observe transfer completion.

source

cleanup

1
cleanup() -> None

Release all engine resources.

Worker guarantees:

• cleanup() runs after a successful start() on shutdown — the common case.
• cleanup() also runs after start() raised, on the partial state the engine may have allocated before failing (inner LLM handle, sockets, background tasks). Implementations must be null-safe: guard each resource with an is None check so a partially constructed engine can be released without raising.
• cleanup() is not called when start() was never invoked (e.g. pre-start shutdown). Engines whose constructors allocate resources should release them via __del__ / context-manager semantics rather than rely on cleanup().

cleanup() is never invoked concurrently with start() or another cleanup() — Worker’s state machine serializes those transitions. The conformance kit asserts that a second cleanup() call after a successful first is a safe no-op.

source

register_prometheus

1
register_prometheus(metrics: 'EngineMetrics') -> None

Bridge a vendor-prefixed Prometheus registry into the runtime’s /metrics output via metrics.add_expfmt_callback. Default no-op. See dynamo.common.backend.metrics for helpers. Do not retain metrics past return.

Framework-owned lifecycle + per-rank gauges (dynamo_component_{cleanup_time_seconds,drain_time_seconds,model_load_time_seconds,total_blocks,gpu_cache_usage_percent,kv_cache_hit_rate}) are owned and registered by the framework Rust-side — they do NOT require the engine to implement this method.

source

component_metrics_dp_ranks

1
component_metrics_dp_ranks() -> list[int]

Declare the data-parallel ranks this engine publishes per-rank snapshots for. Empty (default) opts out.

Stable for the engine’s lifetime. Worker constructs a SnapshotPublisher sized to these ranks and hands it back via attach_snapshot_publisher. The engine then calls publisher.publish(rank, snap) from its stat-logger thread — event-driven, no polling.

ComponentSnapshot.kv_cache_hit_rate is tri-state: None means “no data yet” or “no prefix cache” (gauge skipped), 0.0 is a legitimate measurement (zero hits).

source

attach_snapshot_publisher

1
attach_snapshot_publisher(publisher: Any) -> None

Framework hands the engine the Rust-owned SnapshotPublisher once, after setup_metrics constructed it from component_metrics_dp_ranks. Stash the reference; call publisher.publish(rank, snap) from your stat-logger thereafter.

Only invoked when component_metrics_dp_ranks returns non-empty. Default is no-op so engines that opt out don’t need to override.

source

health_check_payload

1
health_check_payload() -> Optional[dict[str, Any]]

Canary payload the runtime sends through generate when the endpoint is idle. Return None (default) to disable active probing. Worker calls this once after start and resolves DYN_HEALTH_CHECK_PAYLOAD / --health-check-payload overrides on top.

source

supported_controls

1
supported_controls() -> set[str]

Return the set of engine-control capability keys this engine supports.

Controls are semantic operations on the engine’s serving lifecycle. Engines advertise the keys they implement.

source

engine_control

1
engine_control(control: str, body: dict[str, Any]) -> dict[str, Any]

Handle one advertised engine-control request.

source

supported_updates

1
supported_updates() -> set[str]

Return the set of engine-update capability keys this engine supports.

Updates are a sibling surface to supported_controls for operations that mutate engine-managed assets rather than the engine’s serving lifecycle. Engines advertise the keys they implement.

source

engine_update

1
engine_update(update: str, body: dict[str, Any]) -> dict[str, Any]

Handle one advertised engine-update request.

source

on_endpoint_ready

1
on_endpoint_ready(endpoint) -> None

Receive the runtime serving Endpoint once, before serving begins.

Default no-op. Engines that publish their own discovery records stash it for use from engine_update. Worker calls this exactly once; a raised exception is fatal to startup.

source

A RawEngine for diffusion-family generation (image/video via VisualGen, DiffGenerator). Names the family only — non-diffusion raw modalities (e.g. TTS audio) subclass RawEngine directly. Routing keys off RawEngine, so any subclass uses the raw adapter.

1
from dynamo.common.backend import DiffusionEngine

components/src/dynamo/common/backend/engine.py#L427

Disaggregation mode for LLM workers.

1
from dynamo.common.constants import DisaggregationMode

components/src/dynamo/common/constants.py#L15

Embedding transfer mode for LLM workers.

1
from dynamo.common.constants import EmbeddingTransferMode

components/src/dynamo/common/constants.py#L24

Registration metadata returned by an engine’s start.

1
from dynamo.common.backend import EngineConfig

1
EngineConfig(model: str, served_model_name: Optional[str] = None, runtime_data: Optional[dict[str, Any]] = None, llm: Optional[LlmRegistration] = None, model_aliases: list[str] = list()) -> None

The neutral fields (model, served_model_name, model_aliases, runtime_data) apply to every modality; token-pipeline metadata lives in the optional llm sub-record, which raw media engines leave None.

components/src/dynamo/common/backend/engine.py#L135

Public methods

init

1
__init__(model: str, served_model_name: Optional[str] = None, runtime_data: Optional[dict[str, Any]] = None, llm: Optional[LlmRegistration] = None, model_aliases: list[str] = list()) -> None

No summary available.

source

No summary available.

1
from dynamo.common.engine_monitor import EngineHealthMonitorConfig

1
EngineHealthMonitorConfig(interval: float, check_timeout: float, shutdown_timeout: float) -> None

components/src/dynamo/common/engine_monitor.py#L40

Public methods

from_env

1
from_env(cls, *, interval: Optional[float] = None, check_timeout: Optional[float] = None, shutdown_timeout: Optional[float] = None) -> 'EngineHealthMonitorConfig'

No summary available.

source

init

1
__init__(interval: float, check_timeout: float, shutdown_timeout: float) -> None

No summary available.

source

Per-iteration metrics emitted by InstrumentedScheduler.

1
from dynamo.common.forward_pass_metrics import ForwardPassMetrics

One message is emitted per scheduler iteration (one per forward pass). An idle heartbeat (all zeros, wall_time=0) is emitted once when the engine transitions from active to idle.

components/src/dynamo/common/forward_pass_metrics.py#L153

Single chunk yielded by LLMEngine.generate().

1
from dynamo.common.backend import GenerateChunk

Every chunk must include token_ids and index. Use index=0 for single-choice responses. The final chunk must additionally include finish_reason; completion_usage is optional (the OpenAI frontend aggregates it when present, and matches the Rust Option<CompletionUsage> / skip_serializing_if = "Option::is_none" semantics).

Prefill terminals carry disaggregated_params for the PrefillRouter to forward to the decode peer. When the caller requested logprobs, chunks may also carry log_probs and top_logprobs aligned to token_ids — see dynamo.common.backend.logprobs.

Encode terminals carry encoder_result (an opaque object the frontend forwards onto the downstream PreprocessedRequest.encoder_result). Construct with dynamo.common.backend.multimodal.encoder_terminal_chunk.

components/src/dynamo/common/backend/engine.py#L74

Inbound request dict passed to LLMEngine.generate().

1
from dynamo.common.backend import GenerateRequest

token_ids is always present (set by the Rust preprocessor). The remaining groups are optional — engines should access them defensively with .get(key, {}).

Disaggregated-serving keys (prefill_result, bootstrap_info) are set by the frontend’s PrefillRouter on decode requests; engines read them via dynamo.common.backend.disagg helpers.

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 dynamo.common.backend.multimodal.require_encoder_result. 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 request to a loaded adapter.

components/src/dynamo/common/backend/engine.py#L34

Backend-neutral HTTP client.

1
from dynamo.common.http import HttpClient

1
HttpClient(config: Optional[HttpConfigBase] = None) -> None

Subclasses own a backend-specific session/client singleton on the instance. Callers reach the public surface via fetch_bytes and the unified exception classes above; the concrete backend type is invisible past instantiation.

components/src/dynamo/common/http/base.py#L50

Public methods

init

1
__init__(config: Optional[HttpConfigBase] = None) -> None

No summary available.

source

fetch_bytes

1
fetch_bytes(url: str, timeout: float, *, policy: Optional[UrlValidationPolicy] = None) -> bytes

Fetch url and return the response body.

Single-shot: no retries. Raises one of the unified exception classes above; callers never see native httpx/aiohttp classes.

policy=None: use the backend’s built-in redirect handling.

policy set: follow redirects manually and revalidate each hop against the policy via url_validator.validate_url. This is the SSRF-safe path; raises UrlValidationError if any hop fails or the chain exceeds _MAX_REDIRECTS.

source

close

1
close() -> None

Close the backend session/client. Idempotent.

source

Network-layer failure: DNS, refused, reset, half-close.

1
from dynamo.common.http import HttpConnectionError

components/src/dynamo/common/http/base.py#L36

Base class for all HTTP fetch failures.

1
from dynamo.common.http import HttpError

components/src/dynamo/common/http/base.py#L28

Server responded with a non-2xx status.

1
from dynamo.common.http import HttpStatusError

1
HttpStatusError(status: int, message: str, url: str) -> None

components/src/dynamo/common/http/base.py#L40

Public methods

init

1
__init__(status: int, message: str, url: str) -> None

No summary available.

source

Timeout during connect / read / pool-wait.

1
from dynamo.common.http import HttpTimeoutError

components/src/dynamo/common/http/base.py#L32

httpx-backed concrete client.

1
from dynamo.common.http import HttpxClient

1
HttpxClient(config = None) -> None

components/src/dynamo/common/http/httpx_client.py#L36

Public methods

init

1
__init__(config = None) -> None

No summary available.

source

close

1
close() -> None

No summary available.

source

No summary available.

1
from dynamo.common.multimodal import ImageLoader

1
ImageLoader(cache_size: int = CACHE_SIZE_MAXIMUM, http_timeout: float = 30.0, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None)

components/src/dynamo/common/multimodal/image_loader.py#L59

Public methods

init

1
__init__(cache_size: int = CACHE_SIZE_MAXIMUM, http_timeout: float = 30.0, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None)

Initialize the ImageLoader with caching, HTTP settings, and optional NIXL config for receiving frontend decoding.

Parameters

source

load_image

1
load_image(image_url: str) -> Image.Image

No summary available.

source

load_image_batch

1
load_image_batch(image_mm_items: List[Dict[str, Any]], *, preserve_uuid_slots: bool = False) -> list[Any]

Load a batch of images from multimodal data items.

Supports three paths:

1. Url variant: Download and decode image from URL (default)
2. Decoded variant: Read pre-decoded image via NIXL RDMA (requires enable_frontend_decoding=True)
3. UuidOnly variant: Preserve an aligned empty slot for backend cache lookup when preserve_uuid_slots=True

Parameters

Returns

• list[Any] — Loaded images, with None for UUID-only cache slots

Raises

• HttpStatusError — If any image fails with an HTTP status error (e.g. 415 Unsupported Media Type); the status is preserved so the frontend returns the correct client-error code instead of 500.
• UrlValidationError — If a media URL is rejected by the SSRF policy; preserved as a ValueError so the frontend returns a 4xx, not 500.
• Exception — If any image fails to load for any other reason
• ValueError — If enable_frontend_decoding=True but nixl_connector is None
• ValueError — If a UUID-only slot is received without opting in

source

Abstract base for token-based inference engines.

1
from dynamo.common.backend import LLMEngine

The token pipeline: the Rust preprocessor tokenizes the prompt and sets token_ids on the request; generate yields token chunks that the Rust postprocessor detokenizes. Registered with ModelInput.Tokens and served through the token request adapter.

components/src/dynamo/common/backend/engine.py#L347

Public methods

generate

1
generate(request: GenerateRequest, context: Context) -> AsyncGenerator[GenerateChunk, None]

Yield streaming response chunks for a single request.

Called concurrently for multiple in-flight requests.

Each chunk: {"token_ids": [...], "index": 0} Final chunk must include: {"token_ids": [...], "index": 0, "finish_reason": "...", "completion_usage": {...}}

source

kv_event_sources

1
kv_event_sources() -> list[KvEventSource]

KV event sources, one per data-parallel rank. Default opts out of KV-aware routing. Worker calls once after start.

source

logits_processor_spec

1
logits_processor_spec() -> 'LogitsProcessorSpec | None'

Return backend-neutral logits-processor activation data.

The default opts out. An engine that overrides this method resolves and caches the specification during startup, then passes it to logits_processors_for_request from generate. The engine integration remains responsible for realizing each entry into its inference library’s processor type.

source

Token-pipeline registration metadata (KV cache, data-parallel layout, disaggregation bootstrap). Set by LLMEngines; RawEngines leave EngineConfig.llm None. A None field isn’t advertised (the router falls back to its defaults).

1
from dynamo.common.backend import LlmRegistration

1
LlmRegistration(context_length: Optional[int] = None, kv_cache_block_size: Optional[int] = None, total_kv_blocks: Optional[int] = None, max_num_seqs: Optional[int] = None, max_num_batched_tokens: Optional[int] = None, data_parallel_size: Optional[int] = None, data_parallel_start_rank: Optional[int] = None, bootstrap_host: Optional[str] = None, bootstrap_port: Optional[int] = None) -> None

components/src/dynamo/common/backend/engine.py#L109

Public methods

init

1
__init__(context_length: Optional[int] = None, kv_cache_block_size: Optional[int] = None, total_kv_blocks: Optional[int] = None, max_num_seqs: Optional[int] = None, max_num_batched_tokens: Optional[int] = None, data_parallel_size: Optional[int] = None, data_parallel_start_rank: Optional[int] = None, bootstrap_host: Optional[str] = None, bootstrap_port: Optional[int] = None) -> None

No summary available.

source

Metadata for a loaded LoRA adapter.

1
from dynamo.common.lora import LoRAInfo

1
LoRAInfo(id: int, path: str) -> None

components/src/dynamo/common/lora/manager.py#L123

Public methods

init

1
__init__(id: int, path: str) -> None

No summary available.

source

Minimal Python wrapper around Rust core with extension points.

1
from dynamo.common.lora import LoRAManager

1
LoRAManager(cache_path: Optional[Path] = None)

The manager uses the Rust-based LoRADownloader for local, S3, and Hugging Face sources, and allows registering custom Python sources for other protocols.

components/src/dynamo/common/lora/manager.py#L36

Public methods

init

1
__init__(cache_path: Optional[Path] = None)

Initialize LoRA manager.

Parameters

source

register_custom_source

1
register_custom_source(scheme: str, source: LoRASourceProtocol) -> None

Register a custom Python source for a URI scheme.

Parameters

source

download_lora

1
download_lora(lora_uri: str) -> Dict[str, Any]

Download LoRA if needed, return local path.

The source is inferred from the URI scheme:

• file:// -> Local filesystem (Rust)
• s3:// -> S3 (Rust)
• hf:// -> Hugging Face Hub (Rust)
• Custom schemes -> Registered Python sources

Parameters

Returns

• Dict[str, Any] — Dictionary with: - status: “success” or “error” - local_path: Local path to LoRA (if successful) - message: Error message (if error)

source

is_cached

1
is_cached(lora_uri: str) -> bool

Check if LoRA is already cached locally.

source

Protocol for custom Python LoRA sources. Users can implement this to add custom sources.

1
from dynamo.common.lora import LoRASourceProtocol

components/src/dynamo/common/lora/manager.py#L21

Public methods

download

1
download(lora_uri: str, dest_path: Path) -> Path

Download LoRA to dest_path, return actual path

source

exists

1
exists(lora_uri: str) -> bool

Check if LoRA exists in this source

source

Receiver that reads embeddings from a local file path provided in the serialized request.

1
from dynamo.common.multimodal import LocalEmbeddingReceiver

1
LocalEmbeddingReceiver()

components/src/dynamo/common/multimodal/embedding_transfer.py#L203

Public methods

init

1
__init__()

No summary available.

source

receive_embeddings

1
receive_embeddings(request: TransferRequest) -> tuple[int, torch.Tensor]

Receive precomputed embeddings for a given request ID.

Parameters

Returns

• int — A tuple containing the tensor ID and the received embeddings as a torch.Tensor.
• torch.Tensor — Caller should invoke release_tensor(tensor_id) when the tensor is no longer needed to free up resources.

source

release_tensor

1
release_tensor(tensor_id: int) -> None

Indicate that the tensor associated with the ID is no longer in use.

Parameters

source

Sender that saves embeddings to a local file and sends the file path as the serialized request.

1
from dynamo.common.multimodal import LocalEmbeddingSender

1
LocalEmbeddingSender()

components/src/dynamo/common/multimodal/embedding_transfer.py#L136

Public methods

init

1
__init__()

No summary available.

source

save_embeddings_to_file

1
save_embeddings_to_file(embedding_key: str, embeddings: torch.Tensor) -> str

Save the embeddings to a local file and return the file path.

Parameters

Returns: The file path where the embeddings are saved.

source

send_embeddings

1
send_embeddings(embeddings: torch.Tensor, stage_embeddings: bool = False) -> tuple[TransferRequest, Awaitable[None]]

Send precomputed embeddings for a given request ID.

Parameters

Returns: A tuple containing the TransferRequest object and an awaitable that can be awaited to indicate the send is completed.

source

No summary available.

1
from dynamo.common.metadata_upload import MetadataUploader

1
MetadataUploader(url: str) -> None

components/src/dynamo/common/metadata_upload.py#L141

Public methods

from_settings

1
from_settings(cls, settings: dict[str, Any] | None) -> MetadataUploader | None

No summary available.

source

from_backend_request

1
from_backend_request(cls, request: dict[str, Any]) -> MetadataUploader | None

No summary available.

source

upload_choice

1
upload_choice(choice_index: int, metadata: dict[str, Any]) -> None

No summary available.

source

init

1
__init__(url: str) -> None

No summary available.

source

LRU cache for encoder embeddings.

1
from dynamo.common.memory import MultimodalEmbeddingCacheManager

1
MultimodalEmbeddingCacheManager(capacity_bytes: int)

Stores tensors keyed by content hash with automatic eviction when capacity is exceeded.

Thread Safety: This class is NOT thread-safe. It is designed to run within a single thread (e.g., an asyncio event loop). All access must be from the same thread to avoid race conditions. This is intentional to keep the implementation simple and avoid locking overhead.

components/src/dynamo/common/memory/multimodal_embedding_cache_manager.py#L43

Public methods

init

1
__init__(capacity_bytes: int)

Initialize the encoder cache.

Parameters

source

get

1
get(key: str) -> Optional[CachedEmbedding]

Get a cached embedding from the cache.

If found, the entry is moved to the end (most recently used).

Parameters

Returns

• Optional[CachedEmbedding] — The cached embedding, or None if not found.

source

keys

1
keys() -> list[str]

Return the current cache keys in LRU order.

source

set

1
set(key: str, entry: CachedEmbedding) -> bool

No summary available.

source

set_with_delta

1
set_with_delta(key: str, entry: CachedEmbedding) -> CacheMutation

Store a cached embedding in the cache.

If the key already exists, the old value is replaced. If adding the entry would exceed capacity, LRU entries are evicted. If the tensor itself is larger than capacity, it is not stored.

Parameters

Returns

• CacheMutation — CacheMutation describing whether the entry was stored plus the
• CacheMutation — authoritative add/remove delta caused by this mutation.

source

NIXL READ based embedding transfer receiver.

1
from dynamo.common.multimodal import NixlReadEmbeddingReceiver

1
NixlReadEmbeddingReceiver(embedding_hidden_size: int = 8 * 1024, max_item_mm_token: int = 1024, max_items: int = 1024) -> None

Uses nixl_connect.Connector which now natively provides a shared singleton Connection (NIXL agent) and reference-counted Remote agent lifecycle.

components/src/dynamo/common/multimodal/embedding_transfer.py#L852

Public methods

init

1
__init__(embedding_hidden_size: int = 8 * 1024, max_item_mm_token: int = 1024, max_items: int = 1024) -> None

No summary available.

source

receive_embeddings

1
receive_embeddings(request: TransferRequest) -> tuple[int, torch.Tensor]

Receive precomputed embeddings for a given request ID.

Parameters

Returns

• int — A tuple containing the tensor ID and the received embeddings as a torch.Tensor.
• torch.Tensor — Caller should invoke release_tensor(tensor_id) when the tensor is no longer needed to free up resources.

source

release_tensor

1
release_tensor(tensor_id: int) -> None

Indicate that the tensor associated with the ID is no longer in use.

Parameters

source

NIXL READ based embedding transfer sender.

1
from dynamo.common.multimodal import NixlReadEmbeddingSender

1
NixlReadEmbeddingSender()

Uses nixl_connect.Connector which now natively provides a shared singleton Connection (NIXL agent) and reference-counted Remote agent lifecycle.

components/src/dynamo/common/multimodal/embedding_transfer.py#L795

Public methods

init

1
__init__()

No summary available.

source

send_embeddings

1
send_embeddings(embeddings: torch.Tensor, stage_embeddings: bool = False) -> tuple[TransferRequest, Awaitable[None]]

Send precomputed embeddings.

Parameters

Returns: A tuple containing the TransferRequest object and an awaitable that can be awaited to indicate the send is completed.

source

Counter part of ‘NixlWriteEmbeddingSender’, see ‘NixlWriteEmbeddingSender’ for details. The receiver manages a ring buffer for sender to write the embeddings into, and respond to the sender’s transfer request with the buffer information for the WRITE transfer.

1
from dynamo.common.multimodal import NixlWriteEmbeddingReceiver

1
NixlWriteEmbeddingReceiver(buffer_size = 2 * 8 * 1024 * 1024 * 256 * 2)

components/src/dynamo/common/multimodal/embedding_transfer.py#L635

Public methods

init

1
__init__(buffer_size = 2 * 8 * 1024 * 1024 * 256 * 2)

No summary available.

source

receive_embeddings

1
receive_embeddings(request: TransferRequest, receive_timeout = 60) -> tuple[int, torch.Tensor]

Receive precomputed embeddings for a given request ID.

Parameters

Returns

• int — A tuple containing the tensor ID and the received embeddings as a torch.Tensor.
• torch.Tensor — Caller should invoke release_tensor(tensor_id) when the tensor is no longer needed to free up resources.

source

release_tensor

1
release_tensor(tensor_id: int) -> None

Indicate that the tensor associated with the ID is no longer in use.

Parameters

source

NIXL WRITE-based implementation of the embedding sender interface.

1
from dynamo.common.multimodal import NixlWriteEmbeddingSender

1
NixlWriteEmbeddingSender()

Designed for scenarios where the sender transmits dynamically allocated tensors. Because these tensors allocation is external to the sender, NIXL memory registration will perform on each send request. The receiver will manage a pre-allocated buffer, so its NIXL metadata is consistent once initialized. In such acenarios, let sender initiate the WRITE operations requires minimal metadata exchange.

Protocol:

1. Record the receiver NIXL metadata, this is done:

• Implicitly through the first transfer request as fallback if the metadata hasn’t been recorded.
• [REMOVED] Explicitly through add_agent() API before calling send_embeddings(). The receiver provides get_agent_metadata() API to return its NIXL metadata. This complicates the implementation and add extra responsiblity on the caller side, will revisit the necessity if metadata exchange overhead is significant.

2. The sender prepares the embeddings and produces a TransferRequest containing sender contact and tensor metadata (shape, dtype, size, etc).
3. The receiver responds with (optional) receiver contact, target tensor metadata (buffer address, device, etc) and done signal through NIXL notification.
4. The sender performs a NIXL WRITE to push the data into the receiver’s buffer.

components/src/dynamo/common/multimodal/embedding_transfer.py#L376

Public methods

init

1
__init__()

No summary available.

source

send_embeddings

1
send_embeddings(embeddings: torch.Tensor, stage_embeddings: bool = False) -> tuple[TransferRequest, asyncio.Future]

Send precomputed embeddings.

Parameters

Returns: A tuple containing the TransferRequest object and an awaitable that can be awaited to indicate the send is completed.

source

Request for video generation (/v1/videos endpoint).

1
from dynamo.common.protocols import NvCreateVideoRequest

Matches Rust NvCreateVideoRequest in lib/llm/src/protocols/openai/videos.rs.

components/src/dynamo/common/protocols/video_protocol.py#L50

Response structure for video generation.

1
from dynamo.common.protocols import NvVideosResponse

Matches Rust NvVideosResponse in lib/llm/src/protocols/openai/videos.rs.

components/src/dynamo/common/protocols/video_protocol.py#L107

No summary available.

1
from dynamo.common.lora import OnceLock

1
OnceLock() -> None

components/src/dynamo/common/lora/once.py#L15

Public methods

init

1
__init__() -> None

No summary available.

source

get_or_init

1
get_or_init(initializer: Callable[[], T]) -> T

No summary available.

source

get

1
get() -> T | None

No summary available.

source

Metrics for requests waiting in the queue (not scheduled this iteration).

1
from dynamo.common.forward_pass_metrics import QueuedRequestMetrics

All token counts here are raw totals — prefix cache effects are unknown until a request is actually scheduled.

components/src/dynamo/common/forward_pass_metrics.py#L122

Validation error whose message can be returned directly to RL clients.

1
from dynamo.common.rl import RLAdminValidationError

components/src/dynamo/common/rl/admin.py#L21

Registry for worker RL admin route descriptors.

1
from dynamo.common.rl import RLRouteRegistry

1
RLRouteRegistry(runtime: Any, *, logger_: logging.Logger | None = None) -> None

components/src/dynamo/common/rl/admin.py#L88

Public methods

init

1
__init__(runtime: Any, *, logger_: logging.Logger | None = None) -> None

No summary available.

source

add_route

1
add_route(name: str, handler: RLRouteHandler) -> None

No summary available.

source

add_routes

1
add_routes(routes: Mapping[str, RLRouteHandler]) -> None

No summary available.

source

describe

1
describe() -> dict[str, Any]

No summary available.

source

dispatch

1
dispatch(request: Mapping[str, Any] | None = None) -> dict[str, Any]

No summary available.

source

dispatch_stream

1
dispatch_stream(request: Mapping[str, Any] | None = None) -> AsyncIterator[dict[str, Any]]

No summary available.

source

Engines for raw, non-token generation (image, video, audio).

1
from dynamo.common.backend import RawEngine

Named for the contract, not a use case: unlike LLMEngine there is no token pipeline — the frontend forwards the OpenAI-shaped request as a JSON object and generate yields the response object(s) directly. Registered with ModelInput.Text and served through the raw request adapter (no tokenization or KV cache). The dict contract is modality-neutral, so a new media modality is a new engine, not a new framework path; one engine may serve several modalities. Yield one (terminal) object, or intermediate progress objects ending with a terminal one. Subclasses like DiffusionEngine add no contract.

components/src/dynamo/common/backend/engine.py#L397

Public methods

generate

1
generate(request: RawRequest, context: Context) -> AsyncGenerator[RawResponseChunk, None]

Yield response object(s) for a single raw-media request.

request is the raw OpenAI-shaped request body (see RawRequest); yield the response body object(s) (see RawResponseChunk). For non-streaming modalities yield exactly one (terminal) object; for streaming modalities yield intermediate progress objects ending with the terminal one.

source

Metrics for requests scheduled in this iteration

1
from dynamo.common.forward_pass_metrics import ScheduledRequestMetrics

components/src/dynamo/common/forward_pass_metrics.py#L84

Advertise which token-overflow requests the frontend may reject early.

1
from dynamo.common.token_budget import TokenBudget

1
TokenBudget(combined_limit: int, reject_prompt_overflow: bool, reject_total_overflow: bool) -> None

A false flag delegates that overflow dimension to the backend. The backend remains responsible for any clamping, truncation, or rejection after that.

components/src/dynamo/common/token_budget.py#L15

Public methods

init

1
__init__(combined_limit: int, reject_prompt_overflow: bool, reject_total_overflow: bool) -> None