dynamo.frontend

1
from dynamo.frontend.vllm_processor import EngineFactory

1
EngineFactory(config: FrontendConfig, flags: Namespace)

1
__init__(config: FrontendConfig, flags: Namespace)

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

1
from dynamo.frontend.frontend_args import FrontendArgGroup

1
add_arguments(parser) -> None

1
from dynamo.frontend.frontend_args import FrontendConfig

1
validate() -> None

1
from dynamo.frontend.utils import PreprocessError

1
PreprocessError(message: str)

1
__init__(message: str)

1
from dynamo.frontend.prepost import PreprocessResult

1
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) -> None

1
__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) -> None

1
from dynamo.frontend.sglang_processor import SglangEngineFactory

1
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)

1
__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)

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

1
from dynamo.frontend.sglang_prepost import SglangPreprocessResult

1
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) -> None

1
__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) -> None

1
from dynamo.frontend.sglang_processor import SglangPreprocessWorkerResult

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

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

1
from dynamo.frontend.sglang_processor import SglangProcessor

1
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)

1
__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)

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

1
from dynamo.frontend.sglang_prepost import SglangStreamingPostProcessor

1
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, eos_token_ids: list[int] | None = None, prompt_token_ids: list[int] | None = None) -> None

1
__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, eos_token_ids: list[int] | None = None, prompt_token_ids: list[int] | None = None) -> None

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

1
from dynamo.frontend.prepost import StreamingPostProcessor

1
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], stream_response: bool = True) -> None

1
__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], stream_response: bool = True) -> None

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

1
from dynamo.frontend.vllm_processor import VllmProcessor

1
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')

1
__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')

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

1
from dynamo.frontend.thinking import apply_default_thinking_mode_to_template_kwargs

1
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]

1
from dynamo.frontend.main import async_main

1
async_main()

1
from dynamo.frontend.sglang_prepost import build_response_format_guided_decoding

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

1
from dynamo.frontend.prepost import build_tool_call_guided_decoding

1
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

1
from dynamo.frontend.sglang_prepost import build_tool_call_guided_decoding

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

1
from dynamo.frontend.sglang_prepost import convert_tools

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

1
from dynamo.frontend.sglang_prepost import create_parsers

1
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]

1
from dynamo.frontend.sglang_prepost import detect_force_reasoning_from_template

1
detect_force_reasoning_from_template(chat_template: str | None) -> bool

1
from dynamo.frontend.utils import extract_mm_urls

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

1
from dynamo.frontend.main import graceful_shutdown

1
graceful_shutdown(runtime: DistributedRuntime) -> None

1
from dynamo.frontend.utils import handle_engine_error

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

1
from dynamo.frontend.main import load_frontend_route_extensions

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

1
from dynamo.frontend.main import main

1
main() -> None

1
from dynamo.frontend.utils import make_backend_error

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

1
from dynamo.frontend.utils import make_internal_error

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

1
from dynamo.frontend.vllm_processor import map_finish_reason

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

1
from dynamo.frontend.utils import nvext_extra_field_requested

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

1
from dynamo.frontend.main import parse_args

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

1
from dynamo.frontend.prepost import preprocess_chat_request

1
preprocess_chat_request(request: dict[str, Any] | ChatCompletionRequest, *, tokenizer: TokenizerLike, renderer: _Renderer, tool_parser_class: type[ToolParser] | 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

1
from dynamo.frontend.sglang_prepost import preprocess_chat_request

1
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

1
from dynamo.frontend.utils import random_call_id

1
random_call_id() -> str

1
from dynamo.frontend.utils import random_uuid

1
random_uuid() -> str

1
from dynamo.frontend.utils import read_jinja_chat_template

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

1
from dynamo.frontend.utils import resolve_chat_template

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

1
from dynamo.frontend.sglang_prepost import resolve_request_force_reasoning

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

1
from dynamo.frontend.thinking import runtime_default_thinking_mode

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

1
from dynamo.frontend.main import setup_engine_factory

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

1
from dynamo.frontend.main import setup_sglang_engine_factory

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

1
from dynamo.frontend.frontend_args import validate_model_name

1
validate_model_name(value: str) -> str

1
from dynamo.frontend.frontend_args import validate_model_path

1
validate_model_path(value: str) -> str

1
from dynamo.frontend.utils import worker_warmup

1
worker_warmup() -> bool