wehub-resource-sync
1178 lines
41 KiB
Python
1178 lines
41 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
|
"""Tests for the engine-based Qwen3 tool call parser.
|
|
|
|
These validate that the engine-driven parser correctly handles
|
|
Qwen3 XML-style tool calls.
|
|
"""
|
|
|
|
import json
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from tests.parser.engine.conftest import make_mock_tokenizer
|
|
from tests.parser.engine.streaming_helpers import (
|
|
collect_content,
|
|
collect_function_name,
|
|
collect_tool_arguments,
|
|
simulate_tool_streaming,
|
|
)
|
|
from vllm.parser.engine.parser_engine import ParserEngine
|
|
from vllm.parser.qwen3 import (
|
|
TOOL_CALL_END,
|
|
TOOL_CALL_START,
|
|
qwen3_config,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_tokenizer():
|
|
return make_mock_tokenizer(
|
|
{
|
|
TOOL_CALL_START: 100,
|
|
TOOL_CALL_END: 101,
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def parser(mock_tokenizer):
|
|
return ParserEngine(
|
|
mock_tokenizer,
|
|
parser_engine_config=qwen3_config(thinking=False),
|
|
)
|
|
|
|
|
|
class TestNonStreaming:
|
|
def test_no_tool_calls(self, parser, mock_request):
|
|
result = parser.extract_tool_calls(
|
|
"This is a regular response without any tool calls.",
|
|
mock_request,
|
|
)
|
|
assert result.tools_called is False
|
|
assert result.tool_calls == []
|
|
assert result.content == ("This is a regular response without any tool calls.")
|
|
|
|
def test_single_tool_call(self, parser, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=get_weather>\n"
|
|
"<parameter=city>Tokyo</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
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 == {"city": "Tokyo"}
|
|
|
|
def test_parallel_tool_calls(self, parser, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=get_weather>\n"
|
|
"<parameter=city>Tokyo</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
"<tool_call>\n"
|
|
"<function=get_time>\n"
|
|
"<parameter=timezone>Asia/Tokyo</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
assert len(result.tool_calls) == 2
|
|
assert result.tool_calls[0].function.name == "get_weather"
|
|
assert result.tool_calls[1].function.name == "get_time"
|
|
|
|
args0 = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args0 == {"city": "Tokyo"}
|
|
args1 = json.loads(result.tool_calls[1].function.arguments)
|
|
assert args1 == {"timezone": "Asia/Tokyo"}
|
|
|
|
def test_various_data_types(self, parser, mock_request):
|
|
text = (
|
|
"<tool_call>\n<function=test_function>\n"
|
|
"<parameter=string_field>hello</parameter>\n"
|
|
"<parameter=int_field>42</parameter>\n"
|
|
"<parameter=float_field>3.14</parameter>\n"
|
|
"<parameter=bool_field>true</parameter>\n"
|
|
"<parameter=null_field>null</parameter>\n"
|
|
'<parameter=array_field>["a", "b", "c"]</parameter>\n'
|
|
'<parameter=object_field>{"nested": "value"}</parameter>\n'
|
|
"</function>\n</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["string_field"] == "hello"
|
|
assert args["int_field"] == "42"
|
|
assert args["float_field"] == "3.14"
|
|
assert args["bool_field"] == "true"
|
|
assert args["null_field"] == "null"
|
|
assert args["array_field"] == '["a", "b", "c"]'
|
|
assert args["object_field"] == '{"nested": "value"}'
|
|
|
|
def test_empty_arguments(self, parser, mock_request):
|
|
text = "<tool_call>\n<function=refresh>\n</function>\n</tool_call>"
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
assert result.tool_calls[0].function.name == "refresh"
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args == {}
|
|
|
|
def test_surrounding_text(self, parser, mock_request):
|
|
text = (
|
|
"Let me check the weather for you.\n\n"
|
|
"<tool_call>\n<function=get_weather>\n"
|
|
"<parameter=city>Tokyo</parameter>\n"
|
|
"</function>\n</tool_call>\n\n"
|
|
"I will get that information."
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
assert result.content is not None
|
|
assert "Let me check the weather" in result.content
|
|
assert result.tool_calls[0].function.name == "get_weather"
|
|
|
|
def test_escaped_strings(self, parser, mock_request):
|
|
text = (
|
|
"<tool_call>\n<function=test_function>\n"
|
|
'<parameter=quoted>He said "hello"</parameter>\n'
|
|
"<parameter=path>C:\\Users\\file.txt</parameter>\n"
|
|
"<parameter=newline>line1\nline2</parameter>\n"
|
|
"</function>\n</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["quoted"] == 'He said "hello"'
|
|
assert args["path"] == "C:\\Users\\file.txt"
|
|
assert args["newline"] == "line1\nline2"
|
|
|
|
def test_multiple_parameters(self, parser, mock_request):
|
|
text = (
|
|
"<tool_call>\n<function=search>\n"
|
|
"<parameter=query>vllm parsing</parameter>\n"
|
|
"<parameter=limit>10</parameter>\n"
|
|
"<parameter=exact_match>false</parameter>\n"
|
|
"</function>\n</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args == {
|
|
"query": "vllm parsing",
|
|
"limit": "10",
|
|
"exact_match": "false",
|
|
}
|
|
|
|
def test_multiline_param_values(self, parser, mock_request):
|
|
"""Parameter values spanning multiple lines."""
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=Bash>\n"
|
|
"<parameter=command>\n"
|
|
"ls -la /tmp\n"
|
|
"</parameter>\n"
|
|
"<parameter=description>\n"
|
|
"List files in /tmp directory\n"
|
|
"</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
assert len(result.tool_calls) == 1
|
|
assert result.tool_calls[0].function.name == "Bash"
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["command"] == "ls -la /tmp"
|
|
assert args["description"] == "List files in /tmp directory"
|
|
|
|
def test_multiline_two_tool_calls(self, parser, mock_request):
|
|
"""Two tool calls with multi-line parameter values (bug report)."""
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=Bash>\n"
|
|
"<parameter=command>\n"
|
|
"find /workspace -name '*.py' | head -20\n"
|
|
"</parameter>\n"
|
|
"<parameter=description>\n"
|
|
"Find Python files\n"
|
|
"</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
"<tool_call>\n"
|
|
"<function=Read>\n"
|
|
"<parameter=file_path>\n"
|
|
"/workspace/main.py\n"
|
|
"</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
assert len(result.tool_calls) == 2
|
|
assert result.tool_calls[0].function.name == "Bash"
|
|
assert result.tool_calls[1].function.name == "Read"
|
|
args0 = json.loads(result.tool_calls[0].function.arguments)
|
|
assert "find /workspace" in args0["command"]
|
|
assert "Find Python files" in args0["description"]
|
|
args1 = json.loads(result.tool_calls[1].function.arguments)
|
|
assert "/workspace/main.py" in args1["file_path"]
|
|
|
|
def test_consecutive_tool_calls_without_tool_end(self, parser, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=get_weather>\n"
|
|
"<parameter=city>Tokyo</parameter>\n"
|
|
"</function>\n"
|
|
"<tool_call>\n"
|
|
"<function=get_weather>\n"
|
|
"<parameter=city>Paris</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
assert len(result.tool_calls) == 2
|
|
args0 = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args0 == {"city": "Tokyo"}
|
|
args1 = json.loads(result.tool_calls[1].function.arguments)
|
|
assert args1 == {"city": "Paris"}
|
|
|
|
def test_nested_json_array_parameter(self, parser, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=AskUserQuestion>\n"
|
|
"<parameter=questions>"
|
|
'[{"question": "Pick a color",'
|
|
' "multiSelect": false, "answer": null}]'
|
|
"</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
|
|
assert result.tools_called is True
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args == {
|
|
"questions": '[{"question": "Pick a color",'
|
|
' "multiSelect": false, "answer": null}]',
|
|
}
|
|
|
|
|
|
class TestStreaming:
|
|
def test_basic_streaming(self, parser, mock_request):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=get_weather>\n",
|
|
"<parameter=city>Tokyo",
|
|
"</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
name = collect_function_name(results)
|
|
assert name == "get_weather"
|
|
|
|
args_text = collect_tool_arguments(results)
|
|
assert args_text
|
|
parsed = json.loads(args_text)
|
|
assert parsed == {"city": "Tokyo"}
|
|
|
|
def test_streaming_multi_param(self, parser, mock_request):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=get_weather>\n",
|
|
"<parameter=city>Tokyo</parameter>\n",
|
|
"<parameter=unit>celsius</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
name = collect_function_name(results)
|
|
assert name == "get_weather"
|
|
|
|
args_text = collect_tool_arguments(results)
|
|
assert args_text
|
|
parsed = json.loads(args_text)
|
|
assert parsed == {"city": "Tokyo", "unit": "celsius"}
|
|
|
|
def test_streaming_args_arrive_incrementally(self, parser, mock_request):
|
|
"""Arguments must stream as intermediate deltas, not batch at
|
|
tool-end."""
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=get_weather>\n",
|
|
"<parameter=city>Tokyo</parameter>\n",
|
|
"<parameter=unit>celsius</parameter>\n",
|
|
"<parameter=days>5</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
arg_deltas: list[str] = []
|
|
for delta, _ in results:
|
|
if delta and delta.tool_calls:
|
|
for tc in delta.tool_calls:
|
|
if tc.function and tc.function.arguments:
|
|
arg_deltas.append(tc.function.arguments)
|
|
|
|
assert len(arg_deltas) > 1, (
|
|
f"Expected arguments across multiple deltas, got {len(arg_deltas)}: "
|
|
f"{arg_deltas}"
|
|
)
|
|
concatenated = "".join(arg_deltas)
|
|
parsed = json.loads(concatenated)
|
|
assert parsed == {"city": "Tokyo", "unit": "celsius", "days": "5"}
|
|
|
|
def test_streaming_long_string_arg_before_parameter_end(self, parser, mock_request):
|
|
"""Long string arguments should stream before the closing parameter tag."""
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=write_report>\n",
|
|
"<parameter=content>",
|
|
"Artificial intelligence has rapidly transformed the way ",
|
|
"developers build dynamic applications with external tools.",
|
|
"</param",
|
|
"eter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
pre_close_arg_deltas: list[str] = []
|
|
all_arg_deltas: list[str] = []
|
|
for idx, (delta, _) in enumerate(results):
|
|
if delta and delta.tool_calls:
|
|
for tc in delta.tool_calls:
|
|
if tc.function and tc.function.arguments:
|
|
all_arg_deltas.append(tc.function.arguments)
|
|
if idx < 5:
|
|
pre_close_arg_deltas.append(tc.function.arguments)
|
|
|
|
assert len(pre_close_arg_deltas) > 1, (
|
|
"Expected long string arguments to stream incrementally before "
|
|
f"</parameter>, got {pre_close_arg_deltas}"
|
|
)
|
|
partial_args = "".join(pre_close_arg_deltas)
|
|
assert partial_args.startswith('{"content": "Artificial intelligence')
|
|
assert partial_args.endswith("external tools.")
|
|
assert not partial_args.endswith('"}')
|
|
|
|
all_args = "".join(all_arg_deltas)
|
|
assert json.loads(all_args) == {
|
|
"content": (
|
|
"Artificial intelligence has rapidly transformed the way "
|
|
"developers build dynamic applications with external tools."
|
|
)
|
|
}
|
|
|
|
def test_streaming_text_before_tool(self, parser, mock_request):
|
|
chunks = [
|
|
"Let me check ",
|
|
"the weather. ",
|
|
"<tool_call>\n",
|
|
"<function=get_weather>\n",
|
|
"<parameter=city>Tokyo</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
assert collect_content(results).strip().startswith("Let me check")
|
|
|
|
def test_streaming_empty_args(self, parser, mock_request):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=refresh>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
name = collect_function_name(results)
|
|
assert name == "refresh"
|
|
|
|
def test_streaming_split_parameter_tag(self, parser, mock_request):
|
|
"""Parameter tag split across chunks."""
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=test>\n",
|
|
"<parameter=",
|
|
"name>Alice",
|
|
"</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
name = collect_function_name(results)
|
|
assert name == "test"
|
|
|
|
args_text = collect_tool_arguments(results)
|
|
assert args_text
|
|
parsed = json.loads(args_text)
|
|
assert parsed["name"] == "Alice"
|
|
|
|
def test_streaming_split_next_parameter_tag_is_buffered(self, parser, mock_request):
|
|
"""A split opening parameter tag must not leak into previous value."""
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=search>\n",
|
|
"<parameter=query>hello ",
|
|
"<param",
|
|
"eter=limit>10</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
args_after_partial_tag = collect_tool_arguments(results[:4])
|
|
assert "<param" not in args_after_partial_tag
|
|
assert args_after_partial_tag == '{"query": "hello'
|
|
|
|
args_text = collect_tool_arguments(results)
|
|
assert json.loads(args_text) == {"query": "hello", "limit": "10"}
|
|
|
|
def test_streaming_numeric_values(self, parser, mock_request):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=set_config>\n",
|
|
"<parameter=count>42</parameter>\n",
|
|
"<parameter=active>true</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
args_text = collect_tool_arguments(results)
|
|
if args_text:
|
|
parsed = json.loads(args_text)
|
|
assert parsed["count"] == "42"
|
|
assert parsed["active"] == "true"
|
|
|
|
def test_streaming_parallel_calls(self, parser, mock_request):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=get_weather>\n",
|
|
"<parameter=city>Tokyo</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
"<tool_call>\n",
|
|
"<function=get_time>\n",
|
|
"<parameter=tz>JST</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
names = []
|
|
for delta, _ in results:
|
|
if delta and delta.tool_calls:
|
|
for tc in delta.tool_calls:
|
|
if tc.function and tc.function.name:
|
|
names.append(tc.function.name)
|
|
|
|
assert "get_weather" in names
|
|
assert "get_time" in names
|
|
|
|
def test_streaming_value_split_across_chunks(self, parser, mock_request):
|
|
"""Parameter value split across multiple chunks."""
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=search>\n",
|
|
"<parameter=query>hello ",
|
|
"world",
|
|
" test</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
args_text = collect_tool_arguments(results)
|
|
assert args_text
|
|
parsed = json.loads(args_text)
|
|
assert parsed["query"] == "hello world test"
|
|
|
|
def test_streaming_split_tool_call_tag(self, parser, mock_request):
|
|
"""<tool_call> arrives as a single special token; the rest of
|
|
the content is split into fine-grained chunks."""
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=test>\n",
|
|
"<parameter=x>1",
|
|
"</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
name = collect_function_name(results)
|
|
assert name == "test"
|
|
|
|
args_text = collect_tool_arguments(results)
|
|
assert args_text
|
|
parsed = json.loads(args_text)
|
|
assert parsed["x"] == "1"
|
|
|
|
def test_char_by_char_streaming(self, mock_request):
|
|
"""Feed text character-by-character to test lexer robustness.
|
|
|
|
Uses a tokenizer without special token IDs because char-by-char
|
|
delivery only occurs when the tokenizer splits the tag across
|
|
multiple sub-word tokens (i.e., no dedicated special token).
|
|
"""
|
|
tokenizer = MagicMock()
|
|
tokenizer.encode.return_value = [1, 2, 3]
|
|
tokenizer.get_vocab.return_value = {}
|
|
tokenizer.decode.side_effect = lambda ids: "".join(
|
|
chr(i) if i < 128 else f"<{i}>" for i in ids
|
|
)
|
|
no_tid_parser = ParserEngine(
|
|
tokenizer, parser_engine_config=qwen3_config(thinking=False)
|
|
)
|
|
|
|
full_text = (
|
|
"<tool_call>\n"
|
|
"<function=echo>\n"
|
|
"<parameter=msg>hi</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
chunks = list(full_text)
|
|
results = simulate_tool_streaming(no_tid_parser, mock_request, chunks)
|
|
|
|
name = collect_function_name(results)
|
|
assert name == "echo"
|
|
|
|
args_text = collect_tool_arguments(results)
|
|
assert args_text
|
|
parsed = json.loads(args_text)
|
|
assert parsed == {"msg": "hi"}
|
|
|
|
def test_streaming_multiline_param_values(self, parser, mock_request):
|
|
"""Multi-line parameter values in streaming mode."""
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=Bash>\n",
|
|
"<parameter=command>\n",
|
|
"ls -la /tmp\n",
|
|
"</parameter>\n",
|
|
"<parameter=description>\n",
|
|
"List files\n",
|
|
"</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
name = collect_function_name(results)
|
|
assert name == "Bash"
|
|
|
|
args_text = collect_tool_arguments(results)
|
|
assert args_text
|
|
parsed = json.loads(args_text)
|
|
assert "ls -la /tmp" in parsed["command"]
|
|
assert "List files" in parsed["description"]
|
|
|
|
def test_streaming_multiline_two_tool_calls(self, parser, mock_request):
|
|
"""Two tool calls with multi-line values — matches bug report."""
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=Bash>\n",
|
|
"<parameter=command>\n",
|
|
"find /workspace -name '*.py' | head -20\n",
|
|
"</parameter>\n",
|
|
"<parameter=description>\n",
|
|
"Find Python files\n",
|
|
"</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
"<tool_call>\n",
|
|
"<function=Read>\n",
|
|
"<parameter=file_path>\n",
|
|
"/workspace/main.py\n",
|
|
"</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
results = simulate_tool_streaming(parser, mock_request, chunks)
|
|
|
|
names = []
|
|
for delta, _ in results:
|
|
if delta and delta.tool_calls:
|
|
for tc in delta.tool_calls:
|
|
if tc.function and tc.function.name:
|
|
names.append(tc.function.name)
|
|
|
|
assert "Bash" in names
|
|
assert "Read" in names
|
|
|
|
|
|
class TestArgConverter:
|
|
"""Direct tests for the Qwen3 arg_converter with multi-line values."""
|
|
|
|
def test_multiline_param_values(self):
|
|
from vllm.parser.qwen3 import (
|
|
_qwen3_arg_converter,
|
|
)
|
|
|
|
raw = (
|
|
"<parameter=command>\n"
|
|
"ls -la /tmp\n"
|
|
"</parameter>\n"
|
|
"<parameter=description>\n"
|
|
"List files\n"
|
|
"</parameter>\n"
|
|
)
|
|
result = json.loads(_qwen3_arg_converter(raw, partial=False))
|
|
assert result["command"] == "ls -la /tmp"
|
|
assert result["description"] == "List files"
|
|
|
|
def test_two_multiline_params(self):
|
|
from vllm.parser.qwen3 import (
|
|
_qwen3_arg_converter,
|
|
)
|
|
|
|
raw = (
|
|
"<parameter=a>\nfoo\nbar\n</parameter>\n"
|
|
"<parameter=b>\nbaz\nqux\n</parameter>\n"
|
|
)
|
|
result = json.loads(_qwen3_arg_converter(raw, partial=False))
|
|
assert result["a"] == "foo\nbar"
|
|
assert result["b"] == "baz\nqux"
|
|
|
|
def test_partial_multiline(self):
|
|
from vllm.parser.qwen3 import (
|
|
_qwen3_arg_converter,
|
|
)
|
|
|
|
raw = "<parameter=command>\nls -la</parameter>\n<parameter=desc>\npartial value"
|
|
result = json.loads(_qwen3_arg_converter(raw, partial=True))
|
|
assert result["command"] == "ls -la"
|
|
assert result["desc"] == "partial value"
|
|
|
|
def test_partial_value_with_angle_bracket(self):
|
|
from vllm.parser.qwen3 import (
|
|
_qwen3_arg_converter,
|
|
)
|
|
|
|
raw = "<parameter=expr>x<5"
|
|
result = json.loads(_qwen3_arg_converter(raw, partial=True))
|
|
assert result == {"expr": "x<5"}
|
|
|
|
def test_partial_value_with_angle_bracket_and_complete_param(self):
|
|
from vllm.parser.qwen3 import (
|
|
_qwen3_arg_converter,
|
|
)
|
|
|
|
raw = "<parameter=city>Tokyo</parameter>\n<parameter=expr>x<5"
|
|
result = json.loads(_qwen3_arg_converter(raw, partial=True))
|
|
assert result == {"city": "Tokyo", "expr": "x<5"}
|
|
|
|
|
|
class TestSchemaAwareTypeCoercion:
|
|
"""Verify that _fix_arg_types corrects miscoerced values using the
|
|
tool schema."""
|
|
|
|
@pytest.fixture
|
|
def tools(self):
|
|
from vllm.entrypoints.openai.chat_completion.protocol import (
|
|
ChatCompletionToolsParam,
|
|
)
|
|
|
|
return [
|
|
ChatCompletionToolsParam(
|
|
type="function",
|
|
function={
|
|
"name": "TaskUpdate",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"taskId": {"type": "string"},
|
|
"count": {"type": "integer"},
|
|
"ratio": {"type": "number"},
|
|
"flag": {"type": "string"},
|
|
},
|
|
},
|
|
},
|
|
)
|
|
]
|
|
|
|
@pytest.fixture
|
|
def parser_with_tools(self, mock_tokenizer, tools):
|
|
return ParserEngine(
|
|
mock_tokenizer,
|
|
tools=tools,
|
|
parser_engine_config=qwen3_config(thinking=False),
|
|
)
|
|
|
|
def test_string_param_not_coerced_to_int(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=TaskUpdate>\n"
|
|
"<parameter=taskId>1</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
assert result.tools_called
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["taskId"] == "1"
|
|
assert isinstance(args["taskId"], str)
|
|
|
|
def test_string_param_not_coerced_to_bool(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=TaskUpdate>\n"
|
|
"<parameter=flag>true</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["flag"] == "true"
|
|
assert isinstance(args["flag"], str)
|
|
|
|
def test_int_param_still_coerced(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=TaskUpdate>\n"
|
|
"<parameter=count>42</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["count"] == 42
|
|
assert isinstance(args["count"], int)
|
|
|
|
def test_no_tools_keeps_strings(self, parser, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=TaskUpdate>\n"
|
|
"<parameter=taskId>1</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["taskId"] == "1"
|
|
assert isinstance(args["taskId"], str)
|
|
|
|
def test_streaming_string_param_not_coerced(self, parser_with_tools, mock_request):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=TaskUpdate>\n",
|
|
"<parameter=taskId>1</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
results = simulate_tool_streaming(parser_with_tools, mock_request, chunks)
|
|
args_str = collect_tool_arguments(results)
|
|
args = json.loads(args_str)
|
|
assert args["taskId"] == "1"
|
|
assert isinstance(args["taskId"], str)
|
|
|
|
|
|
class TestAnyOfTypeCoercion:
|
|
"""Verify that _fix_arg_types handles union types (anyOf/oneOf)."""
|
|
|
|
@pytest.fixture
|
|
def tools_with_anyof(self):
|
|
from vllm.entrypoints.openai.chat_completion.protocol import (
|
|
ChatCompletionToolsParam,
|
|
)
|
|
|
|
return [
|
|
ChatCompletionToolsParam(
|
|
type="function",
|
|
function={
|
|
"name": "set_config",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"port": {
|
|
"anyOf": [
|
|
{"type": "string"},
|
|
{"type": "null"},
|
|
],
|
|
},
|
|
"count": {"type": "integer"},
|
|
},
|
|
},
|
|
},
|
|
)
|
|
]
|
|
|
|
@pytest.fixture
|
|
def parser_with_anyof(self, mock_tokenizer, tools_with_anyof):
|
|
return ParserEngine(
|
|
mock_tokenizer,
|
|
tools=tools_with_anyof,
|
|
parser_engine_config=qwen3_config(thinking=False),
|
|
)
|
|
|
|
def test_anyof_string_param_not_coerced(self, parser_with_anyof, mock_request):
|
|
"""A param with anyOf including 'string' must not be coerced
|
|
to integer."""
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=set_config>\n"
|
|
"<parameter=port>8080</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_anyof.extract_tool_calls(text, mock_request)
|
|
assert result.tools_called
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["port"] == "8080"
|
|
|
|
|
|
class TestSchemaCoercionBoolNumberNull:
|
|
"""Verify that _fix_arg_types coerces string values to non-string
|
|
schema types using coerce_to_schema_type."""
|
|
|
|
@pytest.fixture
|
|
def tools(self):
|
|
from vllm.entrypoints.openai.chat_completion.protocol import (
|
|
ChatCompletionToolsParam,
|
|
)
|
|
|
|
return [
|
|
ChatCompletionToolsParam(
|
|
type="function",
|
|
function={
|
|
"name": "configure",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"enabled": {"type": "boolean"},
|
|
"ratio": {"type": "number"},
|
|
"count": {"type": "integer"},
|
|
"value": {"type": ["integer", "null"]},
|
|
"label": {"type": "string"},
|
|
},
|
|
},
|
|
},
|
|
)
|
|
]
|
|
|
|
@pytest.fixture
|
|
def parser_with_tools(self, mock_tokenizer, tools):
|
|
return ParserEngine(
|
|
mock_tokenizer,
|
|
tools=tools,
|
|
parser_engine_config=qwen3_config(thinking=False),
|
|
)
|
|
|
|
def test_bool_param_coerced(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=configure>\n"
|
|
"<parameter=enabled>true</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["enabled"] is True
|
|
assert isinstance(args["enabled"], bool)
|
|
|
|
def test_number_param_whole_normalized(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=configure>\n"
|
|
"<parameter=ratio>5.0</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["ratio"] == 5
|
|
assert isinstance(args["ratio"], int)
|
|
|
|
def test_number_param_fractional(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=configure>\n"
|
|
"<parameter=ratio>3.14</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["ratio"] == pytest.approx(3.14)
|
|
assert isinstance(args["ratio"], float)
|
|
|
|
def test_null_coerced_when_in_schema(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=configure>\n"
|
|
"<parameter=value>null</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["value"] is None
|
|
|
|
def test_null_stays_string_without_null_schema(
|
|
self, parser_with_tools, mock_request
|
|
):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=configure>\n"
|
|
"<parameter=label>null</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["label"] == "null"
|
|
assert isinstance(args["label"], str)
|
|
|
|
def test_streaming_bool_param_coerced(self, parser_with_tools, mock_request):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=configure>\n",
|
|
"<parameter=enabled>true</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
results = simulate_tool_streaming(parser_with_tools, mock_request, chunks)
|
|
args_str = collect_tool_arguments(results)
|
|
args = json.loads(args_str)
|
|
assert args["enabled"] is True
|
|
assert isinstance(args["enabled"], bool)
|
|
|
|
def test_streaming_number_param_coerced(self, parser_with_tools, mock_request):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=configure>\n",
|
|
"<parameter=ratio>3.14</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
results = simulate_tool_streaming(parser_with_tools, mock_request, chunks)
|
|
args_str = collect_tool_arguments(results)
|
|
args = json.loads(args_str)
|
|
assert args["ratio"] == pytest.approx(3.14)
|
|
assert isinstance(args["ratio"], float)
|
|
|
|
def test_streaming_matches_non_streaming_comprehensive(
|
|
self, parser_with_tools, mock_request
|
|
):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=configure>\n"
|
|
"<parameter=enabled>true</parameter>\n"
|
|
"<parameter=ratio>5.0</parameter>\n"
|
|
"<parameter=count>42</parameter>\n"
|
|
"<parameter=value>null</parameter>\n"
|
|
"<parameter=label>hello</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
non_stream = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
ns_args = json.loads(non_stream.tool_calls[0].function.arguments)
|
|
|
|
chunks = [line + "\n" for line in text.split("\n") if line]
|
|
results = simulate_tool_streaming(parser_with_tools, mock_request, chunks)
|
|
s_args = json.loads(collect_tool_arguments(results))
|
|
|
|
assert s_args == ns_args
|
|
assert ns_args == {
|
|
"enabled": True,
|
|
"ratio": 5,
|
|
"count": 42,
|
|
"value": None,
|
|
"label": "hello",
|
|
}
|
|
|
|
|
|
class TestNestedSchemaCoercion:
|
|
"""Verify that _fix_arg_types recurses into nested objects and arrays."""
|
|
|
|
@pytest.fixture
|
|
def tools(self):
|
|
from vllm.entrypoints.openai.chat_completion.protocol import (
|
|
ChatCompletionToolsParam,
|
|
)
|
|
|
|
return [
|
|
ChatCompletionToolsParam(
|
|
type="function",
|
|
function={
|
|
"name": "search",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"filters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"language": {"type": "string"},
|
|
"min_stars": {"type": "integer"},
|
|
},
|
|
},
|
|
"tags": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
},
|
|
"limits": {
|
|
"type": "array",
|
|
"items": {"type": "integer"},
|
|
},
|
|
"verbose": {"type": "boolean"},
|
|
},
|
|
},
|
|
},
|
|
),
|
|
ChatCompletionToolsParam(
|
|
type="function",
|
|
function={
|
|
"name": "AskUserQuestion",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"questions": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"question": {"type": "string"},
|
|
"multiSelect": {
|
|
"type": "boolean",
|
|
},
|
|
"answer": {
|
|
"type": ["string", "null"],
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
),
|
|
]
|
|
|
|
@pytest.fixture
|
|
def parser_with_tools(self, mock_tokenizer, tools):
|
|
return ParserEngine(
|
|
mock_tokenizer,
|
|
tools=tools,
|
|
parser_engine_config=qwen3_config(thinking=False),
|
|
)
|
|
|
|
def test_nested_object_coerced(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=search>\n"
|
|
'<parameter=filters>{"language": "python",'
|
|
' "min_stars": 100}</parameter>\n'
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["filters"] == {"language": "python", "min_stars": 100}
|
|
assert isinstance(args["filters"]["min_stars"], int)
|
|
|
|
def test_nested_array_items_coerced(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=search>\n"
|
|
"<parameter=limits>[10, 20, 30]</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["limits"] == [10, 20, 30]
|
|
assert all(isinstance(v, int) for v in args["limits"])
|
|
|
|
def test_nested_string_array_not_coerced(self, parser_with_tools, mock_request):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=search>\n"
|
|
'<parameter=tags>["ml", "42"]</parameter>\n'
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
assert args["tags"] == ["ml", "42"]
|
|
assert all(isinstance(v, str) for v in args["tags"])
|
|
|
|
def test_array_of_objects_with_bool_and_null_coerced(
|
|
self, parser_with_tools, mock_request
|
|
):
|
|
text = (
|
|
"<tool_call>\n"
|
|
"<function=AskUserQuestion>\n"
|
|
"<parameter=questions>"
|
|
'[{"question": "Pick a color",'
|
|
' "multiSelect": false, "answer": null}]'
|
|
"</parameter>\n"
|
|
"</function>\n"
|
|
"</tool_call>"
|
|
)
|
|
result = parser_with_tools.extract_tool_calls(text, mock_request)
|
|
args = json.loads(result.tool_calls[0].function.arguments)
|
|
questions = args["questions"]
|
|
assert isinstance(questions, list)
|
|
assert len(questions) == 1
|
|
assert questions[0]["question"] == "Pick a color"
|
|
assert questions[0]["multiSelect"] is False
|
|
assert questions[0]["answer"] is None
|
|
|
|
def test_streaming_array_of_objects_with_bool_and_null_coerced(
|
|
self, parser_with_tools, mock_request
|
|
):
|
|
chunks = [
|
|
"<tool_call>\n",
|
|
"<function=AskUserQuestion>\n",
|
|
'<parameter=questions>[{"question": "Pick a color",',
|
|
' "multiSelect": false, "answer": null}]',
|
|
"</parameter>\n",
|
|
"</function>\n",
|
|
"</tool_call>",
|
|
]
|
|
results = simulate_tool_streaming(parser_with_tools, mock_request, chunks)
|
|
args_str = collect_tool_arguments(results)
|
|
args = json.loads(args_str)
|
|
questions = args["questions"]
|
|
assert isinstance(questions, list)
|
|
assert len(questions) == 1
|
|
assert questions[0]["question"] == "Pick a color"
|
|
assert questions[0]["multiSelect"] is False
|
|
assert questions[0]["answer"] is None
|