dynamo.frontend | NVIDIA Dynamo Documentation

dynamo.frontend publishes 12 classes and 37 functions. Source: components/src/dynamo/frontend/__init__.py

EngineFactory (class)

No summary available.

from dynamo.frontend.vllm_processor import EngineFactory

EngineFactory(config: FrontendConfig, flags: Namespace)

components/src/dynamo/frontend/vllm_processor.py#L1088

Public methods

init

__init__(config: FrontendConfig, flags: Namespace)

No summary available.

source

chat_engine_factory

chat_engine_factory(instance_id: ModelCardInstanceId, mdc: ModelDeploymentCard, routed_engine: RoutedEngine) -> PythonAsyncEngine

Called by Rust when a model is discovered.

source

FrontendArgGroup (class)

Frontend configuration parameters.

from dynamo.frontend.frontend_args import FrontendArgGroup

components/src/dynamo/frontend/frontend_args.py#L228

Public methods

add_arguments

add_arguments(parser) -> None

No summary available.

source

FrontendConfig (class)

Configuration for the Dynamo frontend.

from dynamo.frontend.frontend_args import FrontendConfig

components/src/dynamo/frontend/frontend_args.py#L55

Public methods

validate

validate() -> None

No summary available.

source

PreprocessError (class)

Raised by preprocess workers for user-facing errors (e.g., n!=1).

from dynamo.frontend.utils import PreprocessError

PreprocessError(message: str)

Carries a plain message because the worker→main-process boundary pickles the exception; the main process re-raises a Dynamo-typed exception so PyO3 can route it through the proper backend-error path.

components/src/dynamo/frontend/utils.py#L131

Public methods

init

__init__(message: str)

No summary available.

source

PreprocessResult (class)

No summary available.

from dynamo.frontend.prepost import PreprocessResult

PreprocessResult(request_for_sampling: ChatCompletionRequest, tool_parser: ToolParser | None, chat_template_kwargs: dict[str, Any], engine_prompt: dict[str, Any], prompt_token_ids: list[int], guided_decoding: dict[str, Any] | None = None, uses_dynamo_json_tool_call_fallback: bool = False) -> None

components/src/dynamo/frontend/prepost.py#L61

Public methods

init

__init__(request_for_sampling: ChatCompletionRequest, tool_parser: ToolParser | None, chat_template_kwargs: dict[str, Any], engine_prompt: dict[str, Any], prompt_token_ids: list[int], guided_decoding: dict[str, Any] | None = None, uses_dynamo_json_tool_call_fallback: bool = False) -> None

No summary available.

source

SglangEngineFactory (class)

No summary available.

from dynamo.frontend.sglang_processor import SglangEngineFactory

SglangEngineFactory(config: FrontendConfig, debug_perf: bool = False, tool_call_parser_name: str | None = None, reasoning_parser_name: str | None = None, chat_template: str | None = None)

components/src/dynamo/frontend/sglang_processor.py#L949

Public methods

init

__init__(config: FrontendConfig, debug_perf: bool = False, tool_call_parser_name: str | None = None, reasoning_parser_name: str | None = None, chat_template: str | None = None)

No summary available.

source

chat_engine_factory

chat_engine_factory(instance_id: ModelCardInstanceId, mdc: ModelDeploymentCard, routed_engine: RoutedEngine) -> PythonAsyncEngine

Called by Rust when a model is discovered.

source

SglangPreprocessResult (class)

Result of SGLang preprocessing.

from dynamo.frontend.sglang_prepost import SglangPreprocessResult

SglangPreprocessResult(prompt_token_ids: list[int], tool_call_parser: ToolCallParserType | None, reasoning_parser: ReasoningParser | None, guided_decoding: dict[str, Any] | None, request: dict[str, Any], force_reasoning: bool = False, named_zero_arg_tool: str | None = None) -> None

components/src/dynamo/frontend/sglang_prepost.py#L47

Public methods

init

__init__(prompt_token_ids: list[int], tool_call_parser: ToolCallParserType | None, reasoning_parser: ReasoningParser | None, guided_decoding: dict[str, Any] | None, request: dict[str, Any], force_reasoning: bool = False, named_zero_arg_tool: str | None = None) -> None

No summary available.

source

SglangPreprocessWorkerResult (class)

