Skip to content
Open
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: 12 additions & 0 deletions components/src/dynamo/frontend/frontend_args.py
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,7 @@ class FrontendConfig(RouterConfigBase, KvRouterConfigBase, AisPerfConfigBase):
model_name: Optional[str]
model_path: Optional[str]
metrics_prefix: Optional[str] = None
ttft_detail: str = "metrics"

kserve_grpc_server: bool
grpc_metrics_port: int
Expand Down Expand Up @@ -228,6 +229,9 @@ class FrontendConfig(RouterConfigBase, KvRouterConfigBase, AisPerfConfigBase):
_VALID_TOKENIZER_BACKENDS = {"default", "fastokens", "basetenkenizer"}

def validate(self) -> None:
if self.ttft_detail not in {"off", "metrics", "logs", "both"}:
raise ValueError("--ttft-detail must be off, metrics, logs, or both")

if self.preprocess_workers is None:
self.preprocess_workers = (
_default_sglang_preprocess_workers()
Expand Down Expand Up @@ -648,6 +652,14 @@ def add_arguments(self, parser) -> None:
arg_type=validate_model_path if self.env is None else str,
env=self.env,
)
add_argument(
g,
flag_name="--ttft-detail",
env_var="DYN_FRONTEND_TTFT_DETAIL",
default="metrics",
choices=["off", "metrics", "logs", "both"],
help="Exclusive TTFT detail; logs also require dynamo_frontend::ttft=debug in DYN_LOG.",
)
add_argument(
g,
flag_name="--metrics-prefix",
Expand Down
1 change: 1 addition & 0 deletions components/src/dynamo/frontend/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -440,6 +440,7 @@ def signal_handler():
"router_config": router_config,
"migration_limit": config.migration_limit,
"metrics_prefix": metrics_prefix,
"ttft_detail": config.ttft_detail,
"enable_anthropic_api": config.enable_anthropic_api,
"enable_systemone_api": config.enable_systemone_api,
"strip_anthropic_preamble": config.strip_anthropic_preamble,
Expand Down
62 changes: 34 additions & 28 deletions components/src/dynamo/frontend/prepost.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
should_attempt_structural_tag,
)
from .thinking import apply_default_thinking_mode_to_template_kwargs
from .ttft import ttft_renderer_stage, ttft_stage
from .utils import legacy_guided_decoding

