diff --git a/components/src/dynamo/frontend/prepost.py b/components/src/dynamo/frontend/prepost.py index 512450a987be..60b24379eba3 100644 --- a/components/src/dynamo/frontend/prepost.py +++ b/components/src/dynamo/frontend/prepost.py @@ -714,6 +714,17 @@ def __init__( self._reasoning_parser_streaming_started = False self._tool_parser_streaming_started = False + # Every ORIGINAL generated token id for this choice, in order (NVBug + # 6678449b). The count is delegated to the pinned vLLM parser's own + # `count_reasoning_tokens`, so it is never re-derived from parser output. + # Deriving it was wrong three ways: the response projections drop or + # defer it (`_suppress_reasoning_output`, and the non-streaming tool + # path's terminal-delta buffering); a per-chunk classifier over-counts a + # chunk carrying both kinds; and re-encoding decoded text drifts from the + # original ids, because detokenisation is not injective -- one generated + # id can re-encode to several. + self._reasoning_token_ids: list[int] = [] + self._control_markers = tuple( t for t in getattr(tokenizer, "all_special_tokens", ()) if t ) @@ -733,6 +744,13 @@ def __init__( self._tool_text_buffer: str | None = None self._dynamo_json_fallback_chunks: dict[int, list[str]] = {} + @property + def reasoning_token_total(self) -> int: + """Reasoning tokens for this choice, per the parser's own counter.""" + if self.reasoning_parser is None: + return 0 + return self.reasoning_parser.count_reasoning_tokens(self._reasoning_token_ids) + def _decode_dynamo_json_fallback_tool_calls( self, text: str ) -> list[dict[str, Any]]: @@ -1097,6 +1115,10 @@ def _process_non_streaming_tool_output(self, output: Any) -> dict[str, Any] | No return self._build_choice(output, delta) def process_output(self, output: Any) -> dict[str, Any] | None: + # BEFORE any branch below: every path must feed the ledger, including + # the ones that emit nothing for this chunk. + if self.reasoning_parser is not None: + self._reasoning_token_ids.extend(output.token_ids or []) if self._uses_dynamo_json_tool_call_fallback: return self._process_dynamo_json_fallback_tool_calls(output) if self._should_buffer_for_non_streaming_tool_parse(): diff --git a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py index b58888b4e878..67384313620b 100644 --- a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py +++ b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py @@ -2539,3 +2539,316 @@ def test_shared_thinking_policy(self, tokenizer, case): tool_parser_class=None, ) assert kwargs == case.expected + + +class _AccountingReasoningParser: + """Emits reasoning until a chunk contains "", then visible content.""" + + engine_based_streaming = False + + def __init__(self, tokenizer, *, chat_template_kwargs=None, model_config=None): + self.tokenizer = tokenizer + self.chat_template_kwargs = chat_template_kwargs or {} + self.model_config = model_config + + def is_reasoning_end(self, prompt_token_ids): + return False + + def adjust_initial_state_from_prompt(self, prompt_token_ids): + return None + + def extract_reasoning_streaming( + self, + previous_text, + current_text, + delta_text, + previous_token_ids, + current_token_ids, + delta_token_ids, + ): + if "" in delta_text: + return prepost_module.DeltaMessage(content=delta_text.split("")[-1]) + return prepost_module.DeltaMessage(reasoning=delta_text) + + def is_reasoning_end_streaming(self, current_token_ids, delta_token_ids): + return False + + def extract_reasoning(self, text, request=None): + if "" in text: + reasoning, _, content = text.partition("") + return reasoning, content + return text, "" + + def count_reasoning_tokens(self, token_ids): + """Everything up to and including the end marker counts as reasoning. + + The production code delegates to this, so the fake has to provide it -- + a fake that omits it inherits ReasoningParser's base, which returns 0. + """ + end = self.tokenizer.encode("", add_special_tokens=False) + end_id = end[-1] if end else None + count = 0 + for tid in token_ids: + count += 1 + if tid == end_id: + break + return count + + +class _PassthroughToolParser: + """Finds no tool calls; returns the text as content. + + The buffered non-streaming path only needs `tool_parser is not None` to engage, + so this keeps the reasoning-accounting test focused on reasoning. + """ + + def __init__(self, tokenizer, *args, **kwargs): + self.tokenizer = tokenizer + + def extract_tool_calls(self, text, request=None): + return SimpleNamespace(tools_called=False, tool_calls=[], content=text) + + +class TestReasoningTokenAccounting: + """Reasoning-token usage on the Python chat-processor path (NVBug 6678449b). + + dynamo #12181 added this only to the Rust OpenAIPreprocessor, which + `--dyn-chat-processor vllm` bypasses, so `reasoning_tokens` was absent from usage + entirely (the worker's _build_completion_usage emits no completion_tokens_details). + + The count is accumulated where the reasoning parser CLASSIFIES output, NOT where + the response is projected. The two regression tests below both reported 0 when it + was derived from the emitted choices. + """ + + @staticmethod + def _post(tokenizer, *, include_reasoning=True, stream_response=True, tools=False): + request = SimpleNamespace( + include_reasoning=include_reasoning, + tool_choice="auto" if tools else "none", + model_fields=frozenset(), + ) + return StreamingPostProcessor( + tokenizer=tokenizer, + request_for_sampling=request, + sampling_params=SamplingParams(), + prompt_token_ids=[], + tool_parser=_PassthroughToolParser(tokenizer) if tools else None, + reasoning_parser_class=_AccountingReasoningParser, + chat_template_kwargs={}, + stream_response=stream_response, + ) + + @staticmethod + def _out(text, token_ids, finish_reason=None, index=0): + return SimpleNamespace( + index=index, + text=text, + token_ids=token_ids, + finish_reason=finish_reason, + logprobs=None, + ) + + @staticmethod + def _annotator(*posts): + from dynamo.frontend.vllm_processor import _ReasoningUsageAnnotator + + return _ReasoningUsageAnnotator({i: p for i, p in enumerate(posts)}) + + # -- streaming classification ------------------------------------------ + + def test_streaming_counts_reasoning_only_not_the_answer(self, tokenizer): + """Real parser, real ids: the visible answer must not be counted.""" + from vllm.reasoning import ReasoningParserManager + + reasoning = tokenizer.encode("abc", add_special_tokens=False) + answer = tokenizer.encode("the answer", add_special_tokens=False) + native = ReasoningParserManager.get_reasoning_parser("qwen3")( + tokenizer + ).count_reasoning_tokens(reasoning + answer) + + post = self._qwen3_post(tokenizer) + for tid in reasoning + answer: + post.process_output(self._out(tokenizer.decode([tid]), [tid])) + assert post.reasoning_token_total == native + assert 0 < native < len(reasoning + answer) + + # -- P1 #1: account before the include_reasoning projection ------------ + + def test_hidden_reasoning_is_counted_when_include_reasoning_false(self, tokenizer): + """include_reasoning=false suppresses the reasoning DELTA, not the tokens. + + Observing emitted choices saw no reasoning_content here and reported 0 for a + response that really did spend tokens reasoning. + """ + post = self._post(tokenizer, include_reasoning=False) + choice = post.process_output(self._out("hidden", [1, 2, 3])) + + # The projection still hides it -- that part is intended... + assert choice is None or "reasoning_content" not in (choice.get("delta") or {}) + # ...but the tokens are accounted anyway. + assert post.reasoning_token_total == 3 + annotated = self._annotator(post).annotate({"completion_tokens": 10}) + assert annotated["completion_tokens_details"]["reasoning_tokens"] == 3 + + # -- P1 #2: survive non-streaming tool buffering ------------------------ + + def test_non_streaming_tool_buffering_preserves_count(self, tokenizer): + """The non-streaming tool path emits nothing per chunk and releases on the + terminal delta, where chunk_tokens is 0. Per-chunk observation of emitted + choices therefore saw only zeros and reported 0. + """ + post = self._post(tokenizer, stream_response=False, tools=True) + for chunk in ("think a", "think b", "think c"): + assert post.process_output(self._out(chunk, [1, 2])) is None + post.process_output(self._out("answer", [3], finish_reason="stop")) + + assert post.reasoning_token_total > 0 + annotated = self._annotator(post).annotate({"completion_tokens": 20}) + assert annotated["completion_tokens_details"]["reasoning_tokens"] > 0 + + def test_non_streaming_buffering_counts_even_when_suppressed(self, tokenizer): + """Both projections at once: buffered AND include_reasoning=false.""" + post = self._post( + tokenizer, stream_response=False, tools=True, include_reasoning=False + ) + post.process_output(self._out("thinking hard", [1, 2])) + post.process_output(self._out("answer", [3], finish_reason="stop")) + assert post.reasoning_token_total > 0 + + def test_no_reasoning_parser_counts_nothing(self, tokenizer): + """Without a reasoning parser there is nothing to delegate to -> 0. + + Note the counterpart is NOT true: with a parser, plain content is not + automatically zero. Qwen3's parser counts an unopened span as reasoning + (the model is trained to reason first), so delegating means inheriting + that contract -- which is the point, since it is what vLLM itself + reports. + """ + request = SimpleNamespace( + include_reasoning=True, tool_choice="none", model_fields=frozenset() + ) + post = StreamingPostProcessor( + tokenizer=tokenizer, + request_for_sampling=request, + sampling_params=SamplingParams(), + prompt_token_ids=[], + tool_parser=None, + reasoning_parser_class=None, + chat_template_kwargs={}, + stream_response=True, + ) + for tid in tokenizer.encode("just an answer", add_special_tokens=False): + post.process_output(self._out(tokenizer.decode([tid]), [tid])) + assert post.reasoning_token_total == 0 + + # -- the parser's OWN counter, not a re-derivation ----------------------- + + @staticmethod + def _qwen3_post(tokenizer): + from vllm.reasoning import ReasoningParserManager + + request = SimpleNamespace( + include_reasoning=True, tool_choice="none", model_fields=frozenset() + ) + return StreamingPostProcessor( + tokenizer=tokenizer, + request_for_sampling=request, + sampling_params=SamplingParams(), + prompt_token_ids=[], + tool_parser=None, + reasoning_parser_class=ReasoningParserManager.get_reasoning_parser("qwen3"), + chat_template_kwargs={}, + stream_response=True, + ) + + def test_count_matches_the_parser_for_split_marker_tokens(self, tokenizer): + """``, `abc`, `<`, `x` -- the parser counts 3. + + A per-chunk classifier that adds len(delta_token_ids) whenever the chunk + carried reasoning reported 2 here. Delegating to the parser's own + count_reasoning_tokens is what makes this exact. + """ + from vllm.reasoning import ReasoningParserManager + + ids = [ + i + for piece in ("", "abc", "<", "x") + for i in tokenizer.encode(piece, add_special_tokens=False) + ] + native = ReasoningParserManager.get_reasoning_parser("qwen3")( + tokenizer + ).count_reasoning_tokens(ids) + assert native == 3, f"fixture drift: parser now counts {native}" + + post = self._qwen3_post(tokenizer) + for tid in ids: + post.process_output(self._out(tokenizer.decode([tid]), [tid])) + assert post.reasoning_token_total == native == 3 + + def test_count_survives_non_injective_detokenisation(self, tokenizer): + """Generated id 77150 decodes to '10', which RE-ENCODES to two tokens. + + Counting by re-encoding the decoded reasoning text therefore reported 2 + for a single generated token. The ledger keeps the original ids, so the + parser sees exactly what was generated. + """ + think = tokenizer.encode("", add_special_tokens=False) + ids = think + [77150] + assert ( + len(tokenizer.encode(tokenizer.decode([77150]), add_special_tokens=False)) + == 2 + ), "fixture drift: 77150 no longer re-encodes to 2 tokens" + + from vllm.reasoning import ReasoningParserManager + + native = ReasoningParserManager.get_reasoning_parser("qwen3")( + tokenizer + ).count_reasoning_tokens(ids) + post = self._qwen3_post(tokenizer) + for tid in ids: + post.process_output(self._out(tokenizer.decode([tid]), [tid])) + assert post.reasoning_token_total == native == 1 + + # -- annotation --------------------------------------------------------- + + def test_total_sums_across_choices(self, tokenizer): + a, b = self._post(tokenizer), self._post(tokenizer) + a.process_output(self._out("x", [1, 2])) + b.process_output(self._out("y", [3, 4, 5])) + assert self._annotator(a, b).total == 5 + + def test_annotate_fills_when_absent_and_does_not_mutate_input(self, tokenizer): + post = self._post(tokenizer) + post.process_output(self._out("x", [1, 2])) + usage = {"completion_tokens": 100} + out = self._annotator(post).annotate(usage) + assert out["completion_tokens_details"]["reasoning_tokens"] == 2 + assert "completion_tokens_details" not in usage + + def test_annotate_preserves_a_positive_backend_value(self, tokenizer): + post = self._post(tokenizer) + post.process_output(self._out("x", [1, 2])) + usage = {"completion_tokens_details": {"reasoning_tokens": 7}} + annotated = self._annotator(post).annotate(usage) + assert annotated["completion_tokens_details"]["reasoning_tokens"] == 7 + + def test_annotate_replaces_a_zero_backend_value(self, tokenizer): + post = self._post(tokenizer) + post.process_output(self._out("x", [1, 2])) + usage = {"completion_tokens_details": {"reasoning_tokens": 0}} + annotated = self._annotator(post).annotate(usage) + assert annotated["completion_tokens_details"]["reasoning_tokens"] == 2 + + @pytest.mark.parametrize("backend", [-1, -100, None, "7"]) + def test_annotate_replaces_a_non_positive_backend_value(self, tokenizer, backend): + """Only a POSITIVE backend count is authoritative. + + -1 is truthy, so a plain falsiness check preserved it and reported a + negative reasoning_tokens to the client. + """ + post = self._post(tokenizer) + post.process_output(self._out("x", [1, 2])) + usage = {"completion_tokens_details": {"reasoning_tokens": backend}} + annotated = self._annotator(post).annotate(usage) + assert annotated["completion_tokens_details"]["reasoning_tokens"] == 2 diff --git a/components/src/dynamo/frontend/vllm_processor.py b/components/src/dynamo/frontend/vllm_processor.py index cb5ded3fe439..e68773687c00 100644 --- a/components/src/dynamo/frontend/vllm_processor.py +++ b/components/src/dynamo/frontend/vllm_processor.py @@ -64,6 +64,54 @@ } +class _ReasoningUsageAnnotator: + """Report reasoning-token usage on the PYTHON chat-processor path. + + NVBug 6678449b. dynamo #12181 added this, but only to the RUST OpenAIPreprocessor + (lib/llm/src/preprocessor.rs). `--dyn-chat-processor vllm` selects this Python path, + which bypasses that code entirely -- and the worker's _build_completion_usage() + emits no `completion_tokens_details` key at all, so `reasoning_tokens` is ABSENT + (not null) from usage. + + The count is NOT derived here. Each StreamingPostProcessor accumulates + `reasoning_token_total` where the reasoning parser CLASSIFIES its output; this + class only sums those totals and annotates usage. Deriving it here -- from the + choices this path emits -- is what reported zero whenever the response projection + dropped or deferred the reasoning (include_reasoning=false, and the non-streaming + tool path's terminal-delta buffering). + + Streaming classification stays chunk-granular, matching the Rust: a chunk carrying + both reasoning and visible content counts entirely as reasoning. The non-streaming + tool path has no per-chunk classification, so it counts the parsed reasoning text + exactly; that is more precise than the Rust rather than divergent from it. + A positive backend-supplied count stays authoritative. + """ + + __slots__ = ("_post_processors",) + + def __init__(self, post_processors: dict[int, StreamingPostProcessor]) -> None: + self._post_processors = post_processors + + @property + def total(self) -> int: + return sum( + post.reasoning_token_total for post in self._post_processors.values() + ) + + def annotate(self, usage: dict[str, Any]) -> dict[str, Any]: + """Return a COPY of usage carrying reasoning_tokens; never mutates the input.""" + annotated = dict(usage) + details = dict(annotated.get("completion_tokens_details") or {}) + # Only a POSITIVE backend count is authoritative. Missing, zero and + # negative all fall back to our own tally -- a negative would otherwise + # survive, since -1 is truthy. + backend = details.get("reasoning_tokens") + if not isinstance(backend, int) or backend <= 0: + details["reasoning_tokens"] = self.total + annotated["completion_tokens_details"] = details + return annotated + + def map_finish_reason(raw_reason: str | None) -> FinishReason | None: if raw_reason is None: return None @@ -838,6 +886,10 @@ async def _generate_and_stream( # content-part counts here too (else frontend metrics report zero media). input_tokens = len(tokens) cumulative_output_tokens = 0 + # Per-request reasoning-token usage (NVBug 6678449b); see + # _ReasoningUsageAnnotator. Must be per-request, never module-level. + # The counts live on the post-processors, which are per-request too. + reasoning_usage = _ReasoningUsageAnnotator(post_processors) _mm_counts, _ = extract_mm_urls(request.get("messages") or []) _mm_counts = _mm_counts or {} image_count = len(_mm_counts.get("image_url", [])) @@ -962,7 +1014,7 @@ async def _generate_and_stream( "object": "chat.completion.chunk", } if usage := engine_response.get("completion_usage"): - dynamo_out["usage"] = usage + dynamo_out["usage"] = reasoning_usage.annotate(usage) envelope["data"] = dynamo_out metrics = {