From bced84e3c140a7d62cff399cc229b907cac38868 Mon Sep 17 00:00:00 2001 From: Maroon Ayoub Date: Thu, 29 Oct 2026 21:52:23 -0700 Subject: [PATCH] [Bugfix][Parser] Parse closed Qwen3 tool parameters once while streaming Every streaming conversion of Qwen3 XML tool arguments re-scanned the whole argument text, twice: once for the closed parameters and once to strip them before finding the open one. The value extender scanned it twice more. A model looping on closed parameters inside one call made the API server's cost per delta grow with the call until the event loop starved and liveness probes failed. Cache the closed parameters per parser instance, as Glm47MoeParser does, so each conversion and extender call resumes after the last closed parameter. Qwen3Parser and the parsers that inherit its converter (Nemotron-3, Step-3.5, Step5, Seed-OSS, MiniCPM-V, Granite thinking) share the change. Co-authored-by: Claude Opus 5.5 Signed-off-by: Maroon Ayoub --- tests/parser/engine/test_qwen3.py | 94 +++++++++++++++++++++++++++++++ vllm/parser/qwen3.py | 84 +++++++++++++++++++++------ 2 files changed, 160 insertions(+), 18 deletions(-) diff --git a/tests/parser/engine/test_qwen3.py b/tests/parser/engine/test_qwen3.py index be4b33e3861f..902a9f61b4f9 100644 --- a/tests/parser/engine/test_qwen3.py +++ b/tests/parser/engine/test_qwen3.py @@ -819,6 +819,14 @@ def test_partial_value_with_angle_bracket_and_complete_param(self): # Escapes, non-ASCII, blank lines and tag-like text that must stay inside # the value, including prefixes of the closing and next-parameter tags. CONTENT_LINE = 'if a < b > c: print("x\\y") #

\n\n\n/repo/app.py\n\n" +) +LOOP_PARAMS = ( + "\na > b\n\n\n644\n\n" +) +LOOP_TAIL = "\n" class TestArgValueExtender: @@ -908,6 +916,92 @@ def counting_converter(raw_args: str, partial: bool) -> str: assert streaming_conversions(lines=20) == streaming_conversions(lines=200) + @pytest.mark.parametrize("size", [1, 7]) + def test_repeated_closed_params_stream_non_streaming_arguments( + self, mock_tokenizer, mock_request, write_tool, size + ): + """Covers a duplicate open key and a second call sharing the first pair.""" + from vllm.parser.qwen3 import Qwen3Parser + + mock_request.tools = [write_tool] + text = ( + LOOP_HEAD + + LOOP_PARAMS * 20 + + "\na > b" + + WRITE_CALL_TAIL + + LOOP_HEAD + + LOOP_TAIL + ) + + def make_parser(): + return Qwen3Parser( + mock_tokenizer, chat_template_kwargs={"enable_thinking": False} + ) + + results = simulate_tool_streaming( + make_parser(), + mock_request, + [text[i : i + size] for i in range(0, len(text), size)], + ) + result = make_parser().extract_tool_calls(text, mock_request) + + assert len(result.tool_calls) == 2 + assert collect_tool_arguments(results) == "".join( + c.function.arguments for c in result.tool_calls + ) + + def test_closed_params_are_parsed_once( + self, mock_tokenizer, mock_request, write_tool, monkeypatch + ): + """Streaming cost must not grow with the closed parameters already seen.""" + from vllm.parser import qwen3 + + mock_request.tools = [write_tool] + param_re = qwen3._PARAM_RE + matches = 0 + + class CountingPattern: + def __getattr__(self, name): + return getattr(param_re, name) + + def finditer(self, *args): + nonlocal matches + for match in param_re.finditer(*args): + matches += 1 + yield match + + def sub(self, repl, *args): + def count(match): + nonlocal matches + matches += 1 + return repl + + return param_re.sub(count, *args) + + monkeypatch.setattr(qwen3, "_PARAM_RE", CountingPattern()) + text = LOOP_HEAD + LOOP_PARAMS * 200 + LOOP_TAIL + simulate_tool_streaming( + qwen3.Qwen3Parser( + mock_tokenizer, chat_template_kwargs={"enable_thinking": False} + ), + mock_request, + [text[i : i + 4] for i in range(0, len(text), 4)], + ) + + assert matches == len(param_re.findall(text)) + + def test_value_closed_by_next_parameter_needs_that_parameter(self): + from vllm.parser.qwen3 import _ClosedParams, _qwen3_arg_converter + + closed = _ClosedParams() + _qwen3_arg_converter( + "\n/a.py\n\n6", True, closed + ) + text = "\n/a.py\n.bak\n" + assert _qwen3_arg_converter(text, False, closed) == _qwen3_arg_converter( + text, False + ) + def test_extender_follows_newline_trimming_and_split_end(self): from vllm.parser.qwen3 import _qwen3_arg_value_extender diff --git a/vllm/parser/qwen3.py b/vllm/parser/qwen3.py index 577e01edf6ed..f99d54c0c8d8 100644 --- a/vllm/parser/qwen3.py +++ b/vllm/parser/qwen3.py @@ -56,6 +56,7 @@ re.DOTALL, ) _PARTIAL_PARAM_RE = re.compile(r"<\s*parameter\s*=\s*([^>]+)>(.*)$", re.DOTALL) +_PARAM_OPENER_RE = re.compile(r"<\s*parameter\s*=") def _trim_wrapping_newlines(value: str) -> str: @@ -67,22 +68,60 @@ def _trim_wrapping_newlines(value: str) -> str: return value -def _qwen3_arg_converter(raw_args: str, partial: bool) -> str: - params: dict[str, object] = {} +class _ClosedParams: + """Closed parameters of a raw argument text, reused by longer texts. - for match in _PARAM_RE.finditer(raw_args): - name = match.group(1) - value = match.group(2) - params[name] = _trim_wrapping_newlines(value) + Raw arguments only grow while a call streams, so a parse of a text that + extends ``text`` resumes after its last closed parameter. + """ + __slots__ = ("text", "end", "params", "outside") + + def __init__(self) -> None: + self.text = "" # raw text the closed parameters depend on + self.end = 0 # end of the last closed parameter in ``text`` + self.params: dict[str, str] = {} + self.outside = "" # text before ``end`` outside the closed parameters + + +def _qwen3_parse_params( + raw_args: str, partial: bool, closed: _ClosedParams | None = None +) -> tuple[dict[str, str], re.Match | None]: + """Return the closed parameters, trimmed, and the open trailing one.""" + params: dict[str, str] = {} + outside: list[str] = [] + pos = 0 + if closed is not None and raw_args.startswith(closed.text): + params.update(closed.params) + outside.append(closed.outside) + pos = closed.end + start = pos + + for match in _PARAM_RE.finditer(raw_args, pos): + outside.append(raw_args[pos : match.start()]) + params[match.group(1)] = _trim_wrapping_newlines(match.group(2)) + pos = match.end() + + if closed is not None and pos > start: + # A value closed by the next opener holds only while that opener follows. + opener = _PARAM_OPENER_RE.match(raw_args, pos) + closed.text = raw_args[: opener.end() if opener else pos] + closed.end = pos + closed.params = dict(params) + closed.outside = "".join(outside) + + open_param = None if partial: - remaining = _PARAM_RE.sub("", raw_args) - m = _PARTIAL_PARAM_RE.search(remaining) - if m: - name = m.group(1) - value = m.group(2) - if name: - params[name] = _trim_wrapping_newlines(value) + open_param = _PARTIAL_PARAM_RE.search("".join(outside) + raw_args[pos:]) + return params, open_param + + +def _qwen3_arg_converter( + raw_args: str, partial: bool, closed: _ClosedParams | None = None +) -> str: + params, open_param = _qwen3_parse_params(raw_args, partial, closed) + if open_param is not None and open_param.group(1): + params[open_param.group(1)] = _trim_wrapping_newlines(open_param.group(2)) return json.dumps(params, ensure_ascii=False) @@ -101,7 +140,7 @@ def _param_value_end_window(text: str) -> str: def _param_value_extender( - raw_args: str, trim_newlines: bool + raw_args: str, trim_newlines: bool, closed: _ClosedParams | None = None ) -> ArgValueExtender | None: """Extend the open trailing ```` value of *raw_args*. @@ -109,12 +148,12 @@ def _param_value_extender( With *trim_newlines* it follows ``_trim_wrapping_newlines``: it drops a leading newline of the value and holds back a trailing one. """ - match = _PARTIAL_PARAM_RE.search(_PARAM_RE.sub("", raw_args)) + closed_params, match = _qwen3_parse_params(raw_args, True, closed) if match is None: return None name, value = match.group(1), match.group(2) # A repeated key keeps its first position, so its value is not last. - if any(m.group(1) == name for m in _PARAM_RE.finditer(raw_args)): + if name in closed_params: return None window = _param_value_end_window(value) @@ -141,8 +180,10 @@ def extend(delta: str) -> str | None: return extend -def _qwen3_arg_value_extender(raw_args: str) -> ArgValueExtender | None: - return _param_value_extender(raw_args, trim_newlines=True) +def _qwen3_arg_value_extender( + raw_args: str, closed: _ClosedParams | None = None +) -> ArgValueExtender | None: + return _param_value_extender(raw_args, trim_newlines=True, closed=closed) @functools.cache @@ -333,6 +374,13 @@ def __init__( tools, **kwargs, ) + # Each streaming conversion resumes after the last closed parameter. + if self._arg_converter is _qwen3_arg_converter: + closed = _ClosedParams() + self._arg_converter = functools.partial(_qwen3_arg_converter, closed=closed) + self._arg_value_extender = functools.partial( + _qwen3_arg_value_extender, closed=closed + ) def extract_reasoning( self,