get_model_structural_tag: Any
Expand Down Expand Up @@ -788,6 +789,7 @@ async def preprocess_chat_request(
structural_tag_scope: str = "auto",
structural_tag_schema: str = "auto",
structural_tag_excludes_reasoning: bool = False,
context: Any | None = None,
) -> PreprocessResult:
validated_request = _validate_chat_completion_request(request)
assistant_guided_decoding = _build_assistant_guided_decoding(validated_request)
Expand Down Expand Up @@ -844,39 +846,43 @@ async def preprocess_chat_request(
# Render first: whether generation starts inside a reasoning block depends
# on the rendered prompt (e.g. a template that ends the generation prompt
# with an open think tag), and the tool-call grammar must match it.
_, engine_prompt = await renderer.render_messages_async(messages, chat_params)
with ttft_renderer_stage(context) as renderer_timing:
_, engine_prompt = await renderer.render_messages_async(messages, chat_params)
renderer_timing[0] = "prompt_token_ids" in engine_prompt

if "prompt_token_ids" in engine_prompt:
tokens = list(engine_prompt["prompt_token_ids"])
else:
async_tokenizer = _get_async_tokenizer(tokenizer)
encoded = await async_tokenizer(
engine_prompt["prompt"],
add_special_tokens=request_for_sampling.add_special_tokens,
)
tokens = list(encoded.input_ids)

tool_guided_decoding = build_tool_call_guided_decoding(
request_for_sampling,
tool_parser,
parser_guided_decoding=parser_guided_decoding,
structural_tag_mode=structural_tag_mode,
structural_tag_scope=structural_tag_scope,
structural_tag_schema=structural_tag_schema,
starts_in_reasoning=(
structural_tag_mode == "on"
and tool_parser is not None
and bool(request_for_sampling.tools)
and not structural_tag_excludes_reasoning
and _prompt_leaves_reasoning_open(
tokens,
tokenizer=tokenizer,
reasoning_parser_class=reasoning_parser_class,
chat_template_kwargs=chat_template_kwargs,
model_config=model_config,
with ttft_stage(context, "tokenize"):
async_tokenizer = _get_async_tokenizer(tokenizer)
encoded = await async_tokenizer(
engine_prompt["prompt"],
add_special_tokens=request_for_sampling.add_special_tokens,
)
),
)
tokens = list(encoded.input_ids)

with ttft_stage(context, "structural_tag_build"):
tool_guided_decoding = build_tool_call_guided_decoding(
request_for_sampling,
tool_parser,
parser_guided_decoding=parser_guided_decoding,
structural_tag_mode=structural_tag_mode,
structural_tag_scope=structural_tag_scope,
structural_tag_schema=structural_tag_schema,
starts_in_reasoning=(
structural_tag_mode == "on"
and tool_parser is not None
and bool(request_for_sampling.tools)
and not structural_tag_excludes_reasoning
and _prompt_leaves_reasoning_open(
tokens,
tokenizer=tokenizer,
reasoning_parser_class=reasoning_parser_class,
chat_template_kwargs=chat_template_kwargs,
model_config=model_config,
)
),
)
# A forced tool choice already claims the decoder's single grammar slot, so a
# caller-set guided_*/structured_outputs constraint over the same token stream
# cannot also be honored. Reject it (InvalidArgument -> 400) rather than
Expand Down
61 changes: 36 additions & 25 deletions components/src/dynamo/frontend/sglang_prepost.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
should_attempt_structural_tag,
)
from .thinking import apply_default_thinking_mode_to_template_kwargs
from .ttft import ttft_coarse, ttft_stage
from .utils import PreprocessError, legacy_guided_decoding, random_call_id

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -522,6 +523,7 @@ def _render_deepseek_v4_prompt_token_ids(
messages: list[dict[str, Any]],
tokenizer,
template_tools: list[dict[str, Any]] | None,
context: Any | None = None,
) -> list[int]:
"""Render DeepSeek-V4 prompt token ids via SGLang's ``encoding_dsv4``.

Expand Down Expand Up @@ -581,12 +583,14 @@ def _render_deepseek_v4_prompt_token_ids(
if reasoning_effort not in ("max", "high", None):
reasoning_effort = None

prompt = encode_messages(
encoding_messages,
thinking_mode=thinking_mode,
reasoning_effort=reasoning_effort,
)
return _normalize_prompt_token_ids(tokenizer.encode(prompt))
with ttft_stage(context, "template_render"):
prompt = encode_messages(
encoding_messages,
thinking_mode=thinking_mode,
reasoning_effort=reasoning_effort,
)
with ttft_stage(context, "tokenize"):
return _normalize_prompt_token_ids(tokenizer.encode(prompt))


@lru_cache(maxsize=64)
Expand Down Expand Up @@ -890,6 +894,7 @@ def preprocess_chat_request(
structural_tag_mode: str = "off",
structural_tag_scope: str = "auto",
structural_tag_schema: str = "auto",
context: Any | None = None,
) -> SglangPreprocessResult:
"""Preprocess a chat request using SGLang tokenizer and parser APIs.

Expand Down Expand Up @@ -977,6 +982,7 @@ def preprocess_chat_request(
messages=messages,
tokenizer=tokenizer,
template_tools=template_tools,
context=context,
)
else:
# Build template kwargs -- single call for rendering + tokenization
Expand Down Expand Up @@ -1004,25 +1010,30 @@ def preprocess_chat_request(

template_messages = _normalize_messages_for_template(messages, tokenizer)

try:
rendered = tokenizer.apply_chat_template(
template_messages, **template_kwargs
)
except (TemplateError, TypeError) as exc:
# Jinja filters such as tojson can raise TypeError for invalid inputs.
raise PreprocessError(str(exc)) from exc
prompt_token_ids = _normalize_prompt_token_ids(rendered)

response_format_guided_decoding = build_response_format_guided_decoding(request)
tool_call_guided_decoding = build_tool_call_guided_decoding(
request,
tool_call_parser_name=tool_call_parser_name,
sglang_tools=sglang_tools,
structural_tag_mode=structural_tag_mode,
structural_tag_scope=structural_tag_scope,
structural_tag_schema=structural_tag_schema,
tokenizer=tokenizer,
)
# A custom apply_chat_template may combine backend-specific rendering
# and tokenization. Preserve its single call and expose the opaque span.
ttft_coarse(context, "opaque_template")
with ttft_stage(context, "template_tokenize"):
try:
rendered = tokenizer.apply_chat_template(
template_messages, **template_kwargs
)
except (TemplateError, TypeError) as exc:
# Jinja filters such as tojson can raise TypeError for invalid inputs.
raise PreprocessError(str(exc)) from exc
prompt_token_ids = _normalize_prompt_token_ids(rendered)

with ttft_stage(context, "structural_tag_build"):
response_format_guided_decoding = build_response_format_guided_decoding(request)
tool_call_guided_decoding = build_tool_call_guided_decoding(
request,
tool_call_parser_name=tool_call_parser_name,
sglang_tools=sglang_tools,
structural_tag_mode=structural_tag_mode,
structural_tag_scope=structural_tag_scope,
structural_tag_schema=structural_tag_schema,
tokenizer=tokenizer,
)
# This path also never reads the legacy guided_json / guided_regex /
# guided_grammar / guided_choice fields at all, so those are dropped silently
# while both other paths honor them (and reject them against a forced choice).
Expand Down
88 changes: 65 additions & 23 deletions components/src/dynamo/frontend/sglang_processor.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,13 @@
)
from .structural_tag_policy import runtime_structural_tag_options
from .thinking import runtime_default_thinking_mode
from .ttft import (
TimingBatch,
WorkerTiming,
import_worker_batch,
ttft_context,
ttft_stage,
)
from .usage import response_usage, update_usage_metadata
from .utils import (
PreprocessError,
Expand Down Expand Up @@ -347,6 +354,16 @@ class SglangPreprocessWorkerResult:
effective_reasoning_parser_name: str | None = None


@dataclass
class SglangTimedPreprocessWorkerResult(SglangPreprocessWorkerResult):
"""Enabled-only payload; the off result does not serialize a timing field."""

timing_batch: TimingBatch | None = None


_WorkerPreprocessTiming = WorkerTiming


def _init_worker(
model_path: str,
tool_call_parser_name: str | None,
Expand Down Expand Up @@ -382,8 +399,10 @@ def _preprocess_worker(
request: dict[str, Any],
model_name: str,
eos_token_ids: list[int] | None,
record_ttft_timing: bool = False,
) -> SglangPreprocessWorkerResult:
"""Preprocess a request in a worker process and return a picklable result."""
timing = _WorkerPreprocessTiming() if record_ttft_timing else None
pre = preprocess_chat_request(
request,
tokenizer=_w_tokenizer,
Expand All @@ -395,6 +414,7 @@ def _preprocess_worker(
structural_tag_mode=_w_structural_tag_mode,
structural_tag_scope=_w_structural_tag_scope,
structural_tag_schema=_w_structural_tag_schema,
context=timing,
)

n = request.get("n", 1)
Expand All @@ -419,14 +439,19 @@ def _preprocess_worker(
_w_reasoning_parser_name if _client_wants_separate_reasoning(request) else None
)

return SglangPreprocessWorkerResult(
result = SglangPreprocessWorkerResult(
prompt_token_ids=pre.prompt_token_ids,
dynamo_preproc=dynamo_preproc,
request=request,
force_reasoning=pre.force_reasoning,
effective_reasoning_parser_name=effective_reasoning_parser_name,
named_zero_arg_tool=pre.named_zero_arg_tool,
)
if timing is not None:
return SglangTimedPreprocessWorkerResult(
**vars(result), timing_batch=timing.batch()
)
return result


def _build_dynamo_preproc(
Expand Down Expand Up @@ -680,23 +705,26 @@ async def _generator_inner(
) -> AsyncGenerator[dict[str, Any], None]:
"""Single-process path: preprocess, dispatch, stream post-process."""
request_id = random_uuid()
timing_context = ttft_context(context)

try:
if self.debug_perf:
t0 = time.monotonic()

pre = preprocess_chat_request(
request,
tokenizer=self.tokenizer,
tool_call_parser_name=self.tool_call_parser_name,
reasoning_parser_name=self.reasoning_parser_name,
exclude_tools_when_tool_choice_none=self.exclude_tools_when_tool_choice_none,
template_force_reasoning=self.template_force_reasoning,
default_thinking_mode=self.default_thinking_mode,
structural_tag_mode=self.structural_tag_mode,
structural_tag_scope=self.structural_tag_scope,
structural_tag_schema=self.structural_tag_schema,
)
with ttft_stage(timing_context, "preprocess"):
pre = preprocess_chat_request(
request,
tokenizer=self.tokenizer,
tool_call_parser_name=self.tool_call_parser_name,
reasoning_parser_name=self.reasoning_parser_name,
exclude_tools_when_tool_choice_none=self.exclude_tools_when_tool_choice_none,
template_force_reasoning=self.template_force_reasoning,
default_thinking_mode=self.default_thinking_mode,
structural_tag_mode=self.structural_tag_mode,
structural_tag_scope=self.structural_tag_scope,
structural_tag_schema=self.structural_tag_schema,
context=timing_context,
)

if self.debug_perf:
t1 = time.monotonic()
Expand Down Expand Up @@ -760,17 +788,31 @@ async def _generator_inner_pool(
# --- Phase 1: Preprocess (semaphore held) ---
assert self._worker_semaphore is not None
assert self.preprocess_pool is not None
timing_context = ttft_context(context)
enabled = timing_context is not None
try:
async with self._worker_semaphore:
future = self.preprocess_pool.submit(
_preprocess_worker,
request,
request["model"],
self.eos_token_ids,
)
preproc_result: SglangPreprocessWorkerResult = (
await asyncio.wrap_future(future)
)
with ttft_stage(timing_context, "preprocess"):
with ttft_stage(timing_context, "preprocess_queue"):
await self._worker_semaphore.acquire()
try:
submitted_ns = time.monotonic_ns() if enabled else 0
# Omit the optional argument entirely when disabled, and use
# the base result without any timing payload on that path.
args = (request, request["model"], self.eos_token_ids)
future = self.preprocess_pool.submit(
_preprocess_worker, *args, *((True,) if enabled else ())
)
preproc_result: SglangPreprocessWorkerResult = (
await asyncio.wrap_future(future)
)
finally:
self._worker_semaphore.release()
if enabled:
import_worker_batch(
timing_context,
getattr(preproc_result, "timing_batch", None),
submitted_ns,
)
except InvalidArgument:
raise
except PreprocessError as exc:
Expand Down
Loading
Loading