[Frontend] Support count_reasoning_tokens in the Streaming Parser Engine (#45802)

Signed-off-by: chaunceyjiang <chaunceyjiang@gmail.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
This commit is contained in:
Chauncey
2026-08-15 06:51:14 -07:00
committed by GitHub
co-authored by mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
parent 5cecfc0137
commit ac2ae8798c
16 changed files with 669 additions and 73 deletions
@@ -28,9 +28,11 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
from vllm.entrypoints.openai.chat_completion.serving import (
OpenAIServingChat,
_get_mm_token_counts,
_make_completion_tokens_details,
_make_prompt_tokens_details,
)
from vllm.entrypoints.openai.engine.protocol import (
DeltaMessage,
ErrorResponse,
RequestResponseMetadata,
)
@@ -626,6 +628,7 @@ def _build_minimal_metrics_serving_chat(
serving.response_role = "assistant"
serving.parser_cls = None
serving.enable_auto_tools = False
serving._include_reasoning_tokens_details = False
serving.enable_prompt_tokens_details = False
serving.enable_log_outputs = False
serving.enable_log_deltas = False
@@ -666,6 +669,13 @@ async def _single_request_output(
yield request_output
async def _stream_request_outputs(
*request_outputs: RequestOutput,
) -> AsyncIterator[RequestOutput]:
for request_output in request_outputs:
yield request_output
async def _collect_metrics_stream_chunks(
serving: OpenAIServingChat,
request: ChatCompletionRequest,
@@ -724,6 +734,7 @@ async def test_chat_per_request_metrics_follow_server_flag():
request_metadata=RequestResponseMetadata(request_id="chatcmpl-test-id"),
)
assert disabled_response.metrics is None
assert disabled_response.usage.completion_tokens_details is None
enabled_serving = _build_minimal_metrics_serving_chat(
enable_per_request_metrics=True
@@ -781,6 +792,55 @@ async def test_chat_streaming_metrics_ride_on_usage_chunk():
assert usage_chunks[-1]["metrics"]["time_to_first_token_ms"] == pytest.approx(500.0)
@pytest.mark.asyncio
async def test_streaming_reasoning_usage_counts_across_deltas():
serving = _build_minimal_metrics_serving_chat(enable_per_request_metrics=False)
serving._include_reasoning_tokens_details = True
serving.model_config = None
parser = MagicMock()
parser.parse_delta.side_effect = [
DeltaMessage(reasoning="reasoning"),
DeltaMessage(content="answer"),
]
parser.count_reasoning_tokens.side_effect = lambda token_ids: sum(
token_id == 20 for token_id in token_ids
)
serving.parser_cls = MagicMock(return_value=parser)
first = _make_metrics_request_output(metrics=None, token_ids=(10, 20))
first.outputs[0].text = "<think>reasoning"
first.outputs[0].finish_reason = None
second = _make_metrics_request_output(metrics=None, token_ids=(11, 30))
second.outputs[0].text = "</think>answer"
request = ChatCompletionRequest(
model="test-model",
messages=[{"role": "user", "content": "Test prompt"}],
max_tokens=10,
stream=True,
stream_options={"include_usage": True},
)
chunks: list[dict[str, Any]] = []
async for line in serving.chat_completion_stream_generator(
request,
_stream_request_outputs(first, second),
"chatcmpl-test-id",
"test-model",
conversation=[{"role": "user", "content": "Test"}],
tokenizer=MagicMock(),
request_metadata=RequestResponseMetadata(request_id="chatcmpl-test-id"),
):
payload = line.removeprefix("data: ").strip()
if payload != "[DONE]":
chunks.append(json.loads(payload))
usage_chunks = [chunk for chunk in chunks if chunk.get("usage")]
assert usage_chunks[-1]["usage"]["completion_tokens_details"] == {
"reasoning_tokens": 1
}
@dataclass
class MockEngine:
model_config: MockModelConfig = field(default_factory=MockModelConfig)
@@ -810,6 +870,7 @@ async def _async_serving_chat_init():
def test_async_serving_chat_init():
serving_completion = asyncio.run(_async_serving_chat_init())
assert serving_completion.chat_template == CHAT_TEMPLATE
assert serving_completion._include_reasoning_tokens_details is False
def test_mm_prompt_tokens_details():
@@ -850,6 +911,10 @@ def test_mm_prompt_tokens_details():
assert details.multimodal_tokens == {"image": 600, "video": 1200}
def test_completion_tokens_details():
assert _make_completion_tokens_details(7).reasoning_tokens == 7
@pytest.mark.asyncio
async def test_serving_chat_returns_correct_model_name():
mock_engine = MagicMock(spec=AsyncLLM)
@@ -360,6 +360,10 @@ async def test_reasoning_tokens_counted_for_text_reasoning_model(monkeypatch):
def get_vocab(self):
return self._vocab
def decode(self, token_ids):
id_to_token = {v: k for k, v in self._vocab.items()}
return "".join(id_to_token.get(token_id, "x") for token_id in token_ids)
# Force non-harmony, SimpleContext path
monkeypatch.setattr(envs, "VLLM_USE_EXPERIMENTAL_PARSER_CONTEXT", False)
+51
View File
@@ -2,6 +2,7 @@
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Tests for the streaming parser engine core pipeline."""
from dataclasses import replace
from unittest.mock import MagicMock
import pytest
@@ -71,6 +72,16 @@ def _think_config() -> ParserEngineConfig:
)
def _token_think_config() -> ParserEngineConfig:
return replace(
_think_config(),
token_id_terminals={
"THINK_START": "<think>",
"THINK_END": "</think>",
},
)
class TestNonStreaming:
def test_plain_text(self):
engine = StreamingParserEngine(_hermes_config(), tokenizer=None)
@@ -441,6 +452,46 @@ class TestTokenIdFiltering:
assert sum(1 for e in all_events if e.type == EventType.TOOL_CALL_END) == 1
class TestReasoningTokenCounts:
def test_counts_reasoning_and_excludes_boundaries_and_content(self):
engine = StreamingParserEngine(_token_think_config(), _make_think_tokenizer())
events = engine.feed(
"<think>tok1tok2</think>tok3",
[_START_ID, 1, 2, _END_ID, 3],
)
events.extend(engine.finish())
reasoning = [e for e in events if e.type == EventType.REASONING_CHUNK]
assert "".join(e.value for e in reasoning) == "tok1tok2"
assert sum(e.token_count for e in reasoning) == 2
assert engine.reasoning_token_count == 2
def test_counts_tokens_after_final_deferred_start_terminal(self):
engine = StreamingParserEngine(_token_think_config(), _make_think_tokenizer())
assert engine.feed("", [_START_ID, 1, 2]) == []
events = engine.feed("<think>tok1tok2", [])
reasoning = [e for e in events if e.type == EventType.REASONING_CHUNK]
assert "".join(e.value for e in reasoning) == "tok1tok2"
assert sum(e.token_count for e in reasoning) == 2
assert engine.reasoning_token_count == 2
def test_deferred_reasoning_tokens_stay_before_next_end_terminal(self):
engine = StreamingParserEngine(_token_think_config(), _make_think_tokenizer())
assert engine.feed("", [_START_ID, 1, 2]) == []
events = engine.feed("<think>tok1tok2</think>tok3", [_END_ID, 3])
assert (
sum(e.token_count for e in events if e.type == EventType.REASONING_CHUNK)
== 2
)
assert sum(e.token_count for e in events if e.type == EventType.TEXT_CHUNK) == 1
assert engine.reasoning_token_count == 2
def _func_prefix_config() -> ParserEngineConfig:
"""Config mixing token-ID terminals (TOOL_START/END) with
text-only terminals (FUNC_PREFIX) and fallback transitions."""
+124
View File
@@ -33,6 +33,7 @@ from vllm.parser.engine.parser_engine_config import (
ParserState,
Transition,
)
from vllm.parser.parser_manager import ParserManager
# ── Shared test configs ──────────────────────────────────────────────
@@ -892,6 +893,129 @@ class _CombinedDelegating(DelegatingParser):
tool_parser_cls = _CombinedToolAdapter
def test_parser_manager_uses_shared_engine_directly(monkeypatch):
monkeypatch.setattr(
ParserManager,
"get_reasoning_parser",
classmethod(lambda cls, name: _CombinedReasoningAdapter),
)
monkeypatch.setattr(
ParserManager,
"get_tool_parser",
classmethod(lambda cls, name, enabled, model: _CombinedToolAdapter),
)
parser_cls = ParserManager.get_parser(
tool_parser_name="combined",
reasoning_parser_name="combined",
enable_auto_tools=True,
)
assert parser_cls is not None
assert parser_cls is _CombinedTestEngine
parser = parser_cls(make_mock_tokenizer(_VOCAB))
request = _make_delegating_request()
reasoning, content, _ = parser.parse(
"ab</think>c",
request,
model_output_token_ids=[ord("a"), ord("b"), 201, ord("c")],
)
assert reasoning == "ab"
assert content == "c"
assert parser.count_reasoning_tokens([]) == 2
def test_parser_manager_preserves_reasoning_only_adapter(monkeypatch):
monkeypatch.setattr(
ParserManager,
"get_reasoning_parser",
classmethod(lambda cls, name: _CombinedReasoningAdapter),
)
monkeypatch.setattr(
ParserManager,
"get_tool_parser",
classmethod(lambda cls, name, enabled, model: None),
)
parser_cls = ParserManager.get_parser(reasoning_parser_name="combined")
assert parser_cls is not None
parser = parser_cls(make_mock_tokenizer(_VOCAB))
assert parser.reasoning_parser is not None
assert parser.tool_parser is None
reasoning, content, _ = parser.parse(
'ab</think><tool_call>{"name":"h","arguments":{}}</tool_call>',
_make_delegating_request(),
model_output_token_ids=[ord("a"), ord("b"), 201],
)
assert reasoning == "ab"
assert content == '<tool_call>{"name":"h","arguments":{}}</tool_call>'
assert parser.count_reasoning_tokens([ord("a"), ord("b"), 201]) == 2
def test_parser_manager_preserves_tool_only_adapter(monkeypatch):
monkeypatch.setattr(
ParserManager,
"get_reasoning_parser",
classmethod(lambda cls, name: None),
)
monkeypatch.setattr(
ParserManager,
"get_tool_parser",
classmethod(lambda cls, name, enabled, model: _CombinedToolAdapter),
)
parser_cls = ParserManager.get_parser(
tool_parser_name="combined", enable_auto_tools=True
)
assert parser_cls is not None
parser = parser_cls(make_mock_tokenizer(_VOCAB))
assert parser.reasoning_parser is None
assert parser.tool_parser is not None
reasoning, content, tool_calls = parser.parse(
"<think>ab</think>c",
_make_delegating_request(),
model_output_token_ids=[200, ord("a"), ord("b"), 201, ord("c")],
)
assert reasoning is None
assert content == "<think>ab</think>c"
assert tool_calls == []
def test_parser_manager_rejects_non_engine_adapter_metadata():
class TraditionalParser:
pass
class InvalidEngineAdapter:
_parser_engine_cls = TraditionalParser
assert ParserManager._get_parser_engine_cls(TraditionalParser) is None
assert ParserManager._get_parser_engine_cls(InvalidEngineAdapter) is None
assert (
ParserManager._get_parser_engine_cls(_CombinedReasoningAdapter)
is _CombinedTestEngine
)
def test_reasoning_adapter_counts_after_final_non_streaming_parse():
parser = _CombinedReasoningAdapter(make_mock_tokenizer(_VOCAB))
request = _make_delegating_request()
token_ids = [ord("a"), ord("b"), 201, ord("c")]
parser.extract_reasoning_streaming(
"",
"ab</think>c",
"ab</think>c",
[],
token_ids,
token_ids,
)
parser.extract_reasoning("ab</think>c", request)
assert parser.count_reasoning_tokens(token_ids) == 2
def _make_delegating_request():
req = MagicMock(spec=ChatCompletionRequest)
req.tools = []
+20 -1
View File
@@ -163,9 +163,28 @@ class TestHoldbackTextRecovery:
assert len(result) == 0
flushed = scanner.flush_pending()
assert len(flushed) == 1
assert len(flushed) == 2
assert isinstance(flushed[0], PreLexedTerminal)
assert flushed[0].terminal == "TOOL_START"
assert isinstance(flushed[1], TextChunk)
assert flushed[1].text == ""
assert flushed[1].token_count == 2
def test_deferred_terminal_preserves_trailing_token_count(self, tokenizer):
tokenizer.decode.side_effect = lambda ids: {
CHANNEL_START_ID: CHANNEL_START,
201: "alpha",
202: "beta",
}[ids[0]]
scanner = TokenIDScanner({CHANNEL_START_ID: "THINK_START"}, tokenizer)
assert scanner.scan("", [CHANNEL_START_ID, 201, 202]) == []
result = scanner.scan(f"{CHANNEL_START}alphabeta", [])
assert isinstance(result[0], PreLexedTerminal)
assert isinstance(result[1], TextChunk)
assert result[1].text == "alphabeta"
assert result[1].token_count == 2
def test_holdback_before_start_tag(self, scanner):
result = scanner.scan(
@@ -36,6 +36,7 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
ChatMessage,
)
from vllm.entrypoints.openai.engine.protocol import (
CompletionTokenUsageInfo,
DeltaMessage,
ErrorResponse,
FunctionCall,
@@ -105,6 +106,12 @@ def _make_prompt_tokens_details(
)
def _make_completion_tokens_details(
reasoning_tokens: int,
) -> CompletionTokenUsageInfo:
return CompletionTokenUsageInfo(reasoning_tokens=reasoning_tokens)
class OpenAIServingChat(GenerateBaseServing):
def __init__(
self,
@@ -146,6 +153,7 @@ class OpenAIServingChat(GenerateBaseServing):
self.enable_log_deltas = enable_log_deltas
self.enable_auto_tools: bool = enable_auto_tools
self._include_reasoning_tokens_details = bool(reasoning_parser)
self.parser_cls = ParserManager.get_parser(
tool_parser_name=tool_parser,
reasoning_parser_name=reasoning_parser,
@@ -439,6 +447,9 @@ class OpenAIServingChat(GenerateBaseServing):
# Send response for each token for each request.n (index)
num_choices = 1 if request.n is None else request.n
previous_num_tokens = [0] * num_choices
# TODO: Remove once all reasoning parsers use the Parser Engine.
generated_token_ids: list[list[int]] = [[] for _ in range(num_choices)]
previous_reasoning_tokens = [0] * num_choices
finish_reason_sent = [False] * num_choices
num_prompt_tokens = 0
num_cached_tokens = None
@@ -537,6 +548,11 @@ class OpenAIServingChat(GenerateBaseServing):
prompt_tokens=num_prompt_tokens,
completion_tokens=0,
total_tokens=num_prompt_tokens,
completion_tokens_details=(
_make_completion_tokens_details(0)
if self._include_reasoning_tokens_details
else None
),
)
data = chunk.model_dump_json(exclude_unset=True)
@@ -573,6 +589,11 @@ class OpenAIServingChat(GenerateBaseServing):
prompt_tokens=num_prompt_tokens,
completion_tokens=0,
total_tokens=num_prompt_tokens,
completion_tokens_details=(
_make_completion_tokens_details(0)
if self._include_reasoning_tokens_details
else None
),
)
data = chunk.model_dump_json(exclude_unset=True)
@@ -631,6 +652,11 @@ class OpenAIServingChat(GenerateBaseServing):
# set the previous values for the next iteration
previous_num_tokens[i] += len(output.token_ids)
if parser is not None:
generated_token_ids[i].extend(output.token_ids)
previous_reasoning_tokens[i] = parser.count_reasoning_tokens(
tuple(generated_token_ids[i])
)
# if the message delta is None (e.g. because it was a
# "control token" for tool calls or the parser otherwise
@@ -756,6 +782,13 @@ class OpenAIServingChat(GenerateBaseServing):
prompt_tokens=num_prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=num_prompt_tokens + completion_tokens,
completion_tokens_details=(
_make_completion_tokens_details(
previous_reasoning_tokens[i]
)
if self._include_reasoning_tokens_details
else None
),
)
data = chunk.model_dump_json(exclude_unset=True)
@@ -769,6 +802,11 @@ class OpenAIServingChat(GenerateBaseServing):
prompt_tokens=num_prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=num_prompt_tokens + completion_tokens,
completion_tokens_details=_make_completion_tokens_details(
sum(previous_reasoning_tokens)
)
if self._include_reasoning_tokens_details
else None,
)
final_usage.prompt_tokens_details = _make_prompt_tokens_details(
self.enable_prompt_tokens_details,
@@ -813,6 +851,11 @@ class OpenAIServingChat(GenerateBaseServing):
prompt_tokens=num_prompt_tokens,
completion_tokens=num_completion_tokens,
total_tokens=num_prompt_tokens + num_completion_tokens,
completion_tokens_details=_make_completion_tokens_details(
sum(previous_reasoning_tokens)
)
if self._include_reasoning_tokens_details
else None,
)
# Log complete streaming response if output logging is enabled
@@ -871,6 +914,7 @@ class OpenAIServingChat(GenerateBaseServing):
)
choices: list[ChatCompletionResponseChoice] = []
total_reasoning_tokens = 0
role = self.get_chat_request_role(request)
tool_parser_cls = (
@@ -908,6 +952,7 @@ class OpenAIServingChat(GenerateBaseServing):
suppress_metadata = not request.include_reasoning and parser is not None
if not request.include_reasoning:
reasoning = None
total_reasoning_tokens += parser.count_reasoning_tokens(token_ids)
if suppress_metadata:
logprobs = None
else:
@@ -1058,6 +1103,11 @@ class OpenAIServingChat(GenerateBaseServing):
prompt_tokens=num_prompt_tokens,
completion_tokens=num_generated_tokens,
total_tokens=num_prompt_tokens + num_generated_tokens,
completion_tokens_details=_make_completion_tokens_details(
total_reasoning_tokens
)
if self._include_reasoning_tokens_details
else None,
)
usage.prompt_tokens_details = _make_prompt_tokens_details(
self.enable_prompt_tokens_details,
@@ -117,11 +117,16 @@ class PromptTokenUsageInfo(OpenAIBaseModel):
request has no multimodal input."""
class CompletionTokenUsageInfo(OpenAIBaseModel):
reasoning_tokens: int = 0
class UsageInfo(OpenAIBaseModel):
prompt_tokens: int = 0
total_tokens: int = 0
completion_tokens: int | None = 0
prompt_tokens_details: PromptTokenUsageInfo | None = None
completion_tokens_details: CompletionTokenUsageInfo | None = None
class PerRequestTimingMetrics(OpenAIBaseModel):
+6 -8
View File
@@ -487,6 +487,7 @@ class OpenAIServingResponses(GenerateBaseServing):
response_parser=response_parser,
)
reasoning_parser_kwargs = None
if (
context.response_parser is not None
and context.response_parser.reasoning_parser is not None
@@ -518,9 +519,7 @@ class OpenAIServingResponses(GenerateBaseServing):
priority=self._get_priority(request, raw_request),
trace_headers=trace_headers,
session_id=session_id,
reasoning_parser_kwargs=reasoning_parser_kwargs
if self.parser and self.parser.reasoning_parser_cls is not None
else None,
reasoning_parser_kwargs=reasoning_parser_kwargs,
)
generators.append(generator)
@@ -873,6 +872,8 @@ class OpenAIServingResponses(GenerateBaseServing):
if final_output.finish_reason == "length":
status = "incomplete"
# TODO: Build final response items from the accumulated streaming
# parser results instead of reparsing the complete output.
output = self._make_response_output_items(
request,
final_output,
@@ -901,13 +902,10 @@ class OpenAIServingResponses(GenerateBaseServing):
num_reasoning_tokens == 0
and isinstance(context, (SimpleContext, ParsableContext))
and context.response_parser is not None
and context.response_parser.reasoning_parser is not None
):
accumulated = getattr(context, "_accumulated_token_ids", []) or []
num_reasoning_tokens = (
context.response_parser.reasoning_parser.count_reasoning_tokens(
accumulated
)
num_reasoning_tokens = context.response_parser.count_reasoning_tokens(
accumulated
)
usage = ResponseUsage(
+10
View File
@@ -367,6 +367,10 @@ class Parser:
tool call extraction via internal stream state.
"""
def count_reasoning_tokens(self, token_ids: Sequence[int]) -> int:
"""Return the number of reasoning tokens in generated token IDs."""
return 0
class DelegatingParser(Parser):
"""
@@ -939,6 +943,12 @@ class DelegatingParser(Parser):
return delta_message
def count_reasoning_tokens(self, token_ids: Sequence[int]) -> int:
"""Count reasoning tokens through the configured reasoning parser."""
if self._reasoning_parser is None:
return 0
return self._reasoning_parser.count_reasoning_tokens(token_ids)
def _flush_engine_parsers(
self, delta_message: DeltaMessage | None
) -> DeltaMessage | None:
+19 -1
View File
@@ -47,6 +47,11 @@ class ParserEngineReasoningAdapter(ReasoningParser):
def __init__(self, tokenizer: TokenizerLike, *args, **kwargs) -> None:
super().__init__(tokenizer, *args, **kwargs)
self._parser_engine = self._parser_engine_cls(tokenizer, **kwargs) # type: ignore[call-arg]
self._parser_engine_kwargs = kwargs
self._counting_parser_engine: ParserEngine | None = None
# TODO: Remove once Responses finalization reuses accumulated streaming
# parser results instead of reparsing the complete output.
self._streaming_count_valid = False
@contextmanager
def _skip_tool_parsing(self) -> Iterator[None]:
@@ -71,6 +76,7 @@ class ParserEngineReasoningAdapter(ReasoningParser):
model_output: str,
request: ChatCompletionRequest | ResponsesRequest,
) -> tuple[str | None, str | None]:
self._streaming_count_valid = False
with self._skip_tool_parsing():
return self._parser_engine.extract_reasoning(model_output, request)
@@ -83,6 +89,7 @@ class ParserEngineReasoningAdapter(ReasoningParser):
current_token_ids: Sequence[int],
delta_token_ids: Sequence[int],
) -> DeltaMessage | None:
self._streaming_count_valid = True
with self._skip_tool_parsing():
return self._parser_engine.extract_reasoning_streaming(
previous_text,
@@ -122,7 +129,18 @@ class ParserEngineReasoningAdapter(ReasoningParser):
return self._parser_engine.get_streaming_fallback_content(text, request)
def count_reasoning_tokens(self, token_ids: Sequence[int]) -> int:
return self._parser_engine.count_reasoning_tokens(token_ids)
if self._streaming_count_valid:
return self._parser_engine.count_reasoning_tokens(token_ids)
if not token_ids:
return 0
if self._counting_parser_engine is None:
self._counting_parser_engine = self._parser_engine_cls(
self.model_tokenizer, **self._parser_engine_kwargs
) # type: ignore[call-arg]
self._counting_parser_engine._single_pass_parse(
self.model_tokenizer.decode(token_ids), token_ids
)
return self._counting_parser_engine.count_reasoning_tokens(token_ids)
class ParserEngineToolAdapter(ToolParser):
+1
View File
@@ -24,3 +24,4 @@ class SemanticEvent:
type: EventType
value: str = ""
tool_index: int = -1
token_count: int = 0
+81 -8
View File
@@ -24,6 +24,7 @@ class TerminalDef:
class LexToken:
terminal: str
value: str
token_count: int = 0
class LexerShape:
@@ -104,6 +105,7 @@ class IncrementalLexer:
self.terminals = shape.terminals
self.content_terminal = content_terminal
self.buffer = ""
self._token_counts: list[int] = []
self._literal_strings = shape.literal_strings
self._max_literal_len = shape.max_literal_len
@@ -114,15 +116,23 @@ class IncrementalLexer:
def reset(self) -> None:
self.buffer = ""
self._token_counts.clear()
def feed(self, text: str) -> list[LexToken]:
def feed(
self,
text: str,
token_texts: tuple[str, ...] = (),
token_count: int = 0,
) -> list[LexToken]:
char_token_counts = self._char_token_counts(text, token_texts, token_count)
if not self.buffer and self._has_only_literals and self._literal_first_chars:
for ch in text:
if ch in self._literal_first_chars:
break
else:
return [LexToken(self.content_terminal, text)]
return [LexToken(self.content_terminal, text, sum(char_token_counts))]
self.buffer += text
self._token_counts.extend(char_token_counts)
return self._drain()
def flush(self) -> list[LexToken]:
@@ -130,10 +140,46 @@ class IncrementalLexer:
if self.buffer:
tokens.extend(self._drain(final=True))
if self.buffer:
tokens.append(LexToken(self.content_terminal, self.buffer))
tokens.append(
LexToken(self.content_terminal, self.buffer, sum(self._token_counts))
)
self.buffer = ""
self._token_counts.clear()
return tokens
@staticmethod
def _char_token_counts(
text: str,
token_texts: tuple[str, ...],
token_count: int,
) -> list[int]:
counts = [0] * len(text)
if not text:
return counts
if token_texts:
pos = 0
assigned = 0
for token_text in token_texts:
if not token_text:
continue
found = text.find(token_text, pos)
if found < 0:
continue
counts[found] += 1
assigned += 1
pos = found + len(token_text)
missing = token_count - assigned
if missing > 0:
counts[0] += missing
elif token_count:
counts[0] = token_count
return counts
def _pop_token_count(self, length: int) -> int:
token_count = sum(self._token_counts[:length])
del self._token_counts[:length]
return token_count
def _drain(self, *, final: bool = False) -> list[LexToken]:
tokens: list[LexToken] = []
first_chars = self._literal_first_chars
@@ -150,8 +196,11 @@ class IncrementalLexer:
has_potential = True
break
if not has_potential:
tokens.append(LexToken(content_terminal, self.buffer))
tokens.append(
LexToken(content_terminal, self.buffer, sum(self._token_counts))
)
self.buffer = ""
self._token_counts.clear()
break
best_match: tuple[str, str, int] | None = None
@@ -175,7 +224,13 @@ class IncrementalLexer:
longer_match = True
break
if not longer_match:
tokens.append(LexToken(best_match[0], best_match[1]))
tokens.append(
LexToken(
best_match[0],
best_match[1],
self._pop_token_count(best_match[2]),
)
)
self.buffer = self.buffer[best_match[2] :]
continue
break
@@ -183,15 +238,33 @@ class IncrementalLexer:
break
if best_match is not None:
tokens.append(LexToken(best_match[0], best_match[1]))
tokens.append(
LexToken(
best_match[0],
best_match[1],
self._pop_token_count(best_match[2]),
)
)
self.buffer = self.buffer[best_match[2] :]
else:
content_end = self._find_content_boundary()
if content_end > 0:
tokens.append(LexToken(content_terminal, self.buffer[:content_end]))
tokens.append(
LexToken(
content_terminal,
self.buffer[:content_end],
self._pop_token_count(content_end),
)
)
self.buffer = self.buffer[content_end:]
else:
tokens.append(LexToken(content_terminal, self.buffer[0]))
tokens.append(
LexToken(
content_terminal,
self.buffer[0],
self._pop_token_count(1),
)
)
self.buffer = self.buffer[1:]
return tokens
+2 -17
View File
@@ -625,23 +625,8 @@ class ParserEngine(Parser):
return None
def count_reasoning_tokens(self, token_ids: Sequence[int]) -> int:
start_id = self._reasoning_start_token_id
end_id = self._reasoning_end_token_id
if start_id is None or end_id is None:
return 0
count = 0
depth = 0
for token_id in token_ids:
if token_id == start_id:
depth += 1
continue
if token_id == end_id:
if depth > 0:
depth -= 1
continue
if depth > 0:
count += 1
return count
"""Return reasoning tokens observed by the parser engine so far."""
return self._engine.reasoning_token_count
# ── Single-pass parse helper ────────────────────────────────────────
+80 -17
View File
@@ -167,6 +167,17 @@ class StreamingParserEngine:
self.skip_tool_parsing = False
self.reset(initial_state=initial_state)
@property
def reasoning_token_count(self) -> int:
return self._reasoning_token_count
def _record_reasoning_tokens(self, events: Sequence[SemanticEvent]) -> None:
self._reasoning_token_count += sum(
event.token_count
for event in events
if event.type == EventType.REASONING_CHUNK
)
def _reset_args_state(self) -> None:
self._args_buffer: str = ""
self._args_safe_end: int = 0
@@ -186,6 +197,7 @@ class StreamingParserEngine:
)
self.tool_index = -1
self._ever_had_token_ids = False
self._reasoning_token_count = 0
# DO NOT reset skip_tool_parsing here — callers set it before
# calling methods that trigger reset() (e.g. extract_reasoning),
# and clearing it silently breaks non-streaming tool-call-as-
@@ -193,6 +205,7 @@ class StreamingParserEngine:
self._scanner.reset()
self._lexer.reset()
self._message_header_buffer = ""
self._message_header_token_count = 0
self._in_skipped_tool_span = False
self._reset_args_state()
@@ -218,18 +231,30 @@ class StreamingParserEngine:
has_special = True
break
if not has_special:
return self._emit_for_state(delta_text)
events = self._emit_for_state(
delta_text, token_count=len(delta_token_ids)
)
self._record_reasoning_tokens(events)
return events
scanner_items = self._scanner.scan(delta_text, delta_token_ids)
if len(scanner_items) == 1 and isinstance(scanner_items[0], TextChunk):
lex_tokens = self._lexer.feed(scanner_items[0].text)
item = scanner_items[0]
lex_tokens = self._lexer.feed(item.text, item.token_texts, item.token_count)
if len(lex_tokens) == 1 and lex_tokens[0].terminal == CONTENT_TERMINAL:
text = lex_tokens[0].value
return self._emit_for_state(text)
return self._process_lex_tokens(lex_tokens)
events = self._emit_for_state(
lex_tokens[0].value,
token_count=lex_tokens[0].token_count,
)
else:
events = self._process_lex_tokens(lex_tokens)
self._record_reasoning_tokens(events)
return events
return self._process_scanner_items(scanner_items)
events = self._process_scanner_items(scanner_items)
self._record_reasoning_tokens(events)
return events
def _process_scanner_items(
self, items: Sequence[LexerInput]
@@ -240,7 +265,18 @@ class StreamingParserEngine:
events.extend(self._process_lex_tokens(self._lexer.flush()))
events.extend(self._on_terminal(item.terminal, item.text))
elif isinstance(item, TextChunk):
events.extend(self._process_lex_tokens(self._lexer.feed(item.text)))
if not item.text and item.token_count:
events.extend(
self._emit_for_state("", token_count=item.token_count)
)
else:
events.extend(
self._process_lex_tokens(
self._lexer.feed(
item.text, item.token_texts, item.token_count
)
)
)
return events
def finish(self) -> list[SemanticEvent]:
@@ -285,11 +321,14 @@ class StreamingParserEngine:
EventType.TEXT_CHUNK,
value=self._message_header_buffer,
tool_index=self.tool_index,
token_count=self._message_header_token_count,
)
)
self._message_header_buffer = ""
self._message_header_token_count = 0
self.state = ParserState.CONTENT
self._record_reasoning_tokens(events)
return events
def parse_complete(self, text: str) -> list[SemanticEvent]:
@@ -303,9 +342,11 @@ class StreamingParserEngine:
strict = self._token_id_terminal_names if self._ever_had_token_ids else None
for tok in tokens:
if tok.terminal == CONTENT_TERMINAL or (strict and tok.terminal in strict):
events.extend(self._on_content(tok.value))
events.extend(self._on_content(tok.value, tok.token_count))
else:
events.extend(self._on_terminal(tok.terminal, tok.value))
events.extend(
self._on_terminal(tok.terminal, tok.value, tok.token_count)
)
return events
_TOOL_STATES = frozenset(
@@ -317,7 +358,9 @@ class StreamingParserEngine:
}
)
def _on_terminal(self, terminal: str, value: str) -> list[SemanticEvent]:
def _on_terminal(
self, terminal: str, value: str, token_count: int = 0
) -> list[SemanticEvent]:
key = (self.state, terminal)
transition = self.config.transitions.get(key)
@@ -327,7 +370,7 @@ class StreamingParserEngine:
# The projected skip state may not define the wrapper closer.
if self.skip_tool_parsing and terminal in self._tool_exit_terminals:
self._in_skipped_tool_span = False
return self._emit_for_state(value)
return self._emit_for_state(value, token_count)
if self.skip_tool_parsing and terminal in self._tool_terminals:
# Inkling reuses one terminal for tool, text, and reasoning exits.
@@ -345,6 +388,7 @@ class StreamingParserEngine:
leaving_message_header = self.state == ParserState.MESSAGE_HEADER
if leaving_message_header:
self._message_header_buffer = ""
self._message_header_token_count = 0
# A tool terminal that implicitly ends reasoning must report
# that even from the header state, or the reasoning pass never
# hands the block to the tool pass.
@@ -381,13 +425,14 @@ class StreamingParserEngine:
return []
if transition.skip_in_token_id_mode and self._ever_had_token_ids:
return self._emit_for_state(value)
return self._emit_for_state(value, token_count)
return self._apply_transition(transition, value)
return self._apply_transition(transition, value, token_count)
def _emit_for_state(self, text: str) -> list[SemanticEvent]:
def _emit_for_state(self, text: str, token_count: int = 0) -> list[SemanticEvent]:
if self.state == ParserState.MESSAGE_HEADER:
self._message_header_buffer += text
self._message_header_token_count += token_count
return []
if self.state == ParserState.TOOL_ARGS:
if self.config.tool_args_json:
@@ -397,26 +442,36 @@ class StreamingParserEngine:
EventType.ARG_VALUE_CHUNK,
value=text,
tool_index=self.tool_index,
token_count=token_count,
)
]
content_type = self.config.content_events.get(self.state)
if content_type is not None:
return [SemanticEvent(content_type, value=text, tool_index=self.tool_index)]
return [
SemanticEvent(
content_type,
value=text,
tool_index=self.tool_index,
token_count=token_count,
)
]
return []
def _on_content(self, text: str) -> list[SemanticEvent]:
def _on_content(self, text: str, token_count: int = 0) -> list[SemanticEvent]:
if not text:
return []
return self._emit_for_state(text)
return self._emit_for_state(text, token_count)
def _apply_transition(
self,
transition: Transition,
value: str,
token_count: int = 0,
) -> list[SemanticEvent]:
events: list[SemanticEvent] = []
previous_state = self.state
message_header = ""
message_header_token_count = 0
if (
self.state == ParserState.TOOL_ARGS
@@ -434,7 +489,9 @@ class StreamingParserEngine:
if previous_state == ParserState.MESSAGE_HEADER:
message_header = self._message_header_buffer
message_header_token_count = self._message_header_token_count
self._message_header_buffer = ""
self._message_header_token_count = 0
self.state = transition.next_state
@@ -454,6 +511,12 @@ class StreamingParserEngine:
event_type,
value=event_value,
tool_index=self.tool_index,
token_count=(
message_header_token_count
if previous_state == ParserState.MESSAGE_HEADER
and event_type == EventType.TEXT_CHUNK
else token_count
),
)
)
+121 -17
View File
@@ -14,6 +14,8 @@ DROP_TERMINAL = "__DROP__"
@dataclass(slots=True)
class TextChunk:
text: str
token_texts: tuple[str, ...] = ()
token_count: int = 0
@dataclass(slots=True)
@@ -49,18 +51,30 @@ class TokenIDScanner:
self.tokenizer = tokenizer
self._token_text_cache: dict[int, str] = {}
self._deferred_terminals: list[PreLexedTerminal] = []
self._deferred_prefix_token_counts: list[int] = []
self._deferred_trailing_token_count = 0
self._deferred_post_text: str = ""
def reset(self) -> None:
"""Clear mutable state for reuse. Preserves the token text cache."""
self._deferred_terminals.clear()
self._deferred_prefix_token_counts.clear()
self._deferred_trailing_token_count = 0
self._deferred_post_text = ""
def _decode_token(self, token_id: int) -> str:
if token_id not in self._token_text_cache:
self._token_text_cache[token_id] = self.tokenizer.decode([token_id])
token_text = self.tokenizer.decode([token_id])
self._token_text_cache[token_id] = (
token_text if isinstance(token_text, str) else ""
)
return self._token_text_cache[token_id]
def _decode_tokens(self, token_ids: Sequence[int]) -> tuple[str, ...]:
if self.tokenizer is None:
return ()
return tuple(self._decode_token(tid) for tid in token_ids)
_EMPTY: tuple[LexerInput, ...] = ()
def scan(
@@ -70,13 +84,23 @@ class TokenIDScanner:
) -> Sequence[LexerInput]:
prefix_items: list[LexerInput] = []
effective_text = delta_text
deferred_trailing_count = 0
if self._deferred_terminals:
prefix_items, effective_text = self._resolve_deferred(delta_text)
prefix_items, effective_text, deferred_trailing_count = (
self._resolve_deferred(delta_text, len(delta_token_ids))
)
if not self.token_id_to_terminal:
if effective_text:
prefix_items.append(TextChunk(effective_text))
token_texts = self._decode_tokens(delta_token_ids)
prefix_items.append(
TextChunk(
effective_text,
token_texts,
deferred_trailing_count + len(token_texts),
)
)
return prefix_items
has_special = False
@@ -89,14 +113,23 @@ class TokenIDScanner:
if not has_special:
if effective_text:
if not prefix_items:
return [TextChunk(effective_text)]
prefix_items.append(TextChunk(effective_text))
token_texts = self._decode_tokens(delta_token_ids)
return [TextChunk(effective_text, token_texts, len(token_texts))]
token_texts = self._decode_tokens(delta_token_ids)
prefix_items.append(
TextChunk(
effective_text,
token_texts,
deferred_trailing_count + len(token_texts),
)
)
return prefix_items or self._EMPTY
token_texts = [self._decode_token(tid) for tid in delta_token_ids]
decoded_token_texts = [self._decode_token(tid) for tid in delta_token_ids]
results: list[LexerInput] = []
text_accum: list[str] = []
token_text_accum: list[str] = []
for idx, tid in enumerate(delta_token_ids):
terminal = self.token_id_to_terminal.get(tid)
@@ -104,16 +137,31 @@ class TokenIDScanner:
if text_accum:
joined = "".join(text_accum)
if joined:
results.append(TextChunk(joined))
results.append(
TextChunk(
joined,
tuple(token_text_accum),
len(token_text_accum),
)
)
text_accum.clear()
results.append(PreLexedTerminal(terminal, tid, token_texts[idx]))
token_text_accum.clear()
results.append(
PreLexedTerminal(terminal, tid, decoded_token_texts[idx])
)
else:
text_accum.append(token_texts[idx])
text_accum.append(decoded_token_texts[idx])
token_text_accum.append(decoded_token_texts[idx])
if text_accum:
joined = "".join(text_accum)
if joined:
results.append(TextChunk(joined))
results.append(
TextChunk(joined, tuple(token_text_accum), len(token_text_accum))
)
if deferred_trailing_count:
results.insert(0, TextChunk("", token_count=deferred_trailing_count))
if effective_text:
results = self._recover_holdback_text(effective_text, results)
@@ -124,9 +172,15 @@ class TokenIDScanner:
# transition before the preceding text has arrived. The
# deferred terminals will be resolved against the actual
# delta_text in a subsequent scan() or flushed by finish().
prefix_token_count = 0
for r in results:
if isinstance(r, TextChunk):
prefix_token_count += r.token_count
if isinstance(r, PreLexedTerminal):
self._deferred_terminals.append(r)
self._deferred_prefix_token_counts.append(prefix_token_count)
prefix_token_count = 0
self._deferred_trailing_token_count += prefix_token_count
results = []
return prefix_items + results
@@ -136,16 +190,30 @@ class TokenIDScanner:
return []
results: list[LexerInput] = []
if self._deferred_post_text:
results.append(TextChunk(self._deferred_post_text))
prefix_count = (
self._deferred_prefix_token_counts[0]
if self._deferred_prefix_token_counts
else 0
)
results.append(
TextChunk(self._deferred_post_text, token_count=prefix_count)
)
self._deferred_post_text = ""
results.extend(self._deferred_terminals)
if self._deferred_trailing_token_count:
results.append(
TextChunk("", token_count=self._deferred_trailing_token_count)
)
self._deferred_terminals.clear()
self._deferred_prefix_token_counts.clear()
self._deferred_trailing_token_count = 0
return results
def _resolve_deferred(
self,
delta_text: str,
) -> tuple[list[LexerInput], str]:
current_token_count: int = 0,
) -> tuple[list[LexerInput], str, int]:
"""Resolve deferred terminals against new delta_text.
When a previous ``scan()`` deferred a terminal (its text hadn't
@@ -160,7 +228,11 @@ class TokenIDScanner:
should be scanned with the current delta's token IDs.
"""
deferred = self._deferred_terminals
prefix_token_counts = self._deferred_prefix_token_counts
trailing_token_count = self._deferred_trailing_token_count
self._deferred_terminals = []
self._deferred_prefix_token_counts = []
self._deferred_trailing_token_count = 0
results: list[LexerInput] = []
remaining = delta_text
@@ -171,10 +243,13 @@ class TokenIDScanner:
# Duplicate-text deferred terminals resolve left-to-right via
# find(); correct when each terminal text appears once in sequence.
for terminal in deferred:
for idx, terminal in enumerate(deferred):
prefix_token_count = prefix_token_counts[idx]
pos = remaining.find(terminal.text)
if pos > 0:
results.append(TextChunk(remaining[:pos]))
results.append(
TextChunk(remaining[:pos], token_count=prefix_token_count)
)
results.append(terminal)
remaining = remaining[pos + len(terminal.text) :]
elif pos == 0:
@@ -185,10 +260,16 @@ class TokenIDScanner:
# only the terminal provides a reliable split point.
if remaining:
self._deferred_post_text += remaining
prefix_token_count += current_token_count
remaining = ""
self._deferred_terminals.append(terminal)
self._deferred_prefix_token_counts.append(prefix_token_count)
return results, remaining
if self._deferred_terminals:
self._deferred_trailing_token_count = trailing_token_count
trailing_token_count = 0
return results, remaining, trailing_token_count
def _recover_holdback_text(
self,
@@ -257,6 +338,16 @@ class TokenIDScanner:
if not anchors:
return [TextChunk(delta_text)]
token_groups: list[list[str]] = [[] for _ in range(len(anchors) + 1)]
count_groups = [0] * (len(anchors) + 1)
group_idx = 0
for item in results:
if isinstance(item, PreLexedTerminal):
group_idx += 1
else:
token_groups[group_idx].extend(item.token_texts)
count_groups[group_idx] += item.token_count
# Resolve positions right-to-left: each anchor gets the
# rightmost occurrence that is still before the next anchor.
positions: list[int] = [-1] * len(anchors)
@@ -274,7 +365,13 @@ class TokenIDScanner:
pos = positions[i]
if pos >= consumed:
if pos > consumed:
new_results.append(TextChunk(delta_text[consumed:pos]))
new_results.append(
TextChunk(
delta_text[consumed:pos],
tuple(token_groups[i]),
count_groups[i],
)
)
new_results.append(anchor)
consumed = pos + len(anchor.text)
else:
@@ -291,6 +388,13 @@ class TokenIDScanner:
self._deferred_post_text += delta_text[consumed:]
consumed = len(delta_text)
self._deferred_terminals.append(anchor)
self._deferred_prefix_token_counts.append(count_groups[i])
if consumed < len(delta_text):
new_results.append(TextChunk(delta_text[consumed:]))
new_results.append(
TextChunk(
delta_text[consumed:],
tuple(token_groups[-1]),
count_groups[-1],
)
)
return new_results
+30 -4
View File
@@ -9,6 +9,7 @@ from vllm.logger import init_logger
if TYPE_CHECKING:
from vllm.parser.abstract_parser import Parser
from vllm.parser.engine.parser_engine import ParserEngine
from vllm.reasoning import ReasoningParser
from vllm.tool_parsers import ToolParser
@@ -17,10 +18,30 @@ logger = init_logger(__name__)
class ParserManager:
"""
Provides a unified Parser by composing individual reasoning and tool
parsers from their respective registries.
Provides a unified Parser from the reasoning and tool parser registries.
Parser engine adapters backed by the same engine are collapsed back into
that engine. Other parser pairs are composed through ``DelegatingParser``.
"""
@staticmethod
def _get_parser_engine_cls(
parser_cls: type[object] | None,
) -> type[ParserEngine] | None:
if parser_cls is None:
return None
parser_engine_cls = getattr(parser_cls, "_parser_engine_cls", None)
if parser_engine_cls is None:
return None
from vllm.parser.engine.parser_engine import ParserEngine
if not isinstance(parser_engine_cls, type) or not issubclass(
parser_engine_cls, ParserEngine
):
return None
return parser_engine_cls
@classmethod
def get_tool_parser(
cls,
@@ -84,8 +105,8 @@ class ParserManager:
"""
Get a Parser that handles both reasoning and tool parsing.
Composes individual reasoning and tool parsers into a single
DelegatingParser subclass.
Reuses a shared parser engine when possible, otherwise composes the
individual parsers into a ``DelegatingParser`` subclass.
Args:
tool_parser_name: The name of the tool parser.
@@ -116,6 +137,11 @@ class ParserManager:
HarmonyParser.tool_parser_cls = tool_parser_cls
return HarmonyParser
reasoning_engine_cls = cls._get_parser_engine_cls(reasoning_parser_cls)
tool_engine_cls = cls._get_parser_engine_cls(tool_parser_cls)
if reasoning_engine_cls is not None and reasoning_engine_cls is tool_engine_cls:
return reasoning_engine_cls
if reasoning_parser_name == "kimi_k3" or tool_parser_name == "kimi_k3":
from vllm.parser.kimi_k3 import KimiK3Parser