Skip to content
Merged
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
86 changes: 67 additions & 19 deletions components/src/dynamo/frontend/sglang_prepost.py
Original file line number Diff line number Diff line change
Expand Up @@ -1279,6 +1279,8 @@ def __init__(
self._pending_stop_text = ""
self._locally_finished = False
self._local_stop_reason: str | None = None
# Decode context, decoded IDs, stop text and hidden stop-token count.
self._local_stop_boundary: tuple[list[int], list[int], str, int] | None = None

# Keep a small, known-complete prompt suffix as decode context. Generated
# tokens are promoted to context only after they decode without a
Expand Down Expand Up @@ -1385,6 +1387,25 @@ def _incremental_decode(
self._pending_decode_ids = []
return delta_text

def _count_tokens_through(
self, context_ids: list[int], token_ids: list[int], text: str
) -> int:
"""Count leading ``token_ids`` whose incremental decode covers ``text``.
Once a decoded prefix covers ``text``, every longer one does, so bisect
instead of decoding every prefix.
"""
context_text = self._decode_ids(context_ids)
low, high = 0, len(token_ids)
while low < high:
count = (low + high) // 2
decoded = self._decode_ids(context_ids + token_ids[:count])
if decoded[len(context_text) :].startswith(text):
high = count
else:
low = count + 1
return low

def _build_openai_logprobs(
self,
log_probs: list[float],
Expand Down Expand Up @@ -1527,6 +1548,18 @@ def locally_finished(self) -> bool:
def local_stop_reason(self) -> str | None:
return self._local_stop_reason

def post_stop_token_count(self) -> int:
"""Count the latest generated IDs that follow a local stop string.
The token completing the stop is kept. The count spans IDs withheld by
incremental decoding and is computed on demand by re-decoding them.
"""
if self._local_stop_boundary is None:
return 0
context_ids, decoded_ids, stop_text, hidden_count = self._local_stop_boundary
kept = self._count_tokens_through(context_ids, decoded_ids, stop_text)
return len(decoded_ids) - kept + hidden_count

@property
def has_pending_stop_text(self) -> bool:
return bool(self._pending_stop_text)
Expand Down Expand Up @@ -1571,8 +1604,10 @@ def _filter_stop_string_delta(
text: str,
finish_reason: str | None,
stop_reason: Any,
) -> tuple[str, bool]:
text = self._pending_stop_text + text
) -> tuple[str, int | None]:
"""Return visible text and, after a local stop, the stop's end in ``text``."""
held_text = self._pending_stop_text
text = held_text + text
self._pending_stop_text = ""

match = self._find_stop_string(text, stop_reason)
Expand All @@ -1587,16 +1622,17 @@ def _filter_stop_string_delta(
del self._pending_logprobs_content[-suppressed_count:]
self._locally_finished = True
self._local_stop_reason = matched_stop_string
return text[:retained_end], True
stop_end = match_index + len(matched_stop_string) - len(held_text)
return text[:retained_end], stop_end

if finish_reason or not text or not self._stop_strings:
return text, False
return text, None

pending_len = trailing_stop_prefix_len(text, self._stop_strings)
if pending_len:
self._pending_stop_text = text[-pending_len:]
return text[:-pending_len], False
return text, False
return text[:-pending_len], None
return text, None

def _parse_reasoning_delta(
self, delta_text: str, finish_reason: str | None
Expand Down Expand Up @@ -1669,33 +1705,45 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
stop_terminated = engine_response.get(
"stop_terminated", finish_reason == "stop"
)
generated_ids = token_ids
if stop_terminated:
raw_token_count = len(token_ids)
token_ids = self._strip_matched_stop_token_ids(list(token_ids), stop_reason)
retained_token_count = len(token_ids)
if log_probs is not None and len(log_probs) == raw_token_count:
log_probs = log_probs[:retained_token_count]
if top_logprobs is not None and len(top_logprobs) == raw_token_count:
top_logprobs = top_logprobs[:retained_token_count]

delta_text = (
context_ids = self._decode_context_ids
held_ids = list(self._pending_decode_ids)
decoded_text = (
self._incremental_decode(token_ids, flush=finish_reason is not None)
if token_ids or finish_reason is not None
else ""
)
openai_logprobs = None
# A hidden stop token is generated output: score it, but do not display it.
stop_logprobs: list[dict[str, Any]] = []
if log_probs is not None:
scored_ids = (
generated_ids if len(log_probs) == len(generated_ids) else token_ids
)
openai_logprobs = self._build_openai_logprobs(
log_probs, top_logprobs, token_ids
log_probs, top_logprobs, scored_ids
)
if openai_logprobs is not None:
self._pending_logprobs_content.extend(openai_logprobs["content"])
content = openai_logprobs["content"]
self._pending_logprobs_content.extend(content[: len(token_ids)])
stop_logprobs = content[len(token_ids) :]
self._logprob_context_ids = (self._logprob_context_ids + token_ids)[-4:]
delta_text, locally_finished = self._filter_stop_string_delta(
delta_text, finish_reason, stop_reason
delta_text, stop_end = self._filter_stop_string_delta(
decoded_text, finish_reason, stop_reason
)
if locally_finished:
if stop_end is None:
# Queued after stop-string suppression, which matches display text.
self._pending_logprobs_content.extend(stop_logprobs)
else:
finish_reason = "stop"
self._local_stop_boundary = (
context_ids,
held_ids + token_ids,
decoded_text[:stop_end],
len(generated_ids) - len(token_ids),
)

if self._fast_plain_text:
if delta_text:
Expand Down
18 changes: 18 additions & 0 deletions components/src/dynamo/frontend/sglang_processor.py
Original file line number Diff line number Diff line change
Expand Up @@ -832,6 +832,10 @@ async def _generate_and_stream(
stream_interval = self.stream_interval
prompt_logprobs = PromptLogprobsAccumulator(request)
logprobs_count = _request_logprobs_count(request)
include_completion_ids = nvext_extra_field_requested(
request, "completion_token_ids"
)
include_prompt_ids = nvext_extra_field_requested(request, "prompt_token_ids")

try:
dynamo_stream = await self.routed_engine.generate(
Expand All @@ -845,6 +849,8 @@ async def _generate_and_stream(
pending_token_ids: list[int] = []
pending_log_probs: list[float] | None = None
pending_top_logprobs: list[list[dict[str, Any]]] | None = None
# Requested generated IDs wait here while parsing emits no chunk.
completion_ids: list[int] = []
latest_usage: dict[str, Any] = {}
pending_terminal: dict[str, Any] | None = None
first_chunk = True
Expand All @@ -868,6 +874,7 @@ def flush_pending(
nonlocal pending_token_ids
nonlocal pending_log_probs
nonlocal pending_top_logprobs
nonlocal completion_ids
nonlocal latest_usage
nonlocal first_chunk
nonlocal post_proc_total_ms
Expand All @@ -889,6 +896,12 @@ def flush_pending(
t_pp0 = time.monotonic()

choice = post.process_output(mapped_response)
if include_completion_ids:
completion_ids.extend(pending_token_ids)
if post.locally_finished:
# IDs that decoding withheld before the stop are still queued.
dropped = post.post_stop_token_count()
del completion_ids[len(completion_ids) - dropped :]

if post.locally_finished and not latest_usage:
latest_usage = {
Expand Down Expand Up @@ -935,6 +948,11 @@ def flush_pending(
request, "engine_data"
):
response_nvext["engine_data"] = engine_data
if include_completion_ids:
response_nvext["completion_token_ids"] = completion_ids
completion_ids = []
if include_prompt_ids and (finish_reason or post.locally_finished):
response_nvext["prompt_token_ids"] = tokens
if response_nvext:
dynamo_out["nvext"] = response_nvext

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4757,7 +4757,9 @@ async def collect():
assert chunks[1]["nvext"]["stop_reason"] == "END"
assert "prompt_logprobs" not in chunks[1]["choices"][0]

def _run_logprob_stream(self, items, request=None):
def _run_logprob_stream(
self, items, request=None, eos_token_ids=None, prompt_ids=()
):
if request is None:
request = {"model": "test-model", "logprobs": True}
processor = SglangProcessor(
Expand All @@ -4772,19 +4774,61 @@ def _run_logprob_stream(self, items, request=None):
tokenizer=self.ByteTokenizer(),
tool_call_parser=None,
reasoning_parser=None,
eos_token_ids=eos_token_ids,
)

async def collect():
return [
item["data"]
async for item in processor._generate_and_stream(
"req-logprobs", request, {}, [], post
"req-logprobs", request, {}, list(prompt_ids), post
)
if "data" in item
]

return asyncio.run(collect())

def test_hidden_eos_keeps_score_and_requested_ids(self):
"""All 48 generated tokens keep a score and ID; only EOS text stays hidden."""
eos = 0
generated = [*b"x" * 47, eos]
logprobs = [-float(position) for position in range(1, len(generated) + 1)]
items = [
{"token_ids": [token_id], "log_probs": [logprob]}
for token_id, logprob in zip(generated, logprobs)
]
items[-1]["finish_reason"] = "stop"

chunks = self._run_logprob_stream(
items,
request={
"model": "test-model",
"logprobs": True,
"nvext": {"extra_fields": ["completion_token_ids", "prompt_token_ids"]},
},
eos_token_ids=[eos],
prompt_ids=[1, 2, 3],
)

choices = [chunk["choices"][0] for chunk in chunks]
content = "".join(choice["delta"].get("content", "") for choice in choices)
assert content == "x" * 47
assert choices[-1]["finish_reason"] == "stop"
scores = [
entry["logprob"]
for choice in choices
for entry in choice["logprobs"]["content"]
]
assert scores == logprobs
completion_ids = [
token_id
for chunk in chunks
for token_id in chunk["nvext"]["completion_token_ids"]
]
assert completion_ids == generated
prompt_ids = [chunk["nvext"].get("prompt_token_ids") for chunk in chunks]
assert prompt_ids == [None] * (len(chunks) - 1) + [[1, 2, 3]]

@pytest.mark.parametrize(
"prompt_count,logprobs,top_logprobs,expected_top_count",
[
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
---
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
title: Troubleshooting Tool Calls
subtitle: Capture raw model output with logprobs so issues can be localized
Expand All @@ -19,7 +19,13 @@ what the response will look like, so you can capture and share useful
diagnostic info.

> [!IMPORTANT]
> Recipe applies to non-streaming requests against Dynamo's OpenAI `/v1/chat/completions` endpoint. For multi-channel reasoning models (`harmony`, `kimi_k2`, `kimi_k25`, `gemma4`), the recipe recovers only the assistant-content channel; the reasoning channel is not surfaced in `logprobs.content`.
> Recipe applies to non-streaming requests against Dynamo's OpenAI `/v1/chat/completions` endpoint.
> On the V2 parser route (see [Parser Generation Routing](#parser-generation-routing)),
> `logprobs.content` lists every token entry the backend returns, in generation order, including
> reasoning, tool-call, and control tokens that the parsed message does not show. Streamed responses
> carry each backend chunk's entries once, on one chunk for that choice, so the entries need not line
> up with the text in `reasoning_content`, `content`, and `tool_calls`. This note does not describe
> the V1 route, where late entries can arrive in a final empty-`delta` chunk after `finish_reason`.
## The request

Expand Down Expand Up @@ -88,11 +94,22 @@ engine's raw token stream:
}
```

Each entry in `logprobs.content` is one generated token with its exact UTF-8
`bytes`. Concatenating those bytes in order reconstructs the raw model
output, before any tool-call parser touched it. That is the key piece for
triage: it tells us what the model actually produced, separately from what
the parser made of it.
Each entry in `logprobs.content` is one generated token and keeps the score
the backend returned for it. The `token` text and `bytes` are decoded
representations: an end-of-sequence token can appear as its literal text and
bytes, such as `<|im_end|>` with `--dyn-chat-processor sglang`, or as empty
text and no bytes, and `return_tokens_as_token_ids` can replace the text with a
`token_id:<id>` label. Without that option, concatenating the `bytes` in
order approximates the model output before any tool-call parser touched it.
It can include control-token literals such as a trailing `<|im_end|>`, omits
tokens that decode to nothing, and, because each token can be decoded on its
own, can lose spacing that the tokenizer adds only in
context. That is the key piece for triage: it tells us what the model
actually produced, separately from what the parser made of it. For an exact
replay, also request the generated token IDs with
`"nvext": {"extra_fields": ["completion_token_ids"]}` where the response
supports them (see
[NVIDIA Request Extensions](../../developer-guide/additional-resources/nvidia-request-extensions-nvext.md)).

## What to include when reporting an issue

Expand All @@ -108,8 +125,9 @@ Share these four things in the bug report or issue thread:
value if set.

With those four pieces, the Dynamo team can usually localize the bug
without standing up your model. The team will reconstruct the raw stream
from the `bytes` arrays and compare it against `message.content` and
without standing up your model. The team will rebuild the raw stream from
the per-token entries, using exact token IDs over decoded `bytes` where the
response provides them, and compare it against `message.content` and
`message.tool_calls` to decide whether the issue is in the model output,
the parser configuration, or the parser logic.

Expand Down
Loading
Loading