Picklable return value from the SGLang preprocess worker.

from dynamo.frontend.sglang_processor import SglangPreprocessWorkerResult

SglangPreprocessWorkerResult(prompt_token_ids: list[int], dynamo_preproc: dict[str, Any], request: dict[str, Any], force_reasoning: bool = False, named_zero_arg_tool: str | None = None, effective_reasoning_parser_name: str | None = None) -> None

components/src/dynamo/frontend/sglang_processor.py#L292

Public methods

init

__init__(prompt_token_ids: list[int], dynamo_preproc: dict[str, Any], request: dict[str, Any], force_reasoning: bool = False, named_zero_arg_tool: str | None = None, effective_reasoning_parser_name: str | None = None) -> None

No summary available.

source

SglangProcessor (class)

No summary available.

from dynamo.frontend.sglang_processor import SglangProcessor

SglangProcessor(tokenizer, routed_engine: RoutedEngine, tool_call_parser_name: str | None, reasoning_parser_name: str | None, eos_token_ids: list[int] | None, debug_perf: bool = False, preprocess_pool: ProcessPoolExecutor | None = None, preprocess_workers: int = 0, stream_interval: int = 1, default_thinking_mode: str | None = None)

components/src/dynamo/frontend/sglang_processor.py#L483

Public methods

init

__init__(tokenizer, routed_engine: RoutedEngine, tool_call_parser_name: str | None, reasoning_parser_name: str | None, eos_token_ids: list[int] | None, debug_perf: bool = False, preprocess_pool: ProcessPoolExecutor | None = None, preprocess_workers: int = 0, stream_interval: int = 1, default_thinking_mode: str | None = None)

No summary available.

source

generator

generator(request: dict[str, Any], context: Any | None = None) -> AsyncGenerator[dict[str, Any], None]

Main entry point: preprocess, route, post-process a chat request.

source

SglangStreamingPostProcessor (class)

Streaming post-processor using SGLang parsers and HF tokenizer detokenization.

from dynamo.frontend.sglang_prepost import SglangStreamingPostProcessor

SglangStreamingPostProcessor(*, tokenizer, tool_call_parser: ToolCallParserType | None, reasoning_parser: ReasoningParser | None, history_tool_calls_count: int = 0, sglang_tools: list[SglangTool] | None = None, tool_call_parser_name: str | None = None, named_zero_arg_tool: str | None = None, eos_token_ids: list[int] | None = None, prompt_token_ids: list[int] | None = None, stop_strings: set[str] | None = None, stop_token_ids: set[int] | None = None, skip_special_tokens: bool | None = None) -> None

Handles:

Incremental detokenization across tokenizer-safe boundaries
Reasoning content extraction via SGLang ReasoningParser
Tool call parsing via SGLang FunctionCallParser or JsonArrayParser

components/src/dynamo/frontend/sglang_prepost.py#L1012

Public methods

init

__init__(*, tokenizer, tool_call_parser: ToolCallParserType | None, reasoning_parser: ReasoningParser | None, history_tool_calls_count: int = 0, sglang_tools: list[SglangTool] | None = None, tool_call_parser_name: str | None = None, named_zero_arg_tool: str | None = None, eos_token_ids: list[int] | None = None, prompt_token_ids: list[int] | None = None, stop_strings: set[str] | None = None, stop_token_ids: set[int] | None = None, skip_special_tokens: bool | None = None) -> None

No summary available.

source

process_output

process_output(engine_response: dict[str, Any]) -> dict[str, Any] | None

Process a single engine response chunk into an OpenAI SSE choice dict.

Parameters

engine_response

dict[str, Any]

Dict with token_ids and optional finish_reason.

Returns

dict[str, Any] | None — OpenAI choice dict or None if nothing to emit yet.

source

StreamingPostProcessor (class)

No summary available.

from dynamo.frontend.prepost import StreamingPostProcessor

StreamingPostProcessor(*, tokenizer: TokenizerLike, request_for_sampling: ChatCompletionRequest, sampling_params: SamplingParams, prompt_token_ids: Sequence[int], tool_parser: ToolParser | None, reasoning_parser_class: type[ReasoningParser] | None, chat_template_kwargs: dict[str, Any], model_config: ModelConfig | None = None, response_reasoning_ended: bool | None = None, stream_response: bool = True, uses_dynamo_json_tool_call_fallback: bool = False) -> None

