From 5fd8707f644d29c4887d5637f6152344bd38db6b Mon Sep 17 00:00:00 2001 From: Jun Yamog Date: Mon, 7 Sep 2026 13:02:41 +0000 Subject: [PATCH] Fix Qwen3 XML streaming tool call boundaries --- .../test_qwen3xml_chunk_boundaries.py | 126 ++++++++++++++++++ vllm/tool_parsers/qwen3xml_tool_parser.py | 16 ++- 2 files changed, 139 insertions(+), 3 deletions(-) create mode 100644 tests/tool_parsers/test_qwen3xml_chunk_boundaries.py diff --git a/tests/tool_parsers/test_qwen3xml_chunk_boundaries.py b/tests/tool_parsers/test_qwen3xml_chunk_boundaries.py new file mode 100644 index 0000000000..72e492f43d --- /dev/null +++ b/tests/tool_parsers/test_qwen3xml_chunk_boundaries.py @@ -0,0 +1,126 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""A closing tag in a chunk must not close the next tool call.""" + +import json +import random +from unittest.mock import Mock + +import pytest + +from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest +from vllm.tool_parsers.qwen3xml_tool_parser import ( + Qwen3XMLToolParser, + StreamingXMLToolCallParser, +) + +PARALLEL_CALLS = ( + "\n\n/tmp/a\n" + "\n\n" + "\n\npwd\n" + "\n" +) + + +def assert_parallel_calls(chunks): + parser = StreamingXMLToolCallParser() + calls = {} + for chunk in chunks: + delta = parser.parse_single_streaming_chunks(chunk) + for call in delta.tool_calls or []: + entry = calls.setdefault( + call.index, {"id": call.id, "name": "", "arguments": ""} + ) + assert call.id == entry["id"] + if call.function: + entry["name"] += call.function.name or "" + entry["arguments"] += call.function.arguments or "" + assert list(calls) == [0, 1] + assert calls[0]["id"] != calls[1]["id"] + assert [call["name"] for call in calls.values()] == ["read", "bash"] + assert [json.loads(call["arguments"]) for call in calls.values()] == [ + {"path": "/tmp/a"}, + {"command": "pwd"}, + ] + + +@pytest.mark.parametrize("width", range(1, len(PARALLEL_CALLS) + 1)) +def test_parallel_calls_fixed_chunks(width): + assert_parallel_calls( + PARALLEL_CALLS[pos : pos + width] + for pos in range(0, len(PARALLEL_CALLS), width) + ) + + +@pytest.mark.parametrize("split", range(1, len(PARALLEL_CALLS))) +def test_parallel_calls_single_split(split): + assert_parallel_calls([PARALLEL_CALLS[:split], PARALLEL_CALLS[split:]]) + + +@pytest.mark.parametrize("seed", range(100)) +def test_parallel_calls_irregular_chunks(seed): + rng = random.Random(seed) + chunks = [] + pos = 0 + while pos < len(PARALLEL_CALLS): + width = rng.randint(1, 50) + chunks.append(PARALLEL_CALLS[pos : pos + width]) + pos += width + assert_parallel_calls(chunks) + + +@pytest.mark.parametrize("width", [1, 25, 33, 58, 135, len(PARALLEL_CALLS)]) +def test_public_streaming_matches_nonstreaming(width): + parser = Qwen3XMLToolParser(Mock()) + request = ChatCompletionRequest(model="test", messages=[]) + expected = parser.extract_tool_calls(PARALLEL_CALLS, request) + assert len(expected.tool_calls) == 2 + calls = {} + previous = "" + for pos in range(0, len(PARALLEL_CALLS), width): + chunk = PARALLEL_CALLS[pos : pos + width] + current = previous + chunk + delta = parser.extract_tool_calls_streaming( + previous, current, chunk, [], [], [], request + ) + previous = current + if delta is None: + continue + for call in delta.tool_calls or []: + entry = calls.setdefault(call.index, {"name": "", "arguments": ""}) + if call.function: + entry["name"] += call.function.name or "" + entry["arguments"] += call.function.arguments or "" + assert list(calls) == [0, 1] + for index, call in enumerate(expected.tool_calls): + assert calls[index]["name"] == call.function.name + assert json.loads(calls[index]["arguments"]) == json.loads( + call.function.arguments + ) + assert [c["name"] for c in parser.prev_tool_call_arr] == ["read", "bash"] + + +@pytest.mark.parametrize("width", range(1, 81)) +def test_empty_and_multiline_calls(width): + text = ( + "\n" + "/tmp/a" + 'line 1\n "quoted" & x < y\nline 3' + "" + ) + parser = StreamingXMLToolCallParser() + calls = {} + for pos in range(0, len(text), width): + delta = parser.parse_single_streaming_chunks(text[pos : pos + width]) + for call in delta.tool_calls or []: + entry = calls.setdefault(call.index, {"name": "", "arguments": ""}) + if call.function: + entry["name"] += call.function.name or "" + entry["arguments"] += call.function.arguments or "" + assert list(calls) == [0, 1] + assert [call["name"] for call in calls.values()] == ["refresh", "write"] + assert json.loads(calls[0]["arguments"]) == {} + assert json.loads(calls[1]["arguments"]) == { + "path": "/tmp/a", + "content": 'line 1\n "quoted" & x < y\nline 3', + } diff --git a/vllm/tool_parsers/qwen3xml_tool_parser.py b/vllm/tool_parsers/qwen3xml_tool_parser.py index d5b87ea074..1b729a10aa 100644 --- a/vllm/tool_parsers/qwen3xml_tool_parser.py +++ b/vllm/tool_parsers/qwen3xml_tool_parser.py @@ -100,6 +100,9 @@ def parse_single_streaming_chunks(self, xml_chunk: str) -> DeltaMessage: self.streaming_buffer += xml_chunk + # A chunk may close one call and open another. Closing tags in that + # chunk must never make the fallback close the newly opened call. + initial_call_id = self.current_call_id found_elements = self._process_complete_xml_elements() if found_elements: @@ -111,6 +114,8 @@ def parse_single_streaming_chunks(self, xml_chunk: str) -> DeltaMessage: # but didn't generate '}', then complete it if ( self.current_call_id is not None + and self.current_call_id == initial_call_id + and self.current_function_open and self.function_end_token in xml_chunk ): # - Added '}' (non-empty parameter ending) @@ -140,6 +145,7 @@ def parse_single_streaming_chunks(self, xml_chunk: str) -> DeltaMessage: # but didn't generate final empty delta, then complete it if ( self.current_call_id is not None + and self.current_call_id == initial_call_id and self.tool_call_end_token in xml_chunk ): has_toolcall_close = any( @@ -186,9 +192,13 @@ def parse_single_streaming_chunks(self, xml_chunk: str) -> DeltaMessage: # Only execute when still on the same call as when entered, # to prevent accidentally closing new calls # in multi scenarios - if self.current_call_id is not None and ( - self.function_end_token in xml_chunk - or self.tool_call_end_token in xml_chunk + if ( + self.current_call_id is not None + and self.current_call_id == initial_call_id + and ( + self.function_end_token in xml_chunk + or self.tool_call_end_token in xml_chunk + ) ): # Close potentially unclosed element if self.current_param_name: