[Frontend] Replace legacy Gemma4 parsers with engine-based implementation (#45588)

Signed-off-by: Ben Browning <bbrownin@redhat.com>
Co-authored-by: Flora Feng <4florafeng@gmail.com>
This commit is contained in:
Ben Browning
2026-06-15 21:34:07 +00:00
committed by GitHub
co-authored by Flora Feng
parent 25ee659db0
commit 76a373eff4
20 changed files with 2809 additions and 1333 deletions
+9 -4
View File
@@ -33,6 +33,7 @@ class Sample:
expected_tool_calls: list[dict] | None
tools: list[dict] | None = None
chat_template_kwargs: dict | None = None
prompt_token_ids: list[int] | None = None
@dataclass
@@ -120,6 +121,7 @@ def replay_streaming(
holdback_chars: int = 0,
finished_on_last: bool = False,
tools: list[dict] | None = None,
prompt_token_ids: list[int] | None = None,
) -> list[DeltaMessage | None]:
"""Feed tokens through ``parser.parse_delta()`` at a given chunk size.
@@ -146,6 +148,7 @@ def replay_streaming(
all_texts = [text for _, text in tokens]
request = _test_request(tools=tools)
first_prompt_ids = prompt_token_ids if prompt_token_ids is not None else []
if holdback_chars <= 0:
chunks = list(range(0, len(tokens), chunk_size))
@@ -159,7 +162,7 @@ def replay_streaming(
delta_text,
batch_ids,
request,
prompt_token_ids=[] if start == 0 else None,
prompt_token_ids=first_prompt_ids if start == 0 else None,
finished=finished_on_last and is_last,
)
results.append(result)
@@ -192,7 +195,7 @@ def replay_streaming(
delta_text,
batch_ids,
request,
prompt_token_ids=[] if is_first else None,
prompt_token_ids=first_prompt_ids if is_first else None,
finished=finished_on_last and is_last_chunk,
)
results.append(result)
@@ -205,7 +208,7 @@ def replay_streaming(
delta_text,
batch_ids,
request,
prompt_token_ids=[] if is_first else None,
prompt_token_ids=first_prompt_ids if is_first else None,
finished=finished_on_last,
)
results.append(result)
@@ -218,6 +221,7 @@ def replay_with_text_holdback(
tokens: list[tuple[int, str]],
text_delay: int = 1,
tools: list[dict] | None = None,
prompt_token_ids: list[int] | None = None,
) -> list[DeltaMessage | None]:
"""Replay token-by-token with text arriving *text_delay* steps late.
@@ -235,6 +239,7 @@ def replay_with_text_holdback(
"""
results: list[DeltaMessage | None] = []
request = _test_request(tools=tools)
first_prompt_ids = prompt_token_ids if prompt_token_ids is not None else []
n = len(tokens)
held_texts: list[str] = []
@@ -256,7 +261,7 @@ def replay_with_text_holdback(
delta_text,
[token_id],
request,
prompt_token_ids=[] if i == 0 else None,
prompt_token_ids=first_prompt_ids if i == 0 else None,
finished=is_last,
)
results.append(result)
+19 -10
View File
@@ -29,8 +29,9 @@ from vllm.parser.parser_manager import ParserManager
_TOOLS_VALIDATOR = TypeAdapter(list[ChatCompletionToolsParam])
_PAIRINGS: dict[str, tuple[str, str]] = {
"engine": ("qwen3_coder", "qwen3"),
_PAIRINGS: dict[str, tuple[str, str, str]] = {
"engine": ("qwen3_coder", "qwen3", "qwen3"),
"gemma4_engine": ("gemma4", "gemma4", "gemma4"),
}
CHUNK_SIZES = [1, 2, 3, 5, 11, 23, None]
@@ -38,7 +39,7 @@ CHUNK_SIZES = [1, 2, 3, 5, 11, 23, None]
@lru_cache
def _get_delegating_parser_cls(pairings: str) -> type[Parser]:
tool_name, reasoning_name = _PAIRINGS[pairings]
tool_name, reasoning_name, _ = _PAIRINGS[pairings]
parser_cls = ParserManager.get_parser(
tool_parser_name=tool_name,
reasoning_parser_name=reasoning_name,
@@ -48,16 +49,23 @@ def _get_delegating_parser_cls(pairings: str) -> type[Parser]:
return parser_cls
_all_samples = build_samples("qwen3")
def _pairing_samples() -> list[tuple[str, object]]:
items: list[tuple[str, object]] = []
for pairing_name, (_, _, model) in _PAIRINGS.items():
for sample in build_samples(model):
items.append((pairing_name, sample))
return items
_all_pairing_samples = _pairing_samples()
@pytest.mark.parametrize(
"pairings",
list(_PAIRINGS),
ids=lambda p: f"mode={p}",
)
@pytest.mark.parametrize("chunk_size", CHUNK_SIZES, ids=lambda c: f"chunk={c}")
@pytest.mark.parametrize("sample", _all_samples, ids=lambda s: s.id)
@pytest.mark.parametrize(
"pairings,sample",
_all_pairing_samples,
ids=lambda v: v.id if hasattr(v, "id") else v,
)
def test_delegating_replay(sample, chunk_size, pairings):
parser_cls = _get_delegating_parser_cls(pairings=pairings)
@@ -77,6 +85,7 @@ def test_delegating_replay(sample, chunk_size, pairings):
chunk_size=chunk_size,
finished_on_last=True,
tools=sample.tools,
prompt_token_ids=sample.prompt_token_ids,
)
output = collect_output(deltas)
assert_parse_output(output, sample)
File diff suppressed because it is too large Load Diff
+97
View File
@@ -631,6 +631,103 @@ class TestFixArgTypes:
original = '{"name": "Alice"}'
assert engine._fix_arg_types(original, "f") == original
@pytest.mark.parametrize(
"properties, input_json, expected_substr",
[
({"count": {"type": "integer"}}, '{"count": "42"}', '"count": 42'),
({"score": {"type": "number"}}, '{"score": "3.14"}', '"score": 3.14'),
({"flag": {"type": "boolean"}}, '{"flag": "true"}', '"flag": true'),
({"flag": {"type": "boolean"}}, '{"flag": "false"}', '"flag": false'),
({"val": {"type": "null"}}, '{"val": "null"}', '"val": null'),
({"val": {"type": ["string", "null"]}}, '{"val": "null"}', '"val": null'),
({"score": {"type": "number"}}, '{"score": "108."}', '"score": 108'),
],
ids=[
"string_to_int",
"string_to_float",
"string_to_bool_true",
"string_to_bool_false",
"string_to_null",
"string_to_null_union",
"trailing_dot_float",
],
)
def test_string_coerced_to_schema_type(
self,
properties,
input_json,
expected_substr,
):
tool = _make_tool("f", properties)
engine = _make_engine(tools=[tool])
result = engine._fix_arg_types(input_json, "f")
assert expected_substr in result
def test_mixed_types_coerced(self):
tool = _make_tool(
"f",
{
"count": {"type": "integer"},
"active": {"type": "boolean"},
"score": {"type": "number"},
},
)
engine = _make_engine(tools=[tool])
result = engine._fix_arg_types(
'{"count": "42", "active": "true", "score": "3.14"}', "f"
)
parsed = json.loads(result)
assert parsed["count"] == 42
assert parsed["active"] is True
assert parsed["score"] == 3.14
def test_nested_object_coercion(self):
tool = _make_tool(
"f",
{
"inner": {
"type": "object",
"properties": {
"count": {"type": "integer"},
},
},
},
)
engine = _make_engine(tools=[tool])
result = engine._fix_arg_types('{"inner": {"count": "42"}}', "f")
parsed = json.loads(result)
assert parsed["inner"]["count"] == 42
def test_array_item_coercion(self):
tool = _make_tool(
"f",
{
"nums": {
"type": "array",
"items": {"type": "integer"},
},
},
)
engine = _make_engine(tools=[tool])
result = engine._fix_arg_types('{"nums": ["42", "5"]}', "f")
parsed = json.loads(result)
assert parsed["nums"] == [42, 5]
def test_array_mixed_item_types(self):
tool = _make_tool(
"f",
{
"vals": {
"type": "array",
"items": {"type": "number"},
},
},
)
engine = _make_engine(tools=[tool])
result = engine._fix_arg_types('{"vals": ["42", "3.14"]}', "f")
parsed = json.loads(result)
assert parsed["vals"] == [42, 3.14]
# ── TestBuildExtractedResult ─────────────────────────────────────────
+85 -1
View File
@@ -17,19 +17,25 @@ from tests.parser.engine.replay_harness import (
collect_output,
make_mock_tokenizer,
replay_streaming,
replay_with_text_holdback,
)
from tests.parser.engine.trace_builder import build_samples
from vllm.parser.abstract_parser import Parser
from vllm.parser.engine.registered_adapters import (
Gemma4Parser,
Qwen3Parser,
)
_ENGINE_PARSERS: dict[str, type[Parser]] = {
"qwen3_engine": Qwen3Parser,
"gemma4_engine": Gemma4Parser,
}
_gemma4_samples = build_samples("gemma4")
_qwen3_samples = build_samples("qwen3")
_GEMMA4_TERMINALS = ["<|channel>", "<channel|>", "<|tool_call>", "<tool_call|>"]
_QWEN3_TERMINALS = [
"<think>",
"</think>",
@@ -56,6 +62,7 @@ class TestQwen3ReplayWithHoldback:
sample.tokens,
chunk_size=chunk_size,
holdback_chars=holdback,
prompt_token_ids=sample.prompt_token_ids,
)
output = collect_output(deltas)
@@ -67,10 +74,85 @@ class TestQwen3ReplayWithHoldback:
)
@pytest.mark.parametrize("holdback", HOLDBACK_CONFIGS, ids=lambda h: f"holdback{h}")
@pytest.mark.parametrize("chunk_size", [3, 5, 10], ids=lambda c: f"chunk{c}")
@pytest.mark.parametrize("sample", _gemma4_samples, ids=lambda s: s.id)
class TestGemma4ReplayWithHoldback:
"""Replay with simulated detokenizer holdback."""
def test_replay(self, sample, chunk_size, holdback):
tokenizer = make_mock_tokenizer(sample)
parser = Gemma4Parser(tokenizer, sample.tools)
deltas = replay_streaming(
parser,
sample.tokens,
chunk_size=chunk_size,
holdback_chars=holdback,
prompt_token_ids=sample.prompt_token_ids,
)
output = collect_output(deltas)
assert_parse_output(output, sample)
assert_no_terminal_leakage(
output,
_GEMMA4_TERMINALS,
context=f"chunk_size={chunk_size}, holdback={holdback}",
)
TEXT_HOLDBACK_DELAYS = [1, 2, 3]
@pytest.mark.parametrize("delay", TEXT_HOLDBACK_DELAYS, ids=lambda d: f"delay{d}")
@pytest.mark.parametrize("sample", _gemma4_samples, ids=lambda s: s.id)
class TestGemma4TextHoldback:
"""Replay with production-like text/token-ID misalignment.
In production the detokenizer sends token IDs immediately but holds
back text by N tokens. This exercises the TokenIDScanner deferred
terminal path that aligned-holdback tests do not cover.
"""
def test_replay(self, sample, delay):
tokenizer = make_mock_tokenizer(sample)
parser = Gemma4Parser(tokenizer, sample.tools)
deltas = replay_with_text_holdback(
parser,
sample.tokens,
text_delay=delay,
prompt_token_ids=sample.prompt_token_ids,
)
output = collect_output(deltas)
assert_parse_output(output, sample)
assert_no_terminal_leakage(
output,
_GEMMA4_TERMINALS,
context=f"text_delay={delay}",
)
class TestParserEngineAdjustRequest:
"""Verify ParserEngine and its adapters set skip_special_tokens=False."""
def test_adjust_request_disables_skip_special_tokens(self):
sample = _gemma4_samples[0]
tokenizer = make_mock_tokenizer(sample)
parser = Gemma4Parser(tokenizer, sample.tools)
request = _test_request()
assert request.skip_special_tokens is True
adjusted = parser.adjust_request(request)
assert adjusted.skip_special_tokens is False
_TOOL_CALL_SAMPLES = [
(Qwen3Parser, s)
for s in _qwen3_samples
if s.expected_tool_calls and s.expected_reasoning
] + [
(Gemma4Parser, s)
for s in _gemma4_samples
if s.expected_tool_calls and s.expected_reasoning
]
@@ -147,7 +229,9 @@ class TestSkipToolParsingReplay:
"".join(all_texts[start:end]),
all_ids[start:end],
request,
prompt_token_ids=[] if start == 0 else None,
prompt_token_ids=(sample.prompt_token_ids or [])
if start == 0
else None,
finished=is_last,
)
results.append(result)
+528 -124
View File
@@ -1,20 +1,19 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Tests for TokenIDScanner, focusing on hold-back text recovery.
Uses gemma4_config for all end-to-end engine tests, covering
reasoning channels, tool calls, and combined flows."""
"""Tests for TokenIDScanner."""
from unittest.mock import MagicMock
import pytest
from vllm.parser.engine.events import EventType
from vllm.parser.engine.streaming_parser_engine import StreamingParserEngine
from vllm.parser.engine.token_id_scanner import (
PreLexedTerminal,
TextChunk,
TokenIDScanner,
)
from vllm.parser.gemma4 import gemma4_config
CHANNEL_START = "<|channel>"
CHANNEL_END = "<channel|>"
@@ -54,8 +53,7 @@ def scanner(tokenizer):
class TestJoinDecodedTextReturnsStr:
"""_join_decoded_text now returns str unconditionally (was
str | None when an isinstance guard made a branch unreachable)."""
"""_join_decoded_text always returns str."""
@pytest.fixture
def bare_scanner(self):
@@ -83,9 +81,7 @@ class TestJoinDecodedTextReturnsStr:
class TestHoldbackTextRecovery:
def test_holdback_text_with_special_token_text_absent(self, scanner):
"""delta_text has hold-back text but the special token's text is
NOT in delta_text (held back by the detokenizer). Terminal is
deferred until the text arrives in a subsequent delta."""
"""Terminal deferred when its text is absent from delta_text."""
result = scanner.scan(
delta_text="processed is appropriate.",
delta_token_ids=[CHANNEL_END_ID],
@@ -93,8 +89,6 @@ class TestHoldbackTextRecovery:
assert len(result) == 0
# Second scan: terminal text arrives (detokenizer flushes).
# Deferred terminal resolves with holdback text before it.
result2 = scanner.scan(
delta_text="<channel|>Understood.",
delta_token_ids=[20, 21],
@@ -108,7 +102,7 @@ class TestHoldbackTextRecovery:
assert "Understood." in combined
def test_holdback_text_with_special_token_text_present(self, scanner):
"""delta_text includes hold-back text AND the special token text."""
"""Hold-back text + special token text both in delta_text."""
result = scanner.scan(
delta_text="holdback text<channel|>",
delta_token_ids=[CHANNEL_END_ID],
@@ -121,7 +115,7 @@ class TestHoldbackTextRecovery:
assert result[1].terminal == "THINK_END"
def test_no_holdback_text(self, scanner):
"""delta_text is exactly the special token text — no hold-back."""
"""delta_text is exactly the special token text."""
result = scanner.scan(
delta_text="<channel|>",
delta_token_ids=[CHANNEL_END_ID],
@@ -132,7 +126,7 @@ class TestHoldbackTextRecovery:
assert result[0].terminal == "THINK_END"
def test_empty_delta_text(self, scanner):
"""delta_text is empty — terminal deferred until text arrives."""
"""Empty delta_text defers the terminal until text arrives."""
result = scanner.scan(
delta_text="",
delta_token_ids=[CHANNEL_END_ID],
@@ -146,9 +140,7 @@ class TestHoldbackTextRecovery:
assert flushed[0].terminal == "THINK_END"
def test_empty_delta_text_drops_individual_decode_text(self, tokenizer):
"""delta_text="" with multiple tokens including special: all
results deferred — individually-decoded TextChunks are unreliable
and PreLexedTerminals wait for text confirmation."""
"""Empty delta_text with multiple tokens: all results deferred."""
tool_start_id = 400
tok_a = 201
tok_b = 202
@@ -176,7 +168,6 @@ class TestHoldbackTextRecovery:
assert flushed[0].terminal == "TOOL_START"
def test_holdback_before_start_tag(self, scanner):
"""Hold-back text before a reasoning start tag."""
result = scanner.scan(
delta_text="prefix text<|channel>",
delta_token_ids=[CHANNEL_START_ID],
@@ -189,8 +180,7 @@ class TestHoldbackTextRecovery:
assert result[1].terminal == "THINK_START"
def test_multi_token_batch_special_in_middle(self, scanner, tokenizer):
"""Stream-interval > 1: batch has regular tokens + special token.
delta_text differs from individual decodes (context-dependent)."""
"""Multi-token batch with special token in the middle."""
tok_a = 201
tok_b = 202
tokenizer.decode.side_effect = lambda ids: {
@@ -215,10 +205,7 @@ class TestHoldbackTextRecovery:
assert "holdback wordA" in "".join(texts)
def test_multi_token_batch_special_token_text_absent(self, scanner, tokenizer):
"""Stream-interval > 1: batch has regular + special token, but
delta_text doesn't contain the special token text at all
(held back by detokenizer along with trailing regular tokens).
Terminal is deferred until text arrives."""
"""Multi-token batch where special token text is absent."""
tok_a = 201
tok_b = 202
tokenizer.decode.side_effect = lambda ids: {
@@ -239,8 +226,6 @@ class TestHoldbackTextRecovery:
assert len(result) == 0
# Next delta: terminal text arrives (detokenizer flushes).
# Deferred terminal resolves with holdback text before it.
result2 = scanner_multi.scan(
delta_text="<channel|> more text",
delta_token_ids=[300],
@@ -254,8 +239,7 @@ class TestHoldbackTextRecovery:
assert "more text" in combined
def test_holdback_with_content_after_special_token(self, tokenizer):
"""delta_text has hold-back + special token + content after,
with corresponding token IDs for all parts."""
"""Hold-back + special token + content after in one delta."""
tok_content = 210
tokenizer.decode.side_effect = lambda ids: {
CHANNEL_END_ID: CHANNEL_END,
@@ -283,8 +267,7 @@ class TestHoldbackTextRecovery:
class TestDropTokens:
def test_drop_token_with_holdback(self, tokenizer):
"""Drop tokens stripped from delta_text, hold-back text preserved.
Terminal is deferred when its text is absent from delta_text."""
"""Drop tokens stripped; hold-back text preserved."""
drop_id = 300
tokenizer.decode.side_effect = lambda ids: {
CHANNEL_END_ID: CHANNEL_END,
@@ -304,7 +287,6 @@ class TestDropTokens:
assert len(result) == 0
# Terminal text arrives in next delta; deferred terminal resolves.
result2 = scanner.scan(
delta_text="<channel|>content",
delta_token_ids=[20],
@@ -321,40 +303,10 @@ class TestDropTokens:
class TestEndToEndReasoningHoldback:
"""End-to-end tests through the full parser engine simulating
stream-interval > 1 and detokenizer hold-back, using
gemma4_config."""
"""End-to-end engine tests with detokenizer hold-back."""
def test_reasoning_content_not_truncated(self):
from vllm.parser.engine.parser_engine_config import (
ParserEngineConfig,
ParserState,
Transition,
)
from vllm.parser.engine.streaming_parser_engine import StreamingParserEngine
config = ParserEngineConfig(
name="test_channel",
initial_state=ParserState.CONTENT,
terminals={
"THINK_START": CHANNEL_START,
"THINK_END": CHANNEL_END,
},
token_id_terminals={
"THINK_START": CHANNEL_START,
"THINK_END": CHANNEL_END,
},
transitions={
(ParserState.CONTENT, "THINK_START"): Transition(
ParserState.REASONING,
(EventType.REASONING_START,),
),
(ParserState.REASONING, "THINK_END"): Transition(
ParserState.CONTENT,
(EventType.REASONING_END,),
),
},
)
config = gemma4_config()
tok = MagicMock()
vocab = {
CHANNEL_START: CHANNEL_START_ID,
@@ -369,30 +321,21 @@ class TestEndToEndReasoningHoldback:
engine = StreamingParserEngine(config, tok)
all_events = []
# Delta 1: channel start token (text includes start tag)
all_events.extend(engine.feed(CHANNEL_START, [CHANNEL_START_ID]))
# Delta 2: reasoning text (normal content, no special tokens)
all_events.extend(
engine.feed(
"thought\nThe request was received and ",
[10, 11, 12, 13, 14],
)
)
# Delta 3: MORE reasoning text, the detokenizer held some back.
# Then channel end token arrives in token_ids, but its text
# is NOT in delta_text (held back by detokenizer).
# delta_text = previously held-back reasoning text only.
# CHANNEL_END token arrives but its text is held back.
all_events.extend(
engine.feed(
"processed is appropriate.",
[CHANNEL_END_ID],
)
)
# Delta 4: detokenizer flushes held-back channel end text
# plus new content tokens.
# Detokenizer flushes the held-back text.
all_events.extend(
engine.feed(
"<channel|>Understood.",
@@ -413,36 +356,7 @@ class TestEndToEndReasoningHoldback:
assert "Understood." in content_text
def test_backtick_content_not_truncated(self):
"""Reproduces the hostname backtick truncation case."""
from vllm.parser.engine.parser_engine_config import (
ParserEngineConfig,
ParserState,
Transition,
)
from vllm.parser.engine.streaming_parser_engine import StreamingParserEngine
config = ParserEngineConfig(
name="test_channel",
initial_state=ParserState.CONTENT,
terminals={
"THINK_START": CHANNEL_START,
"THINK_END": CHANNEL_END,
},
token_id_terminals={
"THINK_START": CHANNEL_START,
"THINK_END": CHANNEL_END,
},
transitions={
(ParserState.CONTENT, "THINK_START"): Transition(
ParserState.REASONING,
(EventType.REASONING_START,),
),
(ParserState.REASONING, "THINK_END"): Transition(
ParserState.CONTENT,
(EventType.REASONING_END,),
),
},
)
config = gemma4_config()
tok = MagicMock()
vocab = {
CHANNEL_START: CHANNEL_START_ID,
@@ -464,17 +378,12 @@ class TestEndToEndReasoningHoldback:
[10, 11, 12, 13],
)
)
# Hold-back text includes backtick content; channel end text
# absent from delta_text.
all_events.extend(
engine.feed(
"`hostname`.\n",
[CHANNEL_END_ID],
)
)
# Next delta flushes channel end + tool call start
all_events.extend(
engine.feed(
"<channel|>tool output",
@@ -491,10 +400,516 @@ class TestEndToEndReasoningHoldback:
assert "`hostname`." in reasoning_text
_CHANNEL_START_TAG = "<|channel>"
_CHANNEL_END_TAG = "<channel|>"
_TOOL_START_TAG = "<|tool_call>"
_TOOL_END_TAG = "<tool_call|>"
_QUOTE_TAG = '<|"|>'
_CHANNEL_START_TID = 100
_CHANNEL_END_TID = 101
_TOOL_START_TID = 102
_TOOL_END_TID = 103
_QUOTE_TID = 104
_TOK = list(range(200, 215))
def _gemma4_vocab() -> dict[str, int]:
return {
_CHANNEL_START_TAG: _CHANNEL_START_TID,
_CHANNEL_END_TAG: _CHANNEL_END_TID,
_TOOL_START_TAG: _TOOL_START_TID,
_TOOL_END_TAG: _TOOL_END_TID,
_QUOTE_TAG: _QUOTE_TID,
}
def _make_gemma4_tokenizer(
extra_decode: dict[int, str] | None = None,
) -> MagicMock:
special = {
_CHANNEL_START_TID: _CHANNEL_START_TAG,
_CHANNEL_END_TID: _CHANNEL_END_TAG,
_TOOL_START_TID: _TOOL_START_TAG,
_TOOL_END_TID: _TOOL_END_TAG,
_QUOTE_TID: _QUOTE_TAG,
}
decode_map = {**special, **(extra_decode or {})}
tok = MagicMock()
tok.get_vocab.return_value = _gemma4_vocab()
tok.decode.side_effect = lambda ids: decode_map.get(ids[0], f"tok{ids[0]}")
return tok
def _collect_events(engine, deltas):
from vllm.parser.engine.events import SemanticEvent
all_events: list[SemanticEvent] = []
for delta_text, delta_token_ids in deltas:
all_events.extend(engine.feed(delta_text, delta_token_ids))
all_events.extend(engine.finish())
return all_events
def _reasoning_text(events) -> str:
return "".join(e.value for e in events if e.type == EventType.REASONING_CHUNK)
def _content_text(events) -> str:
return "".join(e.value for e in events if e.type == EventType.TEXT_CHUNK)
def _arg_text(events) -> str:
return "".join(e.value for e in events if e.type == EventType.ARG_VALUE_CHUNK)
def _has_event(events, event_type) -> bool:
return any(e.type == event_type for e in events)
class TestMultiTokenBoundaryPreservation:
"""No text lost at state boundaries with multi-token deltas."""
def test_empty_delta_text_at_channel_end_unified(self):
"""Empty delta_text when CHANNEL_END arrives; text comes later."""
tok = _make_gemma4_tokenizer()
engine = StreamingParserEngine(gemma4_config(), tok)
events = _collect_events(
engine,
[
("", [_CHANNEL_START_TID]),
("<|channel>thought\nSome reasoning.", [_TOK[0], _TOK[1]]),
("", [_CHANNEL_END_TID]),
("<channel|>Final answer.", [_TOK[2], _TOK[3]]),
],
)
reasoning = _reasoning_text(events)
content = _content_text(events)
assert "Some reasoning." in reasoning
assert "Final answer." in content
assert _has_event(events, EventType.REASONING_START)
assert _has_event(events, EventType.REASONING_END)
def test_deferred_channel_end_flushed_at_finish_unified(self):
"""Deferred CHANNEL_END flushed at end-of-stream."""
tok = _make_gemma4_tokenizer()
engine = StreamingParserEngine(gemma4_config(), tok)
events = _collect_events(
engine,
[
(_CHANNEL_START_TAG, [_CHANNEL_START_TID]),
("thought\nReasoning text.", [_TOK[0]]),
(" Final thought.", [_CHANNEL_END_TID]),
],
)
reasoning = _reasoning_text(events)
assert "Reasoning text. Final thought." in reasoning
assert _has_event(events, EventType.REASONING_END)
def test_reasoning_to_tool_call_handoff_unified(self):
"""Full reasoning -> content -> tool call flow."""
tok = _make_gemma4_tokenizer()
engine = StreamingParserEngine(gemma4_config(), tok)
events = _collect_events(
engine,
[
(_CHANNEL_START_TAG, [_CHANNEL_START_TID]),
("thought\nI need to check the weather.", [_TOK[0], _TOK[1], _TOK[2]]),
(_CHANNEL_END_TAG, [_CHANNEL_END_TID]),
("Let me call a tool.", [_TOK[3], _TOK[4]]),
(_TOOL_START_TAG, [_TOOL_START_TID]),
("call:get_weather{city:", [_TOK[5], _TOK[6]]),
('<|"|>SF<|"|>}', [_QUOTE_TID, _TOK[7], _QUOTE_TID, _TOK[8]]),
(_TOOL_END_TAG, [_TOOL_END_TID]),
],
)
reasoning = _reasoning_text(events)
content = _content_text(events)
assert "I need to check the weather." in reasoning
assert "Let me call a tool." in content
assert _has_event(events, EventType.REASONING_START)
assert _has_event(events, EventType.REASONING_END)
assert _has_event(events, EventType.TOOL_CALL_START)
assert _has_event(events, EventType.TOOL_CALL_END)
assert "SF" in _arg_text(events)
def test_multiple_tool_calls_rapid_transitions_unified(self):
"""Two back-to-back tool calls with correct tool_index tracking."""
tok = _make_gemma4_tokenizer()
engine = StreamingParserEngine(gemma4_config(), tok)
events = _collect_events(
engine,
[
(_TOOL_START_TAG, [_TOOL_START_TID]),
("call:get_weather{city:", [_TOK[0], _TOK[1]]),
('<|"|>NYC<|"|>}', [_QUOTE_TID, _TOK[2], _QUOTE_TID, _TOK[3]]),
(_TOOL_END_TAG, [_TOOL_END_TID]),
(_TOOL_START_TAG, [_TOOL_START_TID]),
("call:get_time{tz:", [_TOK[4], _TOK[5]]),
('<|"|>EST<|"|>}', [_QUOTE_TID, _TOK[6], _QUOTE_TID, _TOK[7]]),
(_TOOL_END_TAG, [_TOOL_END_TID]),
],
)
starts = [e for e in events if e.type == EventType.TOOL_CALL_START]
ends = [e for e in events if e.type == EventType.TOOL_CALL_END]
assert len(starts) == 2
assert len(ends) == 2
assert starts[0].tool_index == 0
assert starts[1].tool_index == 1
names = "".join(e.value for e in events if e.type == EventType.TOOL_NAME)
assert "get_weather" in names
assert "get_time" in names
def test_deferred_channel_end_before_tool_call_unified(self):
"""Deferred CHANNEL_END followed by a tool call."""
tok = _make_gemma4_tokenizer()
engine = StreamingParserEngine(gemma4_config(), tok)
events = _collect_events(
engine,
[
(_CHANNEL_START_TAG, [_CHANNEL_START_TID]),
("thought\nNeed to call a tool.", [_TOK[0], _TOK[1]]),
(" Let me proceed.", [_CHANNEL_END_TID]),
(_CHANNEL_END_TAG, [_TOK[2]]),
(_TOOL_START_TAG, [_TOOL_START_TID]),
("call:get_weather{city:", [_TOK[3], _TOK[4]]),
('<|"|>Tokyo<|"|>}', [_QUOTE_TID, _TOK[5], _QUOTE_TID, _TOK[6]]),
(_TOOL_END_TAG, [_TOOL_END_TID]),
],
)
reasoning = _reasoning_text(events)
assert "Need to call a tool. Let me proceed." in reasoning
assert _has_event(events, EventType.REASONING_END)
assert _has_event(events, EventType.TOOL_CALL_START)
assert _has_event(events, EventType.TOOL_CALL_END)
assert "Tokyo" in _arg_text(events)
class TestStreamInterval10:
"""Tests with stream_interval=10 (large multi-token batches)."""
def test_channel_end_mid_batch_text_present(self):
"""<channel|> mid-batch with its text present in delta_text."""
tok = _make_gemma4_tokenizer({_TOK[i]: f"word{i} " for i in range(15)})
engine = StreamingParserEngine(gemma4_config(), tok)
events: list = []
events.extend(
engine.feed(
"<|channel>thought\nword0 word1 word2 word3 word4 "
"word5 word6 word7 word8 ",
[
_CHANNEL_START_TID,
_TOK[0],
_TOK[1],
_TOK[2],
_TOK[3],
_TOK[4],
_TOK[5],
_TOK[6],
_TOK[7],
_TOK[8],
],
)
)
events.extend(
engine.feed(
"word9 word10 word11 <channel|>word12 word13 word14 word0 word1 word2 ",
[
_TOK[9],
_TOK[10],
_TOK[11],
_CHANNEL_END_TID,
_TOK[12],
_TOK[13],
_TOK[14],
_TOK[0],
_TOK[1],
_TOK[2],
],
)
)
events.extend(engine.finish())
reasoning = _reasoning_text(events)
content = _content_text(events)
for w in ("word9", "word10", "word11"):
assert w in reasoning, f"{w!r} missing from reasoning"
for w in ("word12", "word13", "word14"):
assert w in content, f"{w!r} missing from content"
assert _has_event(events, EventType.REASONING_END)
def test_channel_end_and_tool_start_same_batch_unified(self):
"""Both <channel|> and <|tool_call> in a single batch."""
tok = _make_gemma4_tokenizer({_TOK[i]: f"w{i} " for i in range(15)})
engine = StreamingParserEngine(gemma4_config(), tok)
events: list = []
events.extend(
engine.feed(
"<|channel>thought\nw0 w1 w2 w3 w4 w5 w6 w7 w8 ",
[
_CHANNEL_START_TID,
_TOK[0],
_TOK[1],
_TOK[2],
_TOK[3],
_TOK[4],
_TOK[5],
_TOK[6],
_TOK[7],
_TOK[8],
],
)
)
events.extend(
engine.feed(
"w9 w10 <channel|>w11 <|tool_call>",
[
_TOK[9],
_TOK[10],
_CHANNEL_END_TID,
_TOK[11],
_TOOL_START_TID,
_TOK[12],
_TOK[13],
_TOK[14],
_TOK[0],
_TOK[1],
],
)
)
events.extend(engine.finish())
reasoning = _reasoning_text(events)
assert "w9" in reasoning
assert "w10" in reasoning
assert _has_event(events, EventType.REASONING_END)
assert _has_event(events, EventType.TOOL_CALL_START)
def test_channel_end_mid_batch_text_absent(self):
"""<channel|> mid-batch with its text absent from delta_text."""
tok = _make_gemma4_tokenizer({_TOK[i]: f"word{i} " for i in range(15)})
engine = StreamingParserEngine(gemma4_config(), tok)
events: list = []
events.extend(
engine.feed(
"<|channel>thought\nword0 word1 word2 word3 word4 "
"word5 word6 word7 word8 ",
[
_CHANNEL_START_TID,
_TOK[0],
_TOK[1],
_TOK[2],
_TOK[3],
_TOK[4],
_TOK[5],
_TOK[6],
_TOK[7],
_TOK[8],
],
)
)
events.extend(
engine.feed(
"word9 word10 word11 ",
[
_TOK[9],
_TOK[10],
_TOK[11],
_CHANNEL_END_TID,
_TOK[12],
_TOK[13],
_TOK[14],
_TOK[0],
_TOK[1],
_TOK[2],
],
)
)
events.extend(
engine.feed(
"<channel|>word12 word13 word14 word0 word1 word2 ",
[_TOK[3], _TOK[4], _TOK[5]],
)
)
events.extend(engine.finish())
reasoning = _reasoning_text(events)
content = _content_text(events)
for w in ("word9", "word10", "word11"):
assert w in reasoning, f"{w!r} missing from reasoning"
for w in ("word12", "word13", "word14"):
assert w in content, f"{w!r} missing from content"
assert _has_event(events, EventType.REASONING_END)
def test_tool_end_mid_batch_text_absent_unified(self):
"""<tool_call|> mid-batch with text absent."""
tok = _make_gemma4_tokenizer({_TOK[i]: f"w{i}" for i in range(15)})
engine = StreamingParserEngine(gemma4_config(), tok)
events: list = []
events.extend(
engine.feed(
_CHANNEL_START_TAG,
[_CHANNEL_START_TID],
)
)
events.extend(
engine.feed(
"thought\nNeed a tool.",
[_TOK[0], _TOK[1]],
)
)
events.extend(
engine.feed(
_TOOL_START_TAG,
[_TOOL_START_TID],
)
)
events.extend(
engine.feed(
"call:get_weather{city:",
[_TOK[2], _TOK[3], _TOK[4]],
)
)
events.extend(
engine.feed(
'<|"|>San Francisco<|"|>}',
[
_QUOTE_TID,
_TOK[5],
_TOK[6],
_QUOTE_TID,
_TOK[7],
_TOOL_END_TID,
_TOK[8],
_TOK[9],
_TOK[10],
_TOK[11],
],
)
)
events.extend(
engine.feed(
"<tool_call|>w8w9w10w11w12",
[_TOK[12], _TOK[13]],
)
)
events.extend(engine.finish())
assert _has_event(events, EventType.TOOL_CALL_END)
assert "San Francisco" in _arg_text(events)
def test_large_batch_holdback_spans_two_batches(self):
"""Holdback text spanning two batches with <channel|> in the second."""
tok = _make_gemma4_tokenizer({_TOK[i]: f"w{i} " for i in range(15)})
engine = StreamingParserEngine(gemma4_config(), tok)
events: list = []
events.extend(
engine.feed(
"<|channel>thought\nThe user asked about machine learning "
"and I need to think about the best approach to",
[
_CHANNEL_START_TID,
_TOK[0],
_TOK[1],
_TOK[2],
_TOK[3],
_TOK[4],
_TOK[5],
_TOK[6],
_TOK[7],
_TOK[8],
],
)
)
events.extend(
engine.feed(
" explain this complex topic. Let me organize my thoughts.",
[
_TOK[9],
_TOK[10],
_TOK[11],
_TOK[12],
_TOK[13],
_TOK[14],
_CHANNEL_END_TID,
_TOK[0],
_TOK[1],
_TOK[2],
],
)
)
events.extend(
engine.feed(
"<channel|>w0 w1 w2 Here is what I recommend: start with "
"the fundamentals and build up from there.",
[
_TOK[3],
_TOK[4],
_TOK[5],
_TOK[6],
_TOK[7],
_TOK[8],
_TOK[9],
_TOK[10],
_TOK[11],
_TOK[12],
],
)
)
events.extend(engine.finish())
reasoning = _reasoning_text(events)
content = _content_text(events)
assert "organize my thoughts." in reasoning
assert "explain" in reasoning
assert "recommend" in content
assert _has_event(events, EventType.REASONING_START)
assert _has_event(events, EventType.REASONING_END)
class TestRebuildFromAnchorsLiteralLookalike:
"""When delta_text contains a literal mention of a special token's
text before the real special token, _rebuild_from_anchors must
anchor at the real occurrence, not the literal one."""
"""Literal token text in prose must not be consumed as an anchor."""
@pytest.fixture
def tool_scanner(self):
@@ -513,8 +928,6 @@ class TestRebuildFromAnchorsLiteralLookalike:
)
def test_literal_before_real_anchor(self, tool_scanner):
"""Literal <tool_call> in prose followed by a real <tool_call>
special token — the scanner must split at the real one."""
delta_text = 'Use <tool_call> like this: <tool_call>{"name":"f"}</tool_call>'
delta_token_ids = [1, 2, 3, 4, 5, TOOL_START_ID, 6, 7, TOOL_END_ID]
items = tool_scanner.scan(delta_text, delta_token_ids)
@@ -526,14 +939,11 @@ class TestRebuildFromAnchorsLiteralLookalike:
assert terminals[0].terminal == "TOOL_START"
assert terminals[1].terminal == "TOOL_END"
# The literal mention must appear in a text chunk, not be
# consumed by the TOOL_START anchor.
joined_text = "".join(text_parts)
assert "<tool_call>" in joined_text
assert '{"name":"f"}' in joined_text
def test_multiple_tool_calls_with_literal_between(self, tool_scanner):
"""Two real tool calls with a literal mention between them."""
delta_text = (
'<tool_call>{"name":"a"}</tool_call>'
" see <tool_call> syntax "
@@ -557,14 +967,11 @@ class TestRebuildFromAnchorsLiteralLookalike:
text_parts = [it.text for it in items if isinstance(it, TextChunk)]
joined_text = "".join(text_parts)
# The literal mention between the two real calls must be in text
assert "<tool_call> syntax" in joined_text
class TestRebuildFromAnchorsCascadingDeferral:
"""When a middle anchor's text is absent from delta_text,
only that anchor should be deferred — not subsequent ones
with valid positions."""
"""Missing middle anchor defers only itself, not subsequent ones."""
@pytest.fixture
def bare_scanner(self):
@@ -623,9 +1030,6 @@ class TestRebuildFromAnchorsCascadingDeferral:
texts = [r for r in rebuilt if isinstance(r, TextChunk)]
joined = "".join(t.text for t in texts)
assert "text" in joined
# "more" is deferred along with the missing terminal —
# it will be resolved in the next scan when the terminal
# text arrives.
assert bare_scanner._deferred_post_text == "more"
assert len(bare_scanner._deferred_terminals) == 1
assert bare_scanner._deferred_terminals[0].terminal == "THINK_END"
+113 -2
View File
@@ -29,6 +29,7 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
)
from vllm.parser.engine.registered_adapters import (
Gemma4Parser,
Qwen3Parser,
)
@@ -48,6 +49,7 @@ class Scenario:
reasoning: str | None = None
content: str | None = None
tool_calls: list[ToolCallSpec] | None = None
after_tool_response: bool = False
# ── Scenarios ────────────────────────────────────────────────────────
@@ -132,6 +134,12 @@ SCENARIOS: list[Scenario] = [
reasoning="",
content="The epoch timestamp is 1779111346.",
),
Scenario(
id="tool-after-tool-response",
description="Tool call immediately after tool response (agentic flow)",
tool_calls=[_READ_TOOL],
after_tool_response=True,
),
]
@@ -250,7 +258,13 @@ def _validate_sample(sample: Sample, parser_cls: type, **kwargs) -> None:
"""Replay sample through the real parser and assert correctness."""
tokenizer = MockTokenizer(vocab=dict(sample.vocab), tokens=sample.tokens)
parser = parser_cls(tokenizer, sample.tools, **kwargs)
deltas = replay_streaming(parser, sample.tokens, chunk_size=1, tools=sample.tools)
deltas = replay_streaming(
parser,
sample.tokens,
chunk_size=1,
tools=sample.tools,
prompt_token_ids=sample.prompt_token_ids,
)
output = collect_output(deltas)
assert_parse_output(output, sample)
@@ -273,6 +287,7 @@ def _make_sample(
expected_tool_calls: list[dict] | None,
tools: list[dict] | None,
chat_template_kwargs: dict | None = None,
prompt_token_ids: list[int] | None = None,
) -> Sample:
tokens = _tokenize(segments, vocab)
return Sample(
@@ -286,10 +301,11 @@ def _make_sample(
expected_tool_calls=expected_tool_calls,
tools=_validate_tools(tools),
chat_template_kwargs=chat_template_kwargs,
prompt_token_ids=prompt_token_ids,
)
# ── Qwen3 / NemotronV3 (XML tool format, starts in REASONING) ───────
# ── Qwen3 (XML tool format, starts in REASONING) ────────────────────
_QWEN3_VOCAB: dict[str, int] = {
"<think>": 50,
@@ -376,10 +392,105 @@ def _build_qwen3(
return sample
# ── Gemma4 (channel reasoning, custom arg format) ────────────────────
_GEMMA4_VOCAB: dict[str, int] = {
"<|channel>": 50,
"<channel|>": 51,
"<|tool_call>": 48,
"<tool_call|>": 49,
'<|"|>': 52,
"<|turn>": 53,
"<|tool_response>": 54,
}
_GEMMA4_THOUGHT_PREFIX = "thought\n"
_GEMMA4_QUOTE = '<|"|>'
def _gemma4_value_segments(value: Any) -> list[tuple[str, bool]]:
"""Render a value in Gemma4 arg format as segments."""
if isinstance(value, str):
return [(_GEMMA4_QUOTE, True), (value, False), (_GEMMA4_QUOTE, True)]
if isinstance(value, bool):
return [("true" if value else "false", False)]
if isinstance(value, (int, float)):
return [(str(value), False)]
if isinstance(value, dict):
segs: list[tuple[str, bool]] = [("{", False)]
for i, (k, v) in enumerate(value.items()):
if i > 0:
segs.append((",", False))
segs.append((f"{k}:", False))
segs.extend(_gemma4_value_segments(v))
segs.append(("}", False))
return segs
if isinstance(value, list):
segs = [("[", False)]
for i, item in enumerate(value):
if i > 0:
segs.append((",", False))
segs.extend(_gemma4_value_segments(item))
segs.append(("]", False))
return segs
return [(json.dumps(value, ensure_ascii=False), False)]
def _gemma4_tool_segments(tc: ToolCallSpec) -> list[tuple[str, bool]]:
segs: list[tuple[str, bool]] = [
("<|tool_call>", True),
(f"call:{tc.name}", False),
("{", False),
]
for i, (key, value) in enumerate(tc.arguments.items()):
if i > 0:
segs.append((",", False))
segs.append((f"{key}:", False))
segs.extend(_gemma4_value_segments(value))
segs.append(("}", False))
segs.append(("<tool_call|>", True))
return segs
def _gemma4_segments(scenario: Scenario) -> list[tuple[str, bool]]:
segs: list[tuple[str, bool]] = []
if scenario.reasoning is not None:
segs.append(("<|channel>", True))
segs.append((_GEMMA4_THOUGHT_PREFIX, False))
segs.append((scenario.reasoning, False))
segs.append(("<channel|>", True))
if scenario.content is not None:
segs.append((scenario.content, False))
if scenario.tool_calls:
for tc in scenario.tool_calls:
segs.extend(_gemma4_tool_segments(tc))
return segs
def _build_gemma4(scenario: Scenario, validate: bool = True) -> Sample:
prompt_token_ids = None
if scenario.after_tool_response:
prompt_token_ids = [_GEMMA4_VOCAB["<|tool_response>"]]
sample = _make_sample(
sample_id=f"gemma4-{scenario.id}",
description=scenario.description,
vocab=_GEMMA4_VOCAB,
segments=_gemma4_segments(scenario),
expected_reasoning=scenario.reasoning,
expected_content=_qwen3_expected_content(scenario),
expected_tool_calls=_expected_tc(scenario),
tools=_expected_tools(scenario),
prompt_token_ids=prompt_token_ids,
)
if validate:
_validate_sample(sample, Gemma4Parser)
return sample
# ── Registry and public API ──────────────────────────────────────────
_BUILDERS: dict[str, Any] = {
"qwen3": _build_qwen3,
"gemma4": _build_gemma4,
}
@@ -83,15 +83,15 @@ CHANNEL_NO_END = {
EMPTY = {
"output": "",
"reasoning": None,
"content": "",
"is_reasoning_end": False,
"content": None,
"is_reasoning_end": True,
}
NEW_LINE_NONSTREAMING = {
"output": (
"Before\n<|channel>This is a reasoning section<channel|>\nThis is the rest"
),
"reasoning": "This is a reasoning section",
"content": "\nThis is the rest",
"content": "Before\n\nThis is the rest",
"is_reasoning_end": True,
}
NEW_LINE_STREAMING = {
@@ -111,7 +111,7 @@ THOUGHT_PREFIX = {
}
THOUGHT_PREFIX_ONLY = {
"output": "<|channel>thought\n<channel|>",
"reasoning": "",
"reasoning": None,
"content": None,
"is_reasoning_end": True,
}
+138 -51
View File
@@ -8,31 +8,105 @@ from unittest.mock import MagicMock
import pytest
from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
from vllm.tool_parsers.gemma4_tool_parser import (
from vllm.parser.gemma4 import (
TOOL_CALL_END,
TOOL_CALL_START,
Gemma4ToolParser,
_parse_gemma4_args,
_parse_gemma4_array,
)
from vllm.tool_parsers.gemma4_engine_tool_parser import Gemma4EngineToolParser
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
TOOL_CALL_START_ID = 48
TOOL_CALL_END_ID = 49
CHANNEL_START = "<|channel>"
CHANNEL_END = "<channel|>"
CHANNEL_START_ID = 50
CHANNEL_END_ID = 51
def _make_tool(name, properties):
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
)
return ChatCompletionToolsParam(
type="function",
function={
"name": name,
"parameters": {"type": "object", "properties": properties},
},
)
_TOOLS = [
_make_tool(
"set_status",
{
"is_active": {"type": "boolean"},
"count": {"type": "integer"},
"score": {"type": "number"},
},
),
_make_tool(
"set_config",
{
"count": {"type": "integer"},
"active": {"type": "boolean"},
},
),
_make_tool(
"search",
{
"input": {
"type": "object",
"properties": {"all": {"type": "boolean"}},
},
},
),
_make_tool(
"set",
{
"flag": {"type": "boolean"},
"count": {"type": "integer"},
},
),
_make_tool(
"Edit",
{
"file_path": {"type": "string"},
"old_string": {"type": "string"},
"new_string": {"type": "string"},
"replace_all": {"type": "boolean"},
},
),
]
@pytest.fixture
def mock_tokenizer():
vocab = {
TOOL_CALL_START: TOOL_CALL_START_ID,
TOOL_CALL_END: TOOL_CALL_END_ID,
CHANNEL_START: CHANNEL_START_ID,
CHANNEL_END: CHANNEL_END_ID,
}
decode_map = {v: k for k, v in vocab.items()}
tokenizer = MagicMock()
tokenizer.encode.return_value = [1, 2, 3]
# Include the tool call start token in the vocab for the parser
tokenizer.get_vocab.return_value = {TOOL_CALL_START: 48, TOOL_CALL_END: 49}
tokenizer.get_vocab.return_value = vocab
tokenizer.decode.side_effect = lambda ids: decode_map.get(ids[0], f"tok{ids[0]}")
return tokenizer
@pytest.fixture
def parser(mock_tokenizer):
return Gemma4ToolParser(mock_tokenizer)
return Gemma4EngineToolParser(mock_tokenizer, tools=_TOOLS)
@pytest.fixture
@@ -49,6 +123,9 @@ def mock_request():
class TestParseGemma4Args:
"""Values are returned as strings; type coercion to proper JSON types
happens at the engine layer."""
def test_empty_string(self):
assert _parse_gemma4_args("") == {}
@@ -71,27 +148,23 @@ class TestParseGemma4Args:
def test_integer_value(self):
result = _parse_gemma4_args("count:42")
assert result == {"count": 42}
assert result == {"count": "42"}
def test_float_value(self):
result = _parse_gemma4_args("score:3.14")
assert result == {"score": 3.14}
assert result == {"score": "3.14"}
def test_boolean_true(self):
result = _parse_gemma4_args("flag:true")
assert result == {"flag": True}
assert result == {"flag": "true"}
def test_boolean_false(self):
result = _parse_gemma4_args("flag:false")
assert result == {"flag": False}
assert result == {"flag": "false"}
def test_null_value(self):
# Bare `null` must parse as None (Python), not the string "null".
# Without this, tool_choice=auto would emit `{"param": "null"}`
# instead of `{"param": null}` for nullable tool parameters.
result = _parse_gemma4_args("param:null")
assert result == {"param": None}
assert json.dumps(result) == '{"param": null}'
assert result == {"param": "null"}
def test_mixed_types(self):
result = _parse_gemma4_args(
@@ -99,9 +172,9 @@ class TestParseGemma4Args:
)
assert result == {
"name": "test",
"count": 42,
"active": True,
"score": 3.14,
"count": "42",
"active": "true",
"score": "3.14",
}
def test_nested_object(self):
@@ -112,6 +185,17 @@ class TestParseGemma4Args:
result = _parse_gemma4_args('items:[<|"|>a<|"|>,<|"|>b<|"|>]')
assert result == {"items": ["a", "b"]}
def test_delimited_keys_stripped(self):
"""Keys wrapped in <|"|> delimiters are stripped."""
result = _parse_gemma4_args('<|"|>location<|"|>:<|"|>Paris<|"|>')
assert result == {"location": "Paris"}
result = _parse_gemma4_args('outer:{<|"|>inner<|"|>:<|"|>val<|"|>}')
assert result == {"outer": {"inner": "val"}}
result = _parse_gemma4_args('<|"|>name<|"|>:<|"|>Alice<|"|>,count:42')
assert result == {"name": "Alice", "count": "42"}
def test_unterminated_string(self):
"""Unterminated strings should take everything after the delimiter."""
result = _parse_gemma4_args('key:<|"|>unterminated')
@@ -153,7 +237,7 @@ class TestParseGemma4Args:
# Non-partial mode parses trailing dot normally
result = _parse_gemma4_args("left:108.,right:22.8", partial=False)
assert result == {"left": 108.0, "right": 22.8}
assert result == {"left": "108.", "right": "22.8"}
@pytest.mark.timeout(5)
def test_malformed_partial_array(self):
@@ -172,7 +256,7 @@ class TestParseGemma4Array:
def test_bare_values(self):
result = _parse_gemma4_array("42,true,3.14")
assert result == [42, True, 3.14]
assert result == ["42", "true", "3.14"]
@pytest.mark.timeout(5)
def test_string_element_with_closing_bracket(self):
@@ -182,7 +266,7 @@ class TestParseGemma4Array:
@pytest.mark.timeout(5)
def test_stray_closing_bracket(self):
result = _parse_gemma4_array("42,]trailing")
assert result == [42]
assert result == ["42"]
def test_trailing_dot_float_partial_withheld(self):
"""Array elements with trailing dot withheld in partial mode."""
@@ -191,7 +275,7 @@ class TestParseGemma4Array:
# Stable elements before trailing-dot element are kept
result = _parse_gemma4_array("42,108.,3", partial=True)
assert result == [42]
assert result == ["42"]
# ---------------------------------------------------------------------------
@@ -297,9 +381,11 @@ class TestExtractToolCalls:
model_output = '<|tool_call>call:get_weather{location:<|"|>London'
result = parser.extract_tool_calls(model_output, mock_request)
# Incomplete — no <tool_call|> end marker, regex won't match
assert result.tools_called is False
assert result.content == model_output
assert result.tools_called is True
assert len(result.tool_calls) == 1
assert result.tool_calls[0].function.name == "get_weather"
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"location": "London"}
def test_hyphenated_function_name(self, parser, mock_request):
"""Ensure function names with hyphens are parsed correctly."""
@@ -345,8 +431,15 @@ class TestStreamingExtraction:
verifying that the accumulated argument deltas form valid JSON.
"""
_SPECIAL_TOKEN_IDS = {
TOOL_CALL_START: TOOL_CALL_START_ID,
TOOL_CALL_END: TOOL_CALL_END_ID,
CHANNEL_START: CHANNEL_START_ID,
CHANNEL_END: CHANNEL_END_ID,
}
def _simulate_streaming(
self, parser: Gemma4ToolParser, mock_request: Any, chunks: list[str]
self, parser: Any, mock_request: Any, chunks: list[str]
) -> list[tuple[Any, str]]:
"""Feed chunks through the streaming parser and collect results.
@@ -358,14 +451,17 @@ class TestStreamingExtraction:
for chunk in chunks:
current_text = previous_text + chunk
# Use token ID 48 for tool_call start, 49 for end, 0 otherwise
delta_token_ids: list[int] = []
if TOOL_CALL_START in chunk:
delta_token_ids.append(48)
elif TOOL_CALL_END in chunk:
delta_token_ids.append(49)
else:
delta_token_ids.append(0)
found: list[tuple[int, int]] = []
for token, tid in self._SPECIAL_TOKEN_IDS.items():
pos = 0
while True:
idx = chunk.find(token, pos)
if idx < 0:
break
found.append((idx, tid))
pos = idx + len(token)
found.sort()
delta_token_ids: list[int] = [tid for _, tid in found] if found else [0]
current_token_ids = previous_token_ids + delta_token_ids
@@ -551,10 +647,10 @@ class TestStreamingExtraction:
results = self._simulate_streaming(parser, mock_request, chunks)
args_text = self._collect_arguments(results)
if args_text:
parsed_args = json.loads(args_text)
assert parsed_args["count"] == 42
assert parsed_args["active"] is True
assert args_text is not None
parsed_args = json.loads(args_text)
assert parsed_args["count"] == 42
assert parsed_args["active"] is True
def test_streaming_boolean_split_across_chunks(self, parser, mock_request):
"""Boolean value split across token boundaries must not corrupt JSON."""
@@ -643,23 +739,15 @@ class TestStreamingExtraction:
)
def test_streaming_does_not_duplicate_plain_text_after_tool_call(
self, parser, mock_request, monkeypatch
self, parser, mock_request
):
"""Buffered plain text after a tool call must not corrupt current_text."""
captured_current_texts: list[str] = []
original_extract_streaming = parser._extract_streaming
def wrapped_extract_streaming(previous_text, current_text, delta_text):
captured_current_texts.append(current_text)
return original_extract_streaming(previous_text, current_text, delta_text)
monkeypatch.setattr(parser, "_extract_streaming", wrapped_extract_streaming)
"""Buffered plain text after a tool call must not corrupt content."""
chunks = [
"<|tool_call>",
"call:get_weather{",
'location:<|"|>Paris<|"|>}',
"<tool_call|><",
"<tool_call|>",
"<",
"div>",
]
@@ -668,8 +756,7 @@ class TestStreamingExtraction:
delta.content for delta, _ in results if delta is not None and delta.content
]
assert "".join(content_parts) == "<div>"
assert captured_current_texts[-1].endswith("<tool_call|><div>")
assert not captured_current_texts[-1].endswith("<tool_call|><<div>")
assert "<<div>" not in "".join(content_parts)
def test_streaming_html_argument_does_not_duplicate_tag_prefixes(
self, parser, mock_request
@@ -30,7 +30,9 @@ from openai.types.responses.tool_param import FunctionToolParam
from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
from vllm.tool_parsers.abstract_tool_parser import ToolParser
from vllm.tool_parsers.gemma4_tool_parser import Gemma4ToolParser
from vllm.tool_parsers.gemma4_engine_tool_parser import (
Gemma4EngineToolParser as Gemma4ToolParser,
)
def _get_weather_tool() -> FunctionToolParam:
@@ -59,10 +61,16 @@ def _build_responses_request(*, tool_choice: str) -> ResponsesRequest:
class _StubTokenizer:
"""Minimal tokenizer stub to satisfy ``Gemma4ToolParser.__init__``."""
"""Minimal tokenizer stub to satisfy ``Gemma4EngineToolParser.__init__``."""
def get_vocab(self) -> dict[str, int]:
return {"<|tool_call>": 256_000, "<tool_call|>": 256_001, '<|"|>': 52}
return {
"<|tool_call>": 256_000,
"<tool_call|>": 256_001,
'<|"|>': 52,
"<|channel>": 256_002,
"<channel|>": 256_003,
}
def test_gemma4_adjust_request_sets_skip_special_tokens_on_responses() -> None:
@@ -74,15 +82,14 @@ def test_gemma4_adjust_request_sets_skip_special_tokens_on_responses() -> None:
path, causing raw ``call:fn{...}`` text to leak via
``response.output_text.delta``.
"""
parser = Gemma4ToolParser.__new__(Gemma4ToolParser)
parser.model_tokenizer = _StubTokenizer()
parser = Gemma4ToolParser(_StubTokenizer())
request = _build_responses_request(tool_choice="auto")
assert request.skip_special_tokens is True, (
"Precondition: ResponsesRequest.skip_special_tokens default is True"
)
Gemma4ToolParser.adjust_request(parser, request)
parser.adjust_request(request)
assert request.skip_special_tokens is False
+20 -5
View File
@@ -182,6 +182,21 @@ class ParserEngine(Parser):
request.skip_special_tokens = False
return request
def _preprocess_feed(
self,
delta_text: str,
delta_token_ids: Sequence[int],
) -> tuple[str, Sequence[int]]:
return delta_text, delta_token_ids
def _feed(
self,
delta_text: str,
delta_token_ids: Sequence[int],
) -> list[SemanticEvent]:
delta_text, delta_token_ids = self._preprocess_feed(delta_text, delta_token_ids)
return self._engine.feed(delta_text, delta_token_ids)
# ── Schema-aware type correction ─────────────────────────────────
@staticmethod
@@ -340,7 +355,7 @@ class ParserEngine(Parser):
finished: bool,
) -> DeltaMessage | None:
self._check_skip_tool_parsing(request)
events = self._engine.feed(delta_text, delta_token_ids)
events = self._feed(delta_text, delta_token_ids)
if finished:
events.extend(self._engine.finish())
result = self._events_to_delta(events, finished=finished)
@@ -384,7 +399,7 @@ class ParserEngine(Parser):
request: ChatCompletionRequest | ResponsesRequest,
) -> tuple[str | None, str | None]:
self._reset()
events = self._engine.feed(model_output, [])
events = self._feed(model_output, [])
events.extend(self._engine.finish())
reasoning_parts: list[str] = []
@@ -417,7 +432,7 @@ class ParserEngine(Parser):
delta_token_ids: Sequence[int],
) -> DeltaMessage | None:
self.initialize_streaming()
events = self._engine.feed(delta_text, delta_token_ids)
events = self._feed(delta_text, delta_token_ids)
return self._strip_trailing_reasoning(self._events_to_delta(events))
# ── Non-streaming: extract_tool_calls ─────────────────────────────
@@ -477,7 +492,7 @@ class ParserEngine(Parser):
) -> DeltaMessage | None:
self.initialize_streaming()
self._check_skip_tool_parsing(request)
events = self._engine.feed(delta_text, delta_token_ids)
events = self._feed(delta_text, delta_token_ids)
return self._strip_trailing_reasoning(self._events_to_delta(events))
# ── Reasoning state queries ───────────────────────────────────────
@@ -537,7 +552,7 @@ class ParserEngine(Parser):
state that ``_build_extracted_result`` reads.
"""
self._reset(initial_state=initial_state)
events = self._engine.feed(text, token_ids)
events = self._feed(text, token_ids)
events.extend(self._engine.finish())
delta = self._events_to_delta(events)
@@ -8,8 +8,14 @@ names so that :class:`ReasoningParserManager` and
"""
from vllm.parser.engine.adapters import make_adapters
from vllm.parser.gemma4 import Gemma4Parser
from vllm.parser.qwen3 import Qwen3Parser
(
Gemma4ParserReasoningAdapter,
Gemma4ParserToolAdapter,
) = make_adapters(Gemma4Parser)
(
Qwen3ParserReasoningAdapter,
Qwen3ParserToolAdapter,
+557
View File
@@ -0,0 +1,557 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Gemma4 parser.
Handles channel-based reasoning plus custom tool call format in a single
state machine::
<|channel>thought
...reasoning...<channel|>
<|tool_call>call:func_name{key:<|"|>value<|"|>,num:42}<tool_call|>
"""
from __future__ import annotations
import functools
import json
from collections.abc import Sequence
from typing import TYPE_CHECKING
from vllm.entrypoints.openai.engine.protocol import DeltaMessage
from vllm.logger import init_logger
from vllm.parser.engine.events import EventType, SemanticEvent
from vllm.parser.engine.parser_engine import ParserEngine
from vllm.parser.engine.parser_engine_config import (
ParserEngineConfig,
ParserState,
Transition,
)
if TYPE_CHECKING:
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionRequest,
)
from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
from vllm.tokenizers import TokenizerLike
from vllm.tool_parsers.abstract_tool_parser import Tool
# Tokens the model generates that must not leak into response content.
_GEMMA4_MODEL_DROP_TOKENS: set[str] = {
# Turn boundaries
"<|turn>",
"<turn|>",
# Channel / reasoning
"<|channel>",
"<channel|>",
# Tool protocol tokens
"<|tool>",
"<tool|>",
"<|tool_call>",
"<tool_call|>",
"<|tool_response>",
"<tool_response|>",
'<|"|>',
# Thinking
"<|think|>",
# Multi-modal (defensive — not expected during text completion)
"<|image>",
"<|image|>",
"<image|>",
"<|audio>",
"<|audio|>",
"<audio|>",
"<|video|>",
}
CHANNEL_START = "<|channel>"
CHANNEL_END = "<channel|>"
TOOL_CALL_START = "<|tool_call>"
TOOL_CALL_END = "<tool_call|>"
STRING_DELIM = '<|"|>'
_DELIM_LEN = len(STRING_DELIM)
logger = init_logger(__name__)
# ---------------------------------------------------------------------------
# Gemma4 argument parser
# ---------------------------------------------------------------------------
_PARTIAL_DELIM_SUFFIXES = tuple(
STRING_DELIM[:k] for k in range(len(STRING_DELIM), 0, -1)
)
def _strip_partial_delim(value: str) -> str:
"""Strip a trailing partial ``STRING_DELIM`` prefix from *value*.
Prevents partial delimiters from leaking into the streamed JSON diff.
"""
for suffix in _PARTIAL_DELIM_SUFFIXES:
if value.endswith(suffix):
return value[: -len(suffix)]
return value
def _parse_gemma4_args(args_str: str, *, partial: bool = False) -> dict:
"""Parse Gemma4's custom key:value format into a Python dict.
Format examples::
location:<|"|>Tokyo<|"|>
location:<|"|>San Francisco<|"|>,unit:<|"|>celsius<|"|>
count:42,flag:true
nested:{inner_key:<|"|>val<|"|>}
items:[<|"|>a<|"|>,<|"|>b<|"|>]
Args:
args_str: The raw Gemma4 argument string.
partial: When True (streaming), bare values at end of string are
omitted because they may be incomplete and type-unstable
(e.g. partial boolean parsed as bare string).
Returns a dict ready for ``json.dumps()``.
"""
if not args_str or not args_str.strip():
return {}
result: dict = {}
i = 0
n = len(args_str)
while i < n:
while i < n and args_str[i] in (" ", ",", "\n", "\t"):
i += 1
if i >= n:
break
key_start = i
while i < n and args_str[i] != ":":
i += 1
if i >= n:
break
key = args_str[key_start:i].strip()
if key.startswith(STRING_DELIM) and key.endswith(STRING_DELIM):
key = key[_DELIM_LEN:-_DELIM_LEN]
i += 1
if i >= n:
if not partial:
result[key] = ""
break
while i < n and args_str[i] in (" ", "\n", "\t"):
i += 1
if i >= n:
if not partial:
result[key] = ""
break
if args_str[i : i + _DELIM_LEN] == STRING_DELIM:
i += _DELIM_LEN
val_start = i
end_pos = args_str.find(STRING_DELIM, i)
if end_pos == -1:
# Unterminated string — take rest, strip partial delimiter.
value = args_str[val_start:]
if partial:
value = _strip_partial_delim(value)
result[key] = value
break
result[key] = args_str[val_start:end_pos]
i = end_pos + _DELIM_LEN
elif args_str[i] == "{":
depth = 1
obj_start = i + 1
i += 1
while i < n and depth > 0:
if args_str[i : i + _DELIM_LEN] == STRING_DELIM:
# Skip over string contents to avoid counting { inside strings
i += _DELIM_LEN
next_delim = args_str.find(STRING_DELIM, i)
i = n if next_delim == -1 else next_delim + _DELIM_LEN
continue
if args_str[i] == "{":
depth += 1
elif args_str[i] == "}":
depth -= 1
i += 1
if depth > 0:
# Incomplete nested object — use i (not i-1) to avoid
# dropping the last char, and recurse as partial.
result[key] = _parse_gemma4_args(args_str[obj_start:i], partial=True)
else:
result[key] = _parse_gemma4_args(args_str[obj_start : i - 1])
elif args_str[i] == "[":
depth = 1
arr_start = i + 1
i += 1
while i < n and depth > 0:
if args_str[i : i + _DELIM_LEN] == STRING_DELIM:
i += _DELIM_LEN
next_delim = args_str.find(STRING_DELIM, i)
i = n if next_delim == -1 else next_delim + _DELIM_LEN
continue
if args_str[i] == "[":
depth += 1
elif args_str[i] == "]":
depth -= 1
i += 1
if depth > 0:
result[key] = _parse_gemma4_array(args_str[arr_start:i], partial=True)
else:
result[key] = _parse_gemma4_array(args_str[arr_start : i - 1])
else:
val_start = i
while i < n and args_str[i] not in (",", "}", "]"):
i += 1
if partial and i >= n:
# Value may be incomplete (e.g. partial boolean) —
# withhold to avoid type instability during streaming.
break
if i == val_start:
logger.warning(
"Gemma4 args parser made no progress at position %d; "
"aborting on malformed input.",
i,
)
break
raw_val = args_str[val_start:i].strip()
if partial and raw_val.endswith("."):
# Digits may still arrive (e.g. "108." -> "108.2");
# withhold to avoid corrupting the streaming diff.
break
result[key] = raw_val
return result
def _parse_gemma4_array(arr_str: str, *, partial: bool = False) -> list:
items: list = []
i = 0
n = len(arr_str)
while i < n:
while i < n and arr_str[i] in (" ", ",", "\n", "\t"):
i += 1
if i >= n:
break
if arr_str[i : i + _DELIM_LEN] == STRING_DELIM:
i += _DELIM_LEN
end_pos = arr_str.find(STRING_DELIM, i)
if end_pos == -1:
items.append(arr_str[i:])
break
items.append(arr_str[i:end_pos])
i = end_pos + _DELIM_LEN
elif arr_str[i] == "{":
depth = 1
obj_start = i + 1
i += 1
while i < n and depth > 0:
if arr_str[i : i + _DELIM_LEN] == STRING_DELIM:
i += _DELIM_LEN
nd = arr_str.find(STRING_DELIM, i)
i = nd + _DELIM_LEN if nd != -1 else n
continue
if arr_str[i] == "{":
depth += 1
elif arr_str[i] == "}":
depth -= 1
i += 1
if depth > 0:
items.append(_parse_gemma4_args(arr_str[obj_start:i], partial=True))
else:
items.append(_parse_gemma4_args(arr_str[obj_start : i - 1]))
elif arr_str[i] == "[":
depth = 1
sub_start = i + 1
i += 1
while i < n and depth > 0:
if arr_str[i : i + _DELIM_LEN] == STRING_DELIM:
i += _DELIM_LEN
nd = arr_str.find(STRING_DELIM, i)
i = nd + _DELIM_LEN if nd != -1 else n
continue
if arr_str[i] == "[":
depth += 1
elif arr_str[i] == "]":
depth -= 1
i += 1
if depth > 0:
items.append(_parse_gemma4_array(arr_str[sub_start:i], partial=True))
else:
items.append(_parse_gemma4_array(arr_str[sub_start : i - 1]))
else:
val_start = i
while i < n and arr_str[i] not in (",", "]"):
i += 1
if partial and i >= n:
break
if i == val_start:
logger.warning(
"Gemma4 array parser made no progress at position %d; "
"aborting on malformed input.",
i,
)
break
raw_val = arr_str[val_start:i].strip()
if partial and raw_val.endswith("."):
break
items.append(raw_val)
return items
def _gemma4_arg_converter(raw_args: str, partial: bool) -> str:
"""Convert Gemma4 custom arg format to a JSON string."""
text = raw_args.strip()
if text.endswith("}"):
text = text[:-1]
parsed = _parse_gemma4_args(text, partial=partial)
return json.dumps(parsed, ensure_ascii=False)
@functools.cache
def gemma4_config() -> ParserEngineConfig:
used_tokens = {
CHANNEL_START,
CHANNEL_END,
TOOL_CALL_START,
TOOL_CALL_END,
'<|"|>',
}
return ParserEngineConfig(
name="gemma4",
initial_state=ParserState.CONTENT,
terminals={
"THINK_START": CHANNEL_START,
"THINK_END": CHANNEL_END,
"TOOL_START": TOOL_CALL_START,
"TOOL_END": TOOL_CALL_END,
"CALL_PREFIX": "call:",
"OPEN_BRACE": "{",
},
token_id_terminals={
"THINK_START": CHANNEL_START,
"THINK_END": CHANNEL_END,
"TOOL_START": TOOL_CALL_START,
"TOOL_END": TOOL_CALL_END,
},
transitions={
# -- Reasoning transitions --
(ParserState.CONTENT, "THINK_START"): Transition(
ParserState.REASONING,
(EventType.REASONING_START,),
),
(ParserState.REASONING, "THINK_END"): Transition(
ParserState.CONTENT,
(EventType.REASONING_END,),
),
# Tool call directly from reasoning (no explicit <channel|>)
(ParserState.REASONING, "TOOL_START"): Transition(
ParserState.TOOL_PREAMBLE,
(EventType.REASONING_END, EventType.TOOL_CALL_START),
),
# -- Tool call transitions --
(ParserState.CONTENT, "TOOL_START"): Transition(
ParserState.TOOL_PREAMBLE,
(EventType.REASONING_END, EventType.TOOL_CALL_START),
),
(ParserState.TOOL_PREAMBLE, "CALL_PREFIX"): Transition(
ParserState.TOOL_NAME,
(),
),
(ParserState.TOOL_NAME, "OPEN_BRACE"): Transition(
ParserState.TOOL_ARGS,
(),
),
(ParserState.TOOL_ARGS, "TOOL_END"): Transition(
ParserState.CONTENT,
(EventType.TOOL_CALL_END,),
),
# Back-to-back tool calls
(ParserState.CONTENT, "TOOL_END"): Transition(
ParserState.CONTENT,
(),
),
# Absorb a bare <channel|> that arrives after we already
# returned to CONTENT; prevents leaking it as TEXT_CHUNK.
(ParserState.CONTENT, "THINK_END"): Transition(
ParserState.CONTENT,
(),
),
},
content_events={
ParserState.CONTENT: EventType.TEXT_CHUNK,
ParserState.REASONING: EventType.REASONING_CHUNK,
ParserState.TOOL_NAME: EventType.TOOL_NAME,
ParserState.TOOL_ARGS: EventType.ARG_VALUE_CHUNK,
},
arg_converter=_gemma4_arg_converter,
tool_args_json=False,
arg_structural_chars=frozenset(",:{}[]<"),
drop_tokens=frozenset(_GEMMA4_MODEL_DROP_TOKENS - used_tokens),
)
_GEMMA4_THOUGHT_PREFIX = "thought\n"
_GEMMA4_THOUGHT_TOKEN = "thought"
class Gemma4Parser(ParserEngine):
"""Gemma4 parser: ``<|channel>`` reasoning + ``<|tool_call>``
tool calls in a single engine.
- Strips the ``thought\\n`` prefix from reasoning content
- Sets ``skip_special_tokens=False`` so boundary tokens are visible
- Detects ``<|tool_call>`` token as implicit reasoning end
"""
def __init__(
self,
tokenizer: TokenizerLike,
tools: list[Tool] | None = None,
**kwargs,
) -> None:
super().__init__(
tokenizer,
tools,
parser_engine_config=gemma4_config(),
**kwargs,
)
vocab = self.vocab
self._tool_call_token_id: int | None = vocab.get("<|tool_call>")
self._new_turn_token_id: int | None = vocab.get("<|turn>")
self._tool_response_token_id: int | None = vocab.get("<|tool_response>")
self._reasoning_text: str = ""
self._prefix_stripped: bool = False
self._is_first_feed: bool = True
def _reset(self, initial_state=None) -> None:
super()._reset(initial_state=initial_state)
self._reasoning_text = ""
self._prefix_stripped = False
self._is_first_feed = True
def _preprocess_feed(
self,
delta_text: str,
delta_token_ids: Sequence[int],
) -> tuple[str, Sequence[int]]:
if not self._is_first_feed:
return delta_text, delta_token_ids
self._is_first_feed = False
if (
not delta_text
or self._engine.state != ParserState.CONTENT
or self._reasoning_start_token_id is None
or self._reasoning_end_token_id is None
):
return delta_text, delta_token_ids
if CHANNEL_START in delta_text:
return delta_text, delta_token_ids
needs_injection = (
CHANNEL_END in delta_text
or delta_text.startswith(_GEMMA4_THOUGHT_PREFIX)
or delta_text == _GEMMA4_THOUGHT_TOKEN
)
if not needs_injection:
return delta_text, delta_token_ids
delta_text = CHANNEL_START + delta_text
if delta_token_ids:
delta_token_ids = [self._reasoning_start_token_id, *delta_token_ids]
return delta_text, delta_token_ids
def is_reasoning_end(self, input_ids: list[int]) -> bool:
end_id = self._reasoning_end_token_id
start_id = self._reasoning_start_token_id
tool_call_id = self._tool_call_token_id
new_turn_id = self._new_turn_token_id
tool_response_id = self._tool_response_token_id
if end_id is not None and not input_ids:
return self.parser_engine_config.initial_state != ParserState.REASONING
for i in range(len(input_ids) - 1, -1, -1):
tid = input_ids[i]
if start_id is not None and tid == start_id:
return False
if tool_call_id is not None and tid == tool_call_id:
return True
if new_turn_id is not None and tid == new_turn_id:
return False
if tool_response_id is not None and tid == tool_response_id:
return False
if end_id is not None and tid == end_id:
return True
return self._reasoning_ended
def _events_to_delta(
self,
events: list[SemanticEvent],
finished: bool = False,
) -> DeltaMessage | None:
delta = super()._events_to_delta(events, finished=finished)
if delta is None or delta.reasoning is None:
return delta
if self._prefix_stripped:
return delta
self._reasoning_text += delta.reasoning
if self._reasoning_text.startswith(_GEMMA4_THOUGHT_PREFIX):
prefix_len = len(_GEMMA4_THOUGHT_PREFIX)
prev_reasoning_len = len(self._reasoning_text) - len(delta.reasoning)
if prev_reasoning_len >= prefix_len:
self._prefix_stripped = True
return delta
chars_of_prefix_in_delta = prefix_len - prev_reasoning_len
stripped = delta.reasoning[chars_of_prefix_in_delta:]
if stripped:
self._prefix_stripped = True
delta.reasoning = stripped
return delta
if len(self._reasoning_text) >= prefix_len:
self._prefix_stripped = True
delta.reasoning = None
if delta.content is not None or delta.tool_calls:
return delta
return None
return None
if _GEMMA4_THOUGHT_PREFIX.startswith(self._reasoning_text):
if finished:
self._prefix_stripped = True
return None
self._prefix_stripped = True
delta.reasoning = self._reasoning_text
return delta
def extract_reasoning(
self,
model_output: str,
request: ChatCompletionRequest | ResponsesRequest,
) -> tuple[str | None, str | None]:
reasoning, content = super().extract_reasoning(model_output, request)
if reasoning:
if reasoning.startswith(_GEMMA4_THOUGHT_PREFIX):
reasoning = reasoning[len(_GEMMA4_THOUGHT_PREFIX) :]
elif reasoning == _GEMMA4_THOUGHT_PREFIX.rstrip():
reasoning = None
return reasoning or None, content
+1 -1
View File
@@ -118,7 +118,7 @@ def qwen3_config(thinking: bool = True) -> ParserEngineConfig:
# -- Tool call transitions --
(ParserState.CONTENT, "TOOL_START"): Transition(
ParserState.TOOL_PREAMBLE,
(EventType.TOOL_CALL_START,),
(EventType.REASONING_END, EventType.TOOL_CALL_START),
),
# Fallback: <function= without a preceding <tool_call>
(ParserState.CONTENT, "FUNC_PREFIX"): Transition(
+2 -2
View File
@@ -49,8 +49,8 @@ _REASONING_PARSERS_TO_REGISTER = {
"Ernie45ReasoningParser",
),
"gemma4": (
"gemma4_reasoning_parser",
"Gemma4ReasoningParser",
"gemma4_engine_reasoning_parser",
"Gemma4ParserReasoningAdapter",
),
"glm45": (
"deepseek_v3_reasoning_parser",
@@ -0,0 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from vllm.parser.engine.registered_adapters import Gemma4ParserReasoningAdapter
__all__ = ["Gemma4ParserReasoningAdapter"]
-225
View File
@@ -1,225 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from collections.abc import Sequence
from typing import TYPE_CHECKING
from vllm.entrypoints.openai.engine.protocol import DeltaMessage
from vllm.reasoning.basic_parsers import BaseThinkingReasoningParser
from vllm.tokenizers import TokenizerLike
if TYPE_CHECKING:
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionRequest,
)
from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
# Role label that Gemma4 emits at the start of the thinking channel.
# The model generates: <|channel>thought\n...reasoning...<channel|>
# This prefix must be stripped to expose only the actual reasoning content.
_THOUGHT_PREFIX = "thought\n"
class Gemma4ReasoningParser(BaseThinkingReasoningParser):
"""
Reasoning parser for Google Gemma4 thinking models.
Gemma4 uses <|channel>...<channel|> tokens to delimit reasoning/thinking
content within its output. Thinking mode is activated by passing
``enable_thinking=True`` in the chat template kwargs, which injects a
system turn containing <|think|> (token 98) to trigger chain-of-thought
reasoning.
Output pattern when thinking is enabled::
<|channel>thought
...chain of thought reasoning...<channel|>
Final answer text here.
The ``thought\\n`` role label inside the channel delimiters is a
structural artefact (analogous to ``user\\n`` in ``<|turn>user\\n...``).
This parser strips it so that downstream consumers see only the
actual reasoning text, consistent with the offline parser
(``vllm.reasoning.gemma4_utils._strip_thought_label``).
"""
def __init__(self, tokenizer: TokenizerLike, *args, **kwargs):
super().__init__(tokenizer, *args, **kwargs)
# Instance state for streaming prefix stripping.
# Tracks only the reasoning text received from the base parser,
# independent of current_text (which may contain pre-reasoning
# content and lacks special token text due to
# skip_special_tokens=True).
self._reasoning_text: str = ""
self._prefix_stripped: bool = False
self.new_turn_token_id = self.vocab["<|turn>"]
self.tool_call_token_id = self.vocab["<|tool_call>"]
self.tool_response_token_id = self.vocab["<|tool_response>"]
def adjust_request(
self, request: "ChatCompletionRequest | ResponsesRequest"
) -> "ChatCompletionRequest | ResponsesRequest":
"""Disable special-token stripping to preserve boundary tokens."""
request.skip_special_tokens = False
return request
@property
def start_token(self) -> str:
"""The token that starts reasoning content."""
return "<|channel>"
@property
def end_token(self) -> str:
"""The token that ends reasoning content."""
return "<channel|>"
def is_reasoning_end(self, input_ids: Sequence[int]) -> bool:
start_token_id = self.start_token_id
end_token_id = self.end_token_id
new_turn_token_id = self.new_turn_token_id
tool_call_token_id = self.tool_call_token_id
tool_response_token_id = self.tool_response_token_id
# Search from the end of input_ids to find the last match.
for i in range(len(input_ids) - 1, -1, -1):
if input_ids[i] == start_token_id:
return False
if input_ids[i] == tool_call_token_id:
# We're generating a tool call, so reasoning must be ended.
return True
if input_ids[i] in (new_turn_token_id, tool_response_token_id):
# We found a new turn or tool response token so don't consider
# reasoning ended yet, since the model starts new reasoning
# after these tokens.
return False
if input_ids[i] == end_token_id:
return True
return False
# ------------------------------------------------------------------
# Non-streaming path
# ------------------------------------------------------------------
def extract_reasoning(
self,
model_output: str,
request: "ChatCompletionRequest | ResponsesRequest",
) -> tuple[str | None, str | None]:
"""Extract reasoning, stripping the ``thought\\n`` role label."""
if self.start_token not in model_output and self.end_token not in model_output:
# Default to content history if no tags are present
# (or if they were stripped)
return None, model_output
reasoning, content = super().extract_reasoning(model_output, request)
if reasoning is not None:
reasoning = _strip_thought_label(reasoning)
return reasoning, content
# ------------------------------------------------------------------
# Streaming path
# ------------------------------------------------------------------
def extract_reasoning_streaming(
self,
previous_text: str,
current_text: str,
delta_text: str,
previous_token_ids: Sequence[int],
current_token_ids: Sequence[int],
delta_token_ids: Sequence[int],
) -> DeltaMessage | None:
"""Extract streaming reasoning, stripping ``thought\\n`` from the
first reasoning delta(s).
The ``thought\\n`` prefix may arrive as a single delta or split
across multiple deltas (e.g. ``"thought"`` then ``"\\n"``). We
buffer early reasoning tokens until we can determine whether the
prefix is present, then emit the buffered content minus the
prefix.
Unlike the previous implementation which reconstructed accumulated
reasoning from ``current_text``, this uses instance state
(``_reasoning_text``) to track only the reasoning content returned
by the base parser. This is necessary because
``skip_special_tokens=True`` (the vLLM default) causes the
``<|channel>`` delimiter to be invisible in ``current_text``,
making it impossible to separate pre-reasoning content from
reasoning content via string matching.
"""
result = super().extract_reasoning_streaming(
previous_text,
current_text,
delta_text,
previous_token_ids,
current_token_ids,
delta_token_ids,
)
if result is None:
return None
if result.reasoning is None:
return result
# Accumulate ONLY the reasoning text from base parser results.
# This is immune to pre-reasoning content pollution.
self._reasoning_text += result.reasoning
# Once the prefix has been handled, all subsequent reasoning
# deltas pass through unchanged.
if self._prefix_stripped:
return result
# ---- Prefix stripping logic ----
# Case 1: We've accumulated enough to confirm the prefix is
# present. Strip it and pass through the remainder.
if self._reasoning_text.startswith(_THOUGHT_PREFIX):
prefix_len = len(_THOUGHT_PREFIX)
# How much reasoning was accumulated before this delta?
prev_reasoning_len = len(self._reasoning_text) - len(result.reasoning)
if prev_reasoning_len >= prefix_len:
# Prefix was already consumed by prior deltas; this
# delta is entirely real content — pass through.
self._prefix_stripped = True
return result
else:
# Part or all of the prefix is in this delta.
chars_of_prefix_in_delta = prefix_len - prev_reasoning_len
stripped = result.reasoning[chars_of_prefix_in_delta:]
if stripped:
self._prefix_stripped = True
result.reasoning = stripped
return result
else:
if len(self._reasoning_text) >= prefix_len:
self._prefix_stripped = True
result.reasoning = ""
return result
return None
# Case 2: Accumulated text is a strict prefix of
# _THOUGHT_PREFIX (e.g. we've only seen "thou" so far).
# Buffer by suppressing — we can't yet tell if this will
# become the full prefix or diverge.
if _THOUGHT_PREFIX.startswith(self._reasoning_text):
return None
# Case 3: Accumulated text doesn't match the thought prefix
# at all. This means prior deltas were buffered (suppressed
# by Case 2) but the text diverged. Re-emit the full
# accumulated text to avoid data loss.
self._prefix_stripped = True
result.reasoning = self._reasoning_text
return result
def _strip_thought_label(text: str) -> str:
"""Remove the ``thought\\n`` role label from the beginning of text.
Mirrors ``vllm.reasoning.gemma4_utils._strip_thought_label`` from the
offline parser.
"""
if text.startswith(_THOUGHT_PREFIX):
return text[len(_THOUGHT_PREFIX) :]
return text
+2 -2
View File
@@ -191,8 +191,8 @@ _TOOL_PARSERS_TO_REGISTER = {
"FunctionGemmaToolParser",
),
"gemma4": (
"gemma4_tool_parser",
"Gemma4ToolParser",
"gemma4_engine_tool_parser",
"Gemma4EngineToolParser",
),
"apertus": (
"apertus_tool_parser",
@@ -0,0 +1,8 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from vllm.parser.engine.registered_adapters import Gemma4ParserToolAdapter
class Gemma4EngineToolParser(Gemma4ParserToolAdapter): # type: ignore[valid-type, misc]
supports_required_and_named = False
-896
View File
@@ -1,896 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""
Tool call parser for Google Gemma4 models.
Gemma4 uses a custom serialization format (not JSON) for tool calls::
<|tool_call>call:func_name{key:<|"|>value<|"|>,num:42}<tool_call|>
Strings are delimited by ``<|"|>`` (token 52), keys are unquoted, and
multiple tool calls are concatenated without separators.
Used when ``--enable-auto-tool-choice --tool-call-parser gemma4`` are set.
For offline inference tool call parsing (direct ``tokenizer.decode()`` output),
see ``vllm.tool_parsers.gemma4_utils.parse_tool_calls``.
"""
import json
from collections.abc import Sequence
import regex as re
from openai.types.responses import ToolChoiceFunction
from vllm.entrypoints.chat_utils import make_tool_call_id
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionNamedToolChoiceParam,
ChatCompletionRequest,
)
from vllm.entrypoints.openai.engine.protocol import (
DeltaFunctionCall,
DeltaMessage,
DeltaToolCall,
ExtractedToolCallInformation,
FunctionCall,
ToolCall,
)
from vllm.entrypoints.openai.responses.protocol import (
ResponsesRequest,
)
from vllm.logger import init_logger
from vllm.tokenizers import TokenizerLike
from vllm.tool_parsers.abstract_tool_parser import Tool, ToolParser
from vllm.tool_parsers.utils import find_common_prefix
logger = init_logger(__name__)
# Gemma4 special tokens for tool calls
TOOL_CALL_START = "<|tool_call>"
TOOL_CALL_END = "<tool_call|>"
STRING_DELIM = '<|"|>'
# ---------------------------------------------------------------------------
# Gemma4 argument parser (used by both streaming and non-streaming paths)
# ---------------------------------------------------------------------------
def _parse_gemma4_value(value_str: str) -> object:
"""Parse a single Gemma4 value (after key:) into a Python object."""
value_str = value_str.strip()
if not value_str:
return value_str
# Boolean
if value_str == "true":
return True
if value_str == "false":
return False
# Null
if value_str.lower() in ("null", "none", "nil"):
return None
# Number (int or float)
try:
if "." in value_str:
return float(value_str)
return int(value_str)
except ValueError:
pass
# Bare string (no <|"|> delimiters — shouldn't happen but be safe)
return value_str
def _parse_gemma4_args(args_str: str, *, partial: bool = False) -> dict:
"""Parse Gemma4's custom key:value format into a Python dict.
Format examples::
location:<|"|>Tokyo<|"|>
location:<|"|>San Francisco<|"|>,unit:<|"|>celsius<|"|>
count:42,flag:true
nested:{inner_key:<|"|>val<|"|>}
items:[<|"|>a<|"|>,<|"|>b<|"|>]
Args:
args_str: The raw Gemma4 argument string.
partial: When True (streaming), bare values at end of string are
omitted because they may be incomplete and type-unstable
(e.g. partial boolean parsed as bare string).
Returns a dict ready for ``json.dumps()``.
"""
if not args_str or not args_str.strip():
return {}
result: dict = {}
i = 0
n = len(args_str)
while i < n:
# Skip whitespace and commas
while i < n and args_str[i] in (" ", ",", "\n", "\t"):
i += 1
if i >= n:
break
# Parse key (unquoted, ends at ':')
key_start = i
while i < n and args_str[i] != ":":
i += 1
if i >= n:
break
key = args_str[key_start:i].strip()
i += 1 # skip ':'
# Parse value
if i >= n:
if not partial:
result[key] = ""
break
# Skip whitespace after ':'
while i < n and args_str[i] in (" ", "\n", "\t"):
i += 1
if i >= n:
if not partial:
result[key] = ""
break
# String value: <|"|>...<|"|>
if args_str[i:].startswith(STRING_DELIM):
i += len(STRING_DELIM)
val_start = i
end_pos = args_str.find(STRING_DELIM, i)
if end_pos == -1:
# Unterminated string — take rest
result[key] = args_str[val_start:]
break
result[key] = args_str[val_start:end_pos]
i = end_pos + len(STRING_DELIM)
# Nested object: {...}
elif args_str[i] == "{":
depth = 1
obj_start = i + 1
i += 1
while i < n and depth > 0:
if args_str[i:].startswith(STRING_DELIM):
# Skip over string contents to avoid counting { inside strings
i += len(STRING_DELIM)
next_delim = args_str.find(STRING_DELIM, i)
i = n if next_delim == -1 else next_delim + len(STRING_DELIM)
continue
if args_str[i] == "{":
depth += 1
elif args_str[i] == "}":
depth -= 1
i += 1
if depth > 0:
# Incomplete nested object — use i (not i-1) to avoid
# dropping the last char, and recurse as partial.
result[key] = _parse_gemma4_args(args_str[obj_start:i], partial=True)
else:
result[key] = _parse_gemma4_args(args_str[obj_start : i - 1])
# Array: [...]
elif args_str[i] == "[":
depth = 1
arr_start = i + 1
i += 1
while i < n and depth > 0:
if args_str[i:].startswith(STRING_DELIM):
i += len(STRING_DELIM)
next_delim = args_str.find(STRING_DELIM, i)
i = n if next_delim == -1 else next_delim + len(STRING_DELIM)
continue
if args_str[i] == "[":
depth += 1
elif args_str[i] == "]":
depth -= 1
i += 1
if depth > 0:
result[key] = _parse_gemma4_array(args_str[arr_start:i], partial=True)
else:
result[key] = _parse_gemma4_array(args_str[arr_start : i - 1])
# Bare value (number, boolean, etc.)
else:
val_start = i
while i < n and args_str[i] not in (",", "}", "]"):
i += 1
if partial and i >= n:
# Value may be incomplete (e.g. partial boolean) —
# withhold to avoid type instability during streaming.
break
if i == val_start:
logger.warning(
"Gemma4 args parser made no progress at position %d; "
"aborting on malformed input.",
i,
)
break
if partial:
raw_val = args_str[val_start:i].strip()
if raw_val.endswith("."):
# Trailing dot means decimal digits may still arrive
# (e.g. "108." may become "108.2"). Parsing now would
# yield float("108.") == 108.0, whose json repr "108.0"
# corrupts the streaming diff when the true digit lands.
break
result[key] = _parse_gemma4_value(args_str[val_start:i])
return result
def _parse_gemma4_array(arr_str: str, *, partial: bool = False) -> list:
"""Parse a Gemma4 array content string into a Python list."""
items: list = []
i = 0
n = len(arr_str)
while i < n:
while i < n and arr_str[i] in (" ", ",", "\n", "\t"):
i += 1
if i >= n:
break
# String element
if arr_str[i:].startswith(STRING_DELIM):
i += len(STRING_DELIM)
end_pos = arr_str.find(STRING_DELIM, i)
if end_pos == -1:
items.append(arr_str[i:])
break
items.append(arr_str[i:end_pos])
i = end_pos + len(STRING_DELIM)
# Nested object
elif arr_str[i] == "{":
depth = 1
obj_start = i + 1
i += 1
while i < n and depth > 0:
if arr_str[i:].startswith(STRING_DELIM):
i += len(STRING_DELIM)
nd = arr_str.find(STRING_DELIM, i)
i = nd + len(STRING_DELIM) if nd != -1 else n
continue
if arr_str[i] == "{":
depth += 1
elif arr_str[i] == "}":
depth -= 1
i += 1
if depth > 0:
items.append(_parse_gemma4_args(arr_str[obj_start:i], partial=True))
else:
items.append(_parse_gemma4_args(arr_str[obj_start : i - 1]))
# Nested array
elif arr_str[i] == "[":
depth = 1
sub_start = i + 1
i += 1
while i < n and depth > 0:
if arr_str[i:].startswith(STRING_DELIM):
i += len(STRING_DELIM)
nd = arr_str.find(STRING_DELIM, i)
i = nd + len(STRING_DELIM) if nd != -1 else n
continue
if arr_str[i] == "[":
depth += 1
elif arr_str[i] == "]":
depth -= 1
i += 1
if depth > 0:
items.append(_parse_gemma4_array(arr_str[sub_start:i], partial=True))
else:
items.append(_parse_gemma4_array(arr_str[sub_start : i - 1]))
# Bare value
else:
val_start = i
while i < n and arr_str[i] not in (",", "]"):
i += 1
if partial and i >= n:
break
if i == val_start:
logger.warning(
"Gemma4 array parser made no progress at position %d; "
"aborting on malformed input.",
i,
)
break
if partial:
raw_val = arr_str[val_start:i].strip()
if raw_val.endswith("."):
break
items.append(_parse_gemma4_value(arr_str[val_start:i]))
return items
# ---------------------------------------------------------------------------
# Parser
# ---------------------------------------------------------------------------
class Gemma4ToolParser(ToolParser):
"""
Tool call parser for Google Gemma4 models.
Handles the Gemma4 function call format::
<|tool_call>call:func_name{key:<|"|>value<|"|>}<tool_call|>
Used when ``--enable-auto-tool-choice --tool-call-parser gemma4``
are set.
Streaming strategy: **accumulate-then-parse-then-diff**
Instead of trying to convert Gemma4's custom format to JSON
token-by-token (which fails because Gemma4 uses bare keys, custom
delimiters, and structural braces that differ from JSON), this parser:
1. Accumulates the raw Gemma4 argument string during streaming
2. Parses it with ``_parse_gemma4_args()`` into a Python dict
3. Converts to JSON with ``json.dumps()``
4. Diffs against the previously-streamed JSON string
5. Emits only the new JSON fragment as the delta
This follows the same pattern used by FunctionGemma, Hermes, and Llama
tool parsers.
"""
# Gemma4 emits native special-token tool calls, not generic JSON calls.
supports_required_and_named = False
def __init__(self, tokenizer: TokenizerLike, tools: list[Tool] | None = None):
super().__init__(tokenizer, tools)
if not self.model_tokenizer:
raise ValueError(
"The model tokenizer must be passed to the ToolParser "
"constructor during construction."
)
# Token strings
self.tool_call_start_token = TOOL_CALL_START
self.tool_call_end_token = TOOL_CALL_END
# Token IDs
self.tool_call_start_token_id = self.vocab.get(TOOL_CALL_START)
self.tool_call_end_token_id = self.vocab.get(TOOL_CALL_END)
if self.tool_call_start_token_id is None:
raise RuntimeError(
"Gemma4 ToolParser could not locate the tool call start "
f"token '{TOOL_CALL_START}' in the tokenizer!"
)
# Regex for non-streaming: extract complete tool calls.
# Supports function names with letters, digits, underscores,
# hyphens, and dots (e.g. "get-weather", "module.func").
self.tool_call_regex = re.compile(
r"<\|tool_call>call:([\w\-\.]+)\{(.*?)\}<tool_call\|>",
re.DOTALL,
)
# Streaming state — reset per-request via _reset_streaming_state()
self._reset_streaming_state()
# Delta buffer for handling multi-token special sequences
self.buffered_delta_text = ""
def _reset_streaming_state(self) -> None:
"""Reset all streaming state for a new request."""
self.current_tool_id = -1
self.current_tool_name_sent = False
self.prev_tool_call_arr: list[dict] = []
self.streamed_args_for_tool: list[str] = []
def adjust_request(
self, request: ChatCompletionRequest | ResponsesRequest
) -> ChatCompletionRequest | ResponsesRequest:
if request.tools:
tc = request.tool_choice
if tc == "required" or isinstance(
tc,
(ChatCompletionNamedToolChoiceParam, ToolChoiceFunction),
):
# Do NOT call super().adjust_request() for required/named tool
# choice. The base implementation injects a JSON-array
# `structured_outputs` schema and forces xgrammar guided
# decoding, which conflicts with Gemma4's native
# `<|tool_call>call:...` (non-JSON) tool syntax and crashes
# EngineCore under MTP spec decode. The streaming/extraction
# parser already handles the native output, so guided decoding
# is skipped here (mirrors the GLM4 precedent).
if request.tool_choice != "none":
request.skip_special_tokens = False
return request
request = super().adjust_request(request)
if request.tools and request.tool_choice != "none":
# Don't skip special tokens — <|tool_call> etc. are needed for
# the parser to detect tool calls. Apply to BOTH
# ChatCompletionRequest and ResponsesRequest (the previous
# isinstance(ChatCompletionRequest) guard caused tool-call
# delimiters to be stripped on /v1/responses, leaking raw
# `call:fn{...}` text via output_text.delta).
request.skip_special_tokens = False
return request
# ------------------------------------------------------------------
# Delta buffering for multi-token special sequences
# ------------------------------------------------------------------
def _buffer_delta_text(self, delta_text: str) -> str:
"""Buffer incoming delta text to handle multi-token special sequences.
Accumulates partial tokens that could be the start of
``<|tool_call>`` or ``<tool_call|>`` and only flushes them
when the complete sequence is recognized or the sequence breaks.
This prevents partial special tokens (e.g., ``<|tool``) from being
emitted prematurely as content text.
"""
combined = self.buffered_delta_text + delta_text
# Check if combined ends with a complete special token
if combined.endswith(TOOL_CALL_START) or combined.endswith(TOOL_CALL_END):
self.buffered_delta_text = ""
return combined
# Check if combined ends with a partial prefix of a special token
for tag in [TOOL_CALL_START, TOOL_CALL_END]:
for i in range(1, len(tag)):
if combined.endswith(tag[:i]):
self.buffered_delta_text = combined[-i:]
return combined[:-i]
# No partial match — flush everything
self.buffered_delta_text = ""
return combined
# ------------------------------------------------------------------
# Non-streaming extraction
# ------------------------------------------------------------------
def extract_tool_calls(
self,
model_output: str,
request: ChatCompletionRequest,
) -> ExtractedToolCallInformation:
if self.tool_call_start_token not in model_output:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
try:
matches = self.tool_call_regex.findall(model_output)
if not matches:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
tool_calls: list[ToolCall] = []
for func_name, args_str in matches:
arguments = _parse_gemma4_args(args_str)
tool_calls.append(
ToolCall(
type="function",
function=FunctionCall(
name=func_name,
arguments=json.dumps(arguments, ensure_ascii=False),
),
)
)
# Content = text before first tool call (if any)
content_end = model_output.find(self.tool_call_start_token)
content = model_output[:content_end].strip() if content_end > 0 else None
return ExtractedToolCallInformation(
tools_called=True,
tool_calls=tool_calls,
content=content if content else None,
)
except Exception:
logger.exception("Error extracting tool calls from Gemma4 response")
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
# ------------------------------------------------------------------
# Streaming extraction — accumulate-then-parse-then-diff
# ------------------------------------------------------------------
def extract_tool_calls_streaming(
self,
previous_text: str,
current_text: str,
delta_text: str,
previous_token_ids: Sequence[int],
current_token_ids: Sequence[int],
delta_token_ids: Sequence[int],
request: ChatCompletionRequest,
) -> DeltaMessage | None:
# Buffer delta text to handle multi-token special sequences
delta_text = self._buffer_delta_text(delta_text)
# Keep current_text from the upstream stream state. The buffered delta
# is only for emission, and must not be stitched back into the
# accumulated model text or normal content like "<div>" can be
# duplicated into "<<div>" when a tool call just ended.
# If no tool call token seen yet, emit as content
if self.tool_call_start_token not in current_text:
if delta_text:
return DeltaMessage(content=delta_text)
return None
try:
return self._extract_streaming(
previous_text=previous_text,
current_text=current_text,
delta_text=delta_text,
)
except Exception:
logger.exception("Error in Gemma4 streaming tool call extraction")
return None
def _extract_streaming(
self,
previous_text: str,
current_text: str,
delta_text: str,
) -> DeltaMessage | None:
"""Tag-counting streaming parser.
Uses the proven approach from FunctionGemma/Hermes: count start/end
tags in previous vs current text to determine phase, then
accumulate-parse-diff for arguments.
Format: ``<|tool_call>call:name{args}<tool_call|>``
"""
start_count = current_text.count(self.tool_call_start_token)
end_count = current_text.count(self.tool_call_end_token)
prev_start_count = previous_text.count(self.tool_call_start_token)
prev_end_count = previous_text.count(self.tool_call_end_token)
# Case 1: Not inside any tool call — emit as content
if (
start_count == end_count
and prev_end_count == end_count
and self.tool_call_end_token not in delta_text
):
if delta_text:
return DeltaMessage(content=delta_text)
return None
# Case 2: One or more new tool calls started in this delta.
# A single delta can batch several complete calls, so advance the
# tool id once per newly-seen start token and allocate a tracking
# slot for each.
if start_count > prev_start_count:
num_new = start_count - prev_start_count
for _ in range(num_new):
self.current_tool_id += 1
self.streamed_args_for_tool.append("")
self.prev_tool_call_arr.append({})
self.current_tool_name_sent = False
logger.debug(
"Started %d new tool call(s); current_tool_id=%d",
num_new,
self.current_tool_id,
)
# Don't return yet if this delta also contains call payload or
# the end marker; backends can batch one or more complete tool
# calls into a single streaming chunk. Only wait for more text
# when the delta is just the start token itself.
if start_count > end_count and len(delta_text) <= len(
self.tool_call_start_token
):
return None
# Case 3: One or more tool calls just ended (possibly several in a
# single batched delta) — drain every newly-completed call.
if end_count > prev_end_count:
return self._handle_tool_call_end(
current_text,
prev_end_count=prev_end_count,
end_count=end_count,
start_count=start_count,
)
# Case 4: In the middle of a tool call — parse partial content
if start_count > end_count:
return self._handle_tool_call_middle(current_text)
# Default: generate text outside tool calls
if delta_text:
text = delta_text.replace(self.tool_call_start_token, "")
text = text.replace(self.tool_call_end_token, "")
if text:
return DeltaMessage(content=text)
return None
def _extract_partial_call(self, current_text: str) -> tuple[str | None, str]:
"""Extract function name and raw argument string from partial text.
Returns (func_name, raw_args_str) or (None, "") if not parseable yet.
"""
# Get the text after the last <|tool_call> token
last_start = current_text.rfind(self.tool_call_start_token)
if last_start == -1:
return None, ""
partial_call = current_text[last_start + len(self.tool_call_start_token) :]
# Strip end token if present
if self.tool_call_end_token in partial_call:
partial_call = partial_call.split(self.tool_call_end_token)[0]
# Expect "call:name{args...}" or "call:name{args...}"
if not partial_call.startswith("call:"):
return None, ""
func_part = partial_call[5:] # skip "call:"
if "{" not in func_part:
# Still accumulating function name, not ready yet
return None, ""
func_name, _, args_part = func_part.partition("{")
func_name = func_name.strip()
# Strip trailing '}' if present (Gemma4 structural brace)
if args_part.endswith("}"):
args_part = args_part[:-1]
return func_name, args_part
def _handle_tool_call_middle(self, current_text: str) -> DeltaMessage | None:
"""Handle streaming when we're inside an active tool call.
Accumulates the raw Gemma4 arguments, parses them into JSON, and
diffs against the previously-streamed JSON to emit only the new
fragment.
"""
func_name, args_part = self._extract_partial_call(current_text)
if func_name is None:
return None
# Step 1: Send function name (once)
if not self.current_tool_name_sent and func_name:
self.current_tool_name_sent = True
self.prev_tool_call_arr[self.current_tool_id] = {
"name": func_name,
"arguments": {},
}
return DeltaMessage(
tool_calls=[
DeltaToolCall(
index=self.current_tool_id,
type="function",
id=make_tool_call_id(),
function=DeltaFunctionCall(
name=func_name,
arguments="",
).model_dump(exclude_none=True),
)
]
)
# Step 2: Parse and diff arguments
if self.current_tool_name_sent and args_part:
return self._emit_argument_diff(args_part)
return None
def _handle_tool_call_end(
self,
current_text: str,
prev_end_count: int,
end_count: int,
start_count: int,
) -> DeltaMessage | None:
"""Handle streaming when one or more tool calls have just completed.
A single streaming delta can batch several complete tool calls
(``<|tool_call>...<tool_call|><|tool_call>...<tool_call|>``). Every
call whose ``<tool_call|>`` end marker arrived in this delta — i.e.
those with index in ``[prev_end_count, end_count)`` — is drained and
emitted, with one ``DeltaToolCall`` per call in a single
``DeltaMessage`` (this matches the OpenAI streaming wire format, and
the serving layer iterates over ``delta.tool_calls``).
Per call:
* If the function name was already streamed incrementally (the
token-by-token path), only the remaining argument fragment is
flushed as a diff.
* If the call is seen complete for the first time in this delta (the
batched-complete path), the id + name + full arguments JSON are
emitted exactly once.
"""
# Parse the complete tool calls using regex for accuracy.
all_matches = self.tool_call_regex.findall(current_text)
if not all_matches:
logger.debug("Tool call end detected but no complete tool call parsed yet.")
return None
deltas: list[DeltaToolCall] = []
for idx in range(prev_end_count, end_count):
if idx >= len(all_matches):
break
# Ensure the tracking arrays have a slot for this index (defensive;
# Case 2 normally allocates these when the start token arrives).
while len(self.prev_tool_call_arr) <= idx:
self.prev_tool_call_arr.append({})
self.streamed_args_for_tool.append("")
func_name, args_str = all_matches[idx]
final_args = _parse_gemma4_args(args_str)
final_args_json = json.dumps(final_args, ensure_ascii=False)
# The name is sent exactly once per call. We track that via the
# per-call entry in prev_tool_call_arr (set either by the middle
# path or by the batched-complete branch below), which is robust
# even when several calls are drained in one delta.
name_already_sent = bool(self.prev_tool_call_arr[idx].get("name"))
if not name_already_sent:
# Batched-complete call: emit id + name + full arguments once.
self.streamed_args_for_tool[idx] = final_args_json
self.prev_tool_call_arr[idx] = {
"name": func_name,
"arguments": final_args,
}
deltas.append(
DeltaToolCall(
index=idx,
type="function",
id=make_tool_call_id(),
function=DeltaFunctionCall(
name=func_name, arguments=final_args_json
).model_dump(exclude_none=True),
)
)
else:
# Incrementally-streamed call: flush the remaining argument
# tail that was withheld during the middle phase.
prev_streamed = self.streamed_args_for_tool[idx]
if len(final_args_json) > len(prev_streamed):
diff = final_args_json[len(prev_streamed) :]
self.streamed_args_for_tool[idx] = final_args_json
self.prev_tool_call_arr[idx]["arguments"] = final_args
deltas.append(
DeltaToolCall(
index=idx,
function=DeltaFunctionCall(arguments=diff).model_dump(
exclude_none=True
),
)
)
# Advance streaming state past the calls completed in this delta. If a
# further tool call is still being accumulated (start without a
# matching end), point current_tool_id at it so the middle path can
# stream its arguments next; otherwise settle on the last completed
# call.
if start_count > end_count:
self.current_tool_id = end_count
while len(self.prev_tool_call_arr) <= self.current_tool_id:
self.prev_tool_call_arr.append({})
self.streamed_args_for_tool.append("")
self.current_tool_name_sent = bool(
self.prev_tool_call_arr[self.current_tool_id].get("name")
)
else:
self.current_tool_id = end_count - 1
self.current_tool_name_sent = True
if deltas:
return DeltaMessage(tool_calls=deltas)
return None
def _emit_argument_diff(self, raw_args_str: str) -> DeltaMessage | None:
"""Parse raw Gemma4 arguments, convert to JSON, diff, and emit.
This is the core of the accumulate-then-parse-then-diff strategy:
1. Parse ``raw_args_str`` with ``_parse_gemma4_args()``
2. Convert to JSON string with ``json.dumps()``
3. Withhold trailing closing characters (``"}``) that may move
as more tokens arrive
4. Diff against previously streamed JSON and emit only new chars
**Why withholding is necessary:**
Gemma4's custom format produces *structurally incomplete* JSON
during streaming. For example, when ``<|"|>Paris`` arrives
without a closing delimiter, ``_parse_gemma4_args`` treats it
as a complete value and produces ``{"location": "Paris"}``. But
when ``, France<|"|>`` arrives next, the JSON becomes
``{"location": "Paris, France"}``. If we had sent the closing
``"}`` from the first parse, the concatenated client output
would be ``{"location": "Paris"}France"}``, which is garbage.
The solution: **never send trailing closing chars during
streaming**. They get flushed by ``_handle_tool_call_end()``
when the ``<tool_call|>`` end marker arrives.
Args:
raw_args_str: The raw Gemma4 argument text accumulated so far
(without the surrounding ``{`` ``}``).
Returns:
DeltaMessage with the argument diff, or None if no new content.
"""
try:
current_args = _parse_gemma4_args(raw_args_str, partial=True)
except Exception:
logger.debug(
"Could not parse partial Gemma4 args yet: %s",
raw_args_str[:100],
)
return None
if not current_args:
return None
current_args_json = json.dumps(current_args, ensure_ascii=False)
# Withhold trailing closing characters that may shift as more
# tokens arrive. Strip trailing '}', '"', ']' and partial
# STRING_DELIM fragments ('<', '|', '\\', '>') to get the
# "safe prefix".
safe_json = current_args_json
while safe_json and safe_json[-1] in ("}", '"', "]", "<", "|", "\\", ">"):
safe_json = safe_json[:-1]
prev_streamed = self.streamed_args_for_tool[self.current_tool_id]
if not safe_json or safe_json == prev_streamed:
return None
# Use find_common_prefix to handle cases where the value changed
# structurally (e.g., a string grew).
if prev_streamed:
prefix = find_common_prefix(prev_streamed, safe_json)
sent_len = len(prev_streamed)
prefix_len = len(prefix)
if prefix_len < sent_len:
# Structure changed — we sent too much. Truncate our
# tracking to the common prefix and wait for the final
# flush in _handle_tool_call_end.
self.streamed_args_for_tool[self.current_tool_id] = prefix
return None
# Stream the new stable portion
diff = safe_json[sent_len:]
else:
# First emission
diff = safe_json
if diff:
self.streamed_args_for_tool[self.current_tool_id] = safe_json
self.prev_tool_call_arr[self.current_tool_id]["arguments"] = current_args
return DeltaMessage(
tool_calls=[
DeltaToolCall(
index=self.current_tool_id,
function=DeltaFunctionCall(arguments=diff).model_dump(
exclude_none=True
),
)
]
)
return None