components/src/dynamo/frontend/prepost.py#L775

Public methods

init

__init__(*, tokenizer: TokenizerLike, request_for_sampling: ChatCompletionRequest, sampling_params: SamplingParams, prompt_token_ids: Sequence[int], tool_parser: ToolParser | None, reasoning_parser_class: type[ReasoningParser] | None, chat_template_kwargs: dict[str, Any], model_config: ModelConfig | None = None, response_reasoning_ended: bool | None = None, stream_response: bool = True, uses_dynamo_json_tool_call_fallback: bool = False) -> None

No summary available.

source

process_output

process_output(output: Any) -> dict[str, Any] | None

No summary available.

source

VllmProcessor (class)

No summary available.

from dynamo.frontend.vllm_processor import VllmProcessor

VllmProcessor(tokenizer: TokenizerLike, input_processor: InputProcessor, output_processor: OutputProcessor, tool_parser_class: type[ToolParser] | None, reasoning_parser_class: type[ReasoningParser] | None, routed_engine: RoutedEngine, block_size: int = 16, enable_auto_tool_choice: bool = False, default_chat_template_kwargs: dict[str, Any] | None = None, default_thinking_mode: str | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto')

components/src/dynamo/frontend/vllm_processor.py#L353

Public methods

init

__init__(tokenizer: TokenizerLike, input_processor: InputProcessor, output_processor: OutputProcessor, tool_parser_class: type[ToolParser] | None, reasoning_parser_class: type[ReasoningParser] | None, routed_engine: RoutedEngine, block_size: int = 16, enable_auto_tool_choice: bool = False, default_chat_template_kwargs: dict[str, Any] | None = None, default_thinking_mode: str | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto')

No summary available.

source

generator

generator(request: dict[str, Any], context: Any | None = None) -> AsyncGenerator[dict[str, Any], None]

Run a single request through the engine. Does pre and post processing on this machine, delegates model inference to a backend using the router.

source

apply_default_thinking_mode_to_template_kwargs (function)

Merge deployment thinking default unless the request already controls it.

from dynamo.frontend.thinking import apply_default_thinking_mode_to_template_kwargs

apply_default_thinking_mode_to_template_kwargs(chat_template_kwargs: dict[str, Any], default_thinking_mode: str | None, *, request_has_root_thinking: bool = False) -> dict[str, Any]

components/src/dynamo/frontend/thinking.py#L33

as_error_envelope (function)

Tag an error dict so the binding reads it as an error frame.

from dynamo.frontend.utils import as_error_envelope

as_error_envelope(error_payload: dict[str, Any]) -> dict[str, Any]

Without _dynamo_annotated the binding reads the dict as a completion chunk and fails with missing field `id```. The client then gets a 500 and never sees the message. The HTTP layer reads the text back off comment. See depythonize_annotatedinlib/bindings/python/rust/engine.rs“.

components/src/dynamo/frontend/utils.py#L230

async_main (function)

Main async entry point for the Dynamo frontend.

from dynamo.frontend.main import async_main

async_main()

Initializes the distributed runtime, configures routing, and starts the HTTP server or interactive mode based on command-line arguments.

components/src/dynamo/frontend/main.py#L355

backend_invalid_argument_to_http_error (function)

Recover a worker’s own HTTP status from a serialized backend error.

from dynamo.frontend.utils import backend_invalid_argument_to_http_error

backend_invalid_argument_to_http_error(exc: BaseException) -> HttpError | None

A worker that rejects a request — an unsupported sampling parameter, an unparseable grammar — fails it with {"message": ..., "code": 4xx}. Crossing the Rust boundary turns that into a plain Python exception whose text is BackendInvalidArgument: {json}, so a generic except Exception reports a 500 and discards both the status and the reason.

Rust’s HTTP service already performs this recovery (extract_backend_error_if_present in lib/llm/src/http/service/openai.rs), but the Python chat processors do not go through it, so they need their own. Mirror its status rules: honour an explicit code, and otherwise fall back to 400 — the prefix itself is the discriminator proving the argument was invalid.

Returns None when exc is not that shape, so callers fall through to their existing handling instead of inventing a status.

components/src/dynamo/frontend/utils.py#L245

build_response_format_guided_decoding (function)

Build Dynamo guided decoding from OpenAI chat response_format.

from dynamo.frontend.sglang_prepost import build_response_format_guided_decoding

build_response_format_guided_decoding(request: dict[str, Any]) -> dict[str, Any] | None

components/src/dynamo/frontend/sglang_prepost.py#L645

build_tool_call_guided_decoding (function)

Build tool-call guidance through vLLM’s configured tool parser.

from dynamo.frontend.prepost import build_tool_call_guided_decoding

build_tool_call_guided_decoding(request: ChatCompletionRequest, tool_parser: ToolParser | None, *, parser_guided_decoding: dict[str, Any] | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto') -> dict[str, Any] | None

components/src/dynamo/frontend/prepost.py#L223

build_tool_call_guided_decoding (function)

Build native-SGLang-like tool call constraints for guided decoding.

from dynamo.frontend.sglang_prepost import build_tool_call_guided_decoding

build_tool_call_guided_decoding(request: dict[str, Any], *, tool_call_parser_name: str | None, sglang_tools: list[SglangTool] | None) -> dict[str, Any] | None

components/src/dynamo/frontend/sglang_prepost.py#L565

convert_tools (function)

Convert OpenAI tool dicts to SGLang Tool objects.

from dynamo.frontend.sglang_prepost import convert_tools

convert_tools(tools: list[dict[str, Any]] | None) -> list[SglangTool] | None

components/src/dynamo/frontend/sglang_prepost.py#L202

create_parsers (function)

Create tool call and reasoning parsers for a request.

from dynamo.frontend.sglang_prepost import create_parsers

create_parsers(request: dict[str, Any], *, tool_call_parser_name: str | None, reasoning_parser_name: str | None, sglang_tools: list[SglangTool] | None = None, force_reasoning: bool = False) -> tuple[ToolCallParserType | None, ReasoningParser | None]

Shared by both the single-process preprocessing path and the pool path (which must recreate non-picklable parsers in the main process).

If sglang_tools is provided, reuses them; otherwise converts from the request’s tools field.

For tool_choice="required" or a named function, uses JsonArrayParser (matching native SGLang) since guided decoding constrains the output to a JSON array. Otherwise uses the model-specific FunctionCallParser.

components/src/dynamo/frontend/sglang_prepost.py#L276

detect_force_reasoning_from_template (function)

Return True if the chat template auto-opens a reasoning block.

from dynamo.frontend.sglang_prepost import detect_force_reasoning_from_template

detect_force_reasoning_from_template(chat_template: str | None) -> bool

Intended to be called once at processor startup with tokenizer.chat_template and cached on the processor.

components/src/dynamo/frontend/sglang_prepost.py#L84

extract_mm_urls (function)

Extract media and vLLM processor-cache UUIDs from chat messages.

from dynamo.frontend.utils import extract_mm_urls

extract_mm_urls(messages: list[dict[str, Any]]) -> tuple[dict[str, list[dict[str, str]]] | None, dict[str, list[str | None]] | None]

URL-backed parts become Url variants. Image parts with no URL and an opaque uuid become UuidOnly variants for vLLM’s multimodal processor cache. UUID lists preserve slot order for every media type

({"image_url": [{"Url": "https://..."}, {"UuidOnly": "image-1"}]},
 {"image_url": ["image-1", "image-1"]})

The UUID map is None when no user UUID is present. A media content part with neither a URL nor UUID is rejected instead of being silently dropped.

components/src/dynamo/frontend/utils.py#L148

graceful_shutdown (function)

Handle graceful shutdown of the distributed runtime.

from dynamo.frontend.main import graceful_shutdown

graceful_shutdown(runtime: DistributedRuntime) -> None

Parameters

runtime

DistributedRuntime

The DistributedRuntime instance to shut down.

components/src/dynamo/frontend/main.py#L486

handle_engine_error (function)

Classify an invalid engine response and return an OpenAI-style error dict.

from dynamo.frontend.utils import handle_engine_error

handle_engine_error(engine_response: Any, request_id: str, logger: logging.Logger) -> dict[str, Any]

Called when engine_response is None or missing ‘token_ids’.

components/src/dynamo/frontend/utils.py#L291

legacy_guided_decoding (function)

Convert one legacy guided-decoding constraint and its modifier to a dict.

from dynamo.frontend.utils import legacy_guided_decoding

legacy_guided_decoding(request: dict[str, Any]) -> dict[str, Any] | None

components/src/dynamo/frontend/utils.py#L47

load_frontend_route_extensions (function)

Load trusted frontend route extensions.

from dynamo.frontend.main import load_frontend_route_extensions

load_frontend_route_extensions(extension_names: list[str]) -> list[FrontendRoute]

Each value is either a name registered under the dynamo.frontend.routes entry-point group (preferred) or a direct module:function path.

components/src/dynamo/frontend/main.py#L215

main (function)

Entry point for the Dynamo frontend CLI.

from dynamo.frontend.main import main

main() -> None

components/src/dynamo/frontend/main.py#L495

make_backend_error (function)

Build an OpenAI-style error dict, guarding against None/missing message.

from dynamo.frontend.utils import make_backend_error

make_backend_error(engine_response: dict[str, Any]) -> dict[str, Any]

components/src/dynamo/frontend/utils.py#L208

make_internal_error (function)

Build an OpenAI-style internal error dict with request-specific fallback.

from dynamo.frontend.utils import make_internal_error

make_internal_error(request_id: str, detail: str | None = None) -> dict[str, Any]

components/src/dynamo/frontend/utils.py#L219

map_finish_reason (function)

No summary available.

from dynamo.frontend.vllm_processor import map_finish_reason

map_finish_reason(raw_reason: str | None) -> FinishReason | None

components/src/dynamo/frontend/vllm_processor.py#L118

named_closed_zero_arg_tool (function)

Return the named tool when its only valid argument value is {}.

from dynamo.frontend.sglang_prepost import named_closed_zero_arg_tool

named_closed_zero_arg_tool(request: dict[str, Any]) -> str | None

components/src/dynamo/frontend/sglang_prepost.py#L336

nvext_extra_field_requested (function)

Return whether a request opted into a response nvext field.

from dynamo.frontend.utils import nvext_extra_field_requested

nvext_extra_field_requested(request: dict[str, Any], field: str) -> bool

components/src/dynamo/frontend/utils.py#L117

parse_args (function)

Parse command-line arguments for the Dynamo frontend.

from dynamo.frontend.main import parse_args

parse_args() -> tuple[FrontendConfig, Optional[Namespace], Optional[Namespace]]

Returns

tuple[FrontendConfig, Optional[Namespace], Optional[Namespace]] — Tuple of (FrontendConfig, vllm_flags, sglang_flags).

components/src/dynamo/frontend/main.py#L243

preprocess_chat_request (function)

No summary available.

from dynamo.frontend.prepost import preprocess_chat_request

preprocess_chat_request(request: dict[str, Any] | ChatCompletionRequest, *, tokenizer: TokenizerLike, renderer: _Renderer, tool_parser_class: type[ToolParser] | None, reasoning_parser_class: type[ReasoningParser] | None = None, model_config: ModelConfig | None = None, exclude_tools_when_tool_choice_none: bool = True, enable_auto_tool_choice: bool = False, default_chat_template_kwargs: dict[str, Any] | None = None, default_thinking_mode: str | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto') -> PreprocessResult

components/src/dynamo/frontend/prepost.py#L590

preprocess_chat_request (function)

Preprocess a chat request using SGLang tokenizer and parser APIs.

from dynamo.frontend.sglang_prepost import preprocess_chat_request

preprocess_chat_request(request: dict[str, Any], *, tokenizer, tool_call_parser_name: str | None, reasoning_parser_name: str | None, exclude_tools_when_tool_choice_none: bool = True, template_force_reasoning: bool = False, default_thinking_mode: str | None = None) -> SglangPreprocessResult

template_force_reasoning is the static per-server flag derived from the chat template (see detect_force_reasoning_from_template); the effective per-request value combines it with the configured parser and request-level thinking controls.

Synchronous — suitable for both main-process and worker-process execution.

components/src/dynamo/frontend/sglang_prepost.py#L730

random_call_id (function)

Generate a random tool call ID in OpenAI format.

from dynamo.frontend.utils import random_call_id

random_call_id() -> str

components/src/dynamo/frontend/utils.py#L112

random_uuid (function)

Generate a random 16-character hex UUID.

from dynamo.frontend.utils import random_uuid

random_uuid() -> str

components/src/dynamo/frontend/utils.py#L107

read_jinja_chat_template (function)

Read a Jinja chat template using backend-specific file semantics.

from dynamo.frontend.utils import read_jinja_chat_template

read_jinja_chat_template(template_path: str, *, backend: ChatProcessorBackend) -> str

components/src/dynamo/frontend/utils.py#L66

resolve_chat_template (function)

Return a chat template stored beside the model, or None.

from dynamo.frontend.utils import resolve_chat_template

resolve_chat_template(source_path: str, *, backend: ChatProcessorBackend = 'vllm') -> str | None

Covers models (e.g. Qwen3-Omni) whose template lives in chat_template.json or chat_template.jinja rather than tokenizer_config.json, which the HF tokenizer does not merge. The backend selects native .jinja file semantics.

components/src/dynamo/frontend/utils.py#L81

resolve_request_force_reasoning (function)

Resolve the effective force_reasoning flag for a single request.

from dynamo.frontend.sglang_prepost import resolve_request_force_reasoning

resolve_request_force_reasoning(request: dict[str, Any], reasoning_parser_name: str | None, template_default: bool) -> bool

Mirrors sglang.srt.entrypoints.openai.serving_chat._get_reasoning_from_request combined with template_manager.force_reasoning:

opt-out families (glm45/qwen3/kimi_k2/kimi_k3/…): on by default, chat_template_kwargs.enable_thinking=False (or thinking=False for Kimi) disables it.
MiniMax-M3 defaults to adaptive, but SGLang still enables the reasoning parser unless chat_template_kwargs.thinking_mode is explicitly "disabled".
Mistral is enabled only when reasoning_effort is present and not "none".
opt-in families (deepseek-v3/gemma4): off by default, enabled by chat_template_kwargs.{thinking,enable_thinking}=True.
anything else: follow the statically-detected template default.

components/src/dynamo/frontend/sglang_prepost.py#L128

resolve_skip_special_tokens (function)

Honor explicit decoding options without hiding parser delimiters.

from dynamo.frontend.sglang_prepost import resolve_skip_special_tokens

resolve_skip_special_tokens(requested: bool | None, *, has_parser: bool) -> bool

components/src/dynamo/frontend/sglang_prepost.py#L1005

runtime_default_thinking_mode (function)

Read deployment-level default thinking mode from model runtime metadata.

from dynamo.frontend.thinking import runtime_default_thinking_mode

runtime_default_thinking_mode(runtime_config: dict[str, Any] | None) -> str | None

components/src/dynamo/frontend/thinking.py#L20

setup_engine_factory (function)

When using vllm pre and post processor, create the EngineFactory that creates the engines that run requests.

from dynamo.frontend.main import setup_engine_factory

setup_engine_factory(config: FrontendConfig, vllm_flags: Namespace) -> EngineFactory

components/src/dynamo/frontend/main.py#L98

setup_sglang_engine_factory (function)

When using sglang pre and post processor, create the SglangEngineFactory that creates the engines that run requests.

from dynamo.frontend.main import setup_sglang_engine_factory

setup_sglang_engine_factory(config: FrontendConfig, sglang_flags: Optional[Namespace] = None)

components/src/dynamo/frontend/main.py#L111

validate_legacy_guided_decoding_constraints (function)

Reject malformed or multiple legacy guided-decoding constraints.

from dynamo.frontend.utils import validate_legacy_guided_decoding_constraints

validate_legacy_guided_decoding_constraints(request: dict[str, Any]) -> None

components/src/dynamo/frontend/utils.py#L20

validate_model_name (function)

Validate that model-name is a non-empty string.

from dynamo.frontend.frontend_args import validate_model_name

validate_model_name(value: str) -> str

components/src/dynamo/frontend/frontend_args.py#L37

validate_model_path (function)

Validate that model-path is a valid directory on disk.

from dynamo.frontend.frontend_args import validate_model_path

validate_model_path(value: str) -> str

components/src/dynamo/frontend/frontend_args.py#L46

worker_warmup (function)

Dummy task to ensure a ProcessPoolExecutor worker is fully initialized.

from dynamo.frontend.utils import worker_warmup

worker_warmup() -> bool

components/src/dynamo/frontend/utils.py#L126