@tduncan / Nvidia-Optimization-Stack-VLLM Mirror · read-only / commits / be12afd284
Fork 0

A high-throughput and memory-efficient inference and serving engine for LLMs

Mirrored from https://github.com/tduncan-lum/Nvidia-Optimization-Stack---VLLM.git

Synced Nov 6, 09:30 UTC

[Bugfix] Fix Deepseekv32 tool parser when stream interval > 1 (#36056)

Flora Feng committed Mar 19, 2026 at 19:51 UTC be12afd284f3f09991a7fcf506553375dc58be36
2 files changed +622 -437
tests/tool_parsers/test_deepseekv32_tool_parser.py new
+476
@@ -0,0 +1,476 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +
4 +"""Unit tests for DeepSeekV32ToolParser.
5 +
6 +These tests use a minimal mock tokenizer so no real model weights are required.
7 +"""
8 +
9 +import json
10 +from unittest.mock import MagicMock
11 +
12 +import pytest
13 +
14 +from vllm.tool_parsers.deepseekv32_tool_parser import DeepSeekV32ToolParser
15 +
16 +# ---------------------------------------------------------------------------
17 +# Helpers
18 +# ---------------------------------------------------------------------------
19 +
20 +# Token IDs are not used by the V32 parser logic, so we only need the
21 +# tokenizer object to be truthy (the parser checks `if not self.model_tokenizer`).
22 +MOCK_TOKENIZER = MagicMock()
23 +MOCK_TOKENIZER.get_vocab.return_value = {}
24 +
25 +
26 +def make_parser() -> DeepSeekV32ToolParser:
27 + return DeepSeekV32ToolParser(MOCK_TOKENIZER)
28 +
29 +
30 +def make_tool_param(name: str, params: dict) -> MagicMock:
31 + """Build a mock tool matching the ChatCompletionToolsParam shape."""
32 + tool = MagicMock()
33 + tool.function.name = name
34 + tool.function.parameters = params
35 + return tool
36 +
37 +
38 +def make_request(tools=None) -> MagicMock:
39 + req = MagicMock()
40 + req.tools = tools
41 + return req
42 +
43 +
44 +# Shorthand for the DSML tokens used throughout
45 +FC_START = "<｜DSML｜function_calls>"
46 +FC_END = "</｜DSML｜function_calls>"
47 +INV_START = '<｜DSML｜invoke name="'
48 +INV_END = "</｜DSML｜invoke>"
49 +PARAM_START = '<｜DSML｜parameter name="'
50 +PARAM_END = "</｜DSML｜parameter>"
51 +
52 +
53 +def build_tool_call(func_name: str, params: dict[str, str]) -> str:
54 + """Build a complete model-output tool call string."""
55 + param_strs = "".join(
56 + f'{PARAM_START}{k}" string="true">{v}{PARAM_END}' for k, v in params.items()
57 + )
58 + return f'{FC_START}\n{INV_START}{func_name}">\n{param_strs}\n{INV_END}\n{FC_END}'
59 +
60 +
61 +# ---------------------------------------------------------------------------
62 +# Tests: DeepSeekV32ToolParser._convert_param_value
63 +# ---------------------------------------------------------------------------
64 +
65 +
66 +class TestConvertParamValue:
67 + @pytest.fixture
68 + def parser(self):
69 + return make_parser()
70 +
71 + def test_null(self, parser):
72 + assert parser._convert_param_value("null", "string") is None
73 + assert parser._convert_param_value("NULL", "integer") is None
74 +
75 + def test_string(self, parser):
76 + assert parser._convert_param_value("hello", "string") == "hello"
77 +
78 + def test_integer_valid(self, parser):
79 + assert parser._convert_param_value("42", "integer") == 42
80 +
81 + def test_integer_invalid_falls_back_to_str(self, parser):
82 + assert parser._convert_param_value("abc", "int") == "abc"
83 +
84 + def test_number_float(self, parser):
85 + assert parser._convert_param_value("3.14", "number") == pytest.approx(3.14)
86 +
87 + def test_number_whole_returns_int(self, parser):
88 + assert parser._convert_param_value("5.0", "number") == 5
89 + assert isinstance(parser._convert_param_value("5.0", "number"), int)
90 +
91 + def test_boolean_true(self, parser):
92 + assert parser._convert_param_value("true", "boolean") is True
93 + assert parser._convert_param_value("1", "bool") is True
94 +
95 + def test_boolean_false(self, parser):
96 + assert parser._convert_param_value("false", "boolean") is False
97 + assert parser._convert_param_value("False", "bool") is False
98 +
99 + def test_object_valid_json(self, parser):
100 + assert parser._convert_param_value('{"k": 1}', "object") == {"k": 1}
101 +
102 + def test_object_invalid_json_falls_back(self, parser):
103 + assert parser._convert_param_value("not-json", "object") == "not-json"
104 +
105 + def test_array_valid_json(self, parser):
106 + assert parser._convert_param_value("[1, 2]", "array") == [1, 2]
107 +
108 + def test_unknown_type_tries_json_then_string(self, parser):
109 + assert parser._convert_param_value("123", "unknown") == 123
110 + assert parser._convert_param_value("hello", "unknown") == "hello"
111 +
112 +
113 +# ---------------------------------------------------------------------------
114 +# Tests: extract_tool_calls (non-streaming)
115 +# ---------------------------------------------------------------------------
116 +
117 +
118 +class TestExtractToolCalls:
119 + @pytest.fixture
120 + def parser(self):
121 + return make_parser()
122 +
123 + def test_no_tool_call(self, parser):
124 + result = parser.extract_tool_calls("just some text", None)
125 + assert not result.tools_called
126 + assert result.tool_calls == []
127 + assert result.content == "just some text"
128 +
129 + def test_single_tool_no_params(self, parser):
130 + model_output = f'{FC_START}\n{INV_START}get_time">\n{INV_END}\n{FC_END}'
131 + result = parser.extract_tool_calls(model_output, None)
132 + assert result.tools_called
133 + assert len(result.tool_calls) == 1
134 + assert result.tool_calls[0].function.name == "get_time"
135 + assert json.loads(result.tool_calls[0].function.arguments) == {}
136 +
137 + def test_single_tool_with_params(self, parser):
138 + model_output = build_tool_call(
139 + "get_weather", {"location": "SF", "date": "2024-01-16"}
140 + )
141 + result = parser.extract_tool_calls(model_output, None)
142 + assert result.tools_called
143 + assert len(result.tool_calls) == 1
144 + tc = result.tool_calls[0]
145 + assert tc.function.name == "get_weather"
146 + assert json.loads(tc.function.arguments) == {
147 + "location": "SF",
148 + "date": "2024-01-16",
149 + }
150 +
151 + def test_content_before_tool_call(self, parser):
152 + model_output = "Sure, let me check! " + build_tool_call(
153 + "get_weather", {"location": "NYC"}
154 + )
155 + result = parser.extract_tool_calls(model_output, None)
156 + assert result.tools_called
157 + assert result.content == "Sure, let me check! "
158 +
159 + def test_no_content_prefix_returns_none(self, parser):
160 + model_output = build_tool_call("get_weather", {"location": "NYC"})
161 + result = parser.extract_tool_calls(model_output, None)
162 + assert result.tools_called
163 + assert result.content is None
164 +
165 + def test_multiple_tools(self, parser):
166 + model_output = (
167 + f"{FC_START}\n"
168 + f'{INV_START}get_weather">\n'
169 + f'{PARAM_START}location" string="true">SF{PARAM_END}\n'
170 + f"{INV_END}\n"
171 + f'{INV_START}get_weather">\n'
172 + f'{PARAM_START}location" string="true">NYC{PARAM_END}\n'
173 + f"{INV_END}\n"
174 + f"{FC_END}"
175 + )
176 + result = parser.extract_tool_calls(model_output, None)
177 + assert result.tools_called
178 + assert len(result.tool_calls) == 2
179 + assert json.loads(result.tool_calls[0].function.arguments) == {"location": "SF"}
180 + assert json.loads(result.tool_calls[1].function.arguments) == {
181 + "location": "NYC"
182 + }
183 +
184 +
185 +# ---------------------------------------------------------------------------
186 +# Tests: extract_tool_calls_streaming
187 +# ---------------------------------------------------------------------------
188 +
189 +
190 +class TestExtractToolCallsStreaming:
191 + """Simulate character-by-character streaming and verify reconstructed args."""
192 +
193 + @pytest.fixture
194 + def parser(self):
195 + return make_parser()
196 +
197 + def _stream(self, parser, full_text: str, request=None):
198 + """Drive the parser line-by-line and collect non-None deltas.
199 +
200 + Real tokenizers emit multi-character chunks, not individual characters.
201 + Streaming character-by-character would never deliver the full sentinel
202 + token (e.g. '｜DSML｜') in a single delta, so we split on newlines to
203 + ensure each sentinel always lands in one chunk.
204 + """
205 + if request is None:
206 + request = make_request()
207 + # Split into lines, preserving the trailing newline in each chunk.
208 + chunks: list[str] = []
209 + remaining = full_text
210 + while remaining:
211 + nl = remaining.find("\n")
212 + if nl == -1:
213 + chunks.append(remaining)
214 + break
215 + chunks.append(remaining[: nl + 1])
216 + remaining = remaining[nl + 1 :]
217 +
218 + deltas = []
219 + prev = ""
220 + for chunk in chunks:
221 + curr = prev + chunk
222 + result = parser.extract_tool_calls_streaming(
223 + previous_text=prev,
224 + current_text=curr,
225 + delta_text=chunk,
226 + previous_token_ids=[],
227 + current_token_ids=[],
228 + delta_token_ids=[1],
229 + request=request,
230 + )
231 + prev = curr
232 + if result is not None:
233 + deltas.append(result)
234 + return deltas
235 +
236 + def _reconstruct_args(self, deltas, tool_index=0) -> str:
237 + """Concatenate all argument fragments for a given tool index."""
238 + fragments = []
239 + for d in deltas:
240 + if d.tool_calls:
241 + for tc in d.tool_calls:
242 + if tc.index == tool_index and tc.function and tc.function.arguments:
243 + fragments.append(tc.function.arguments)
244 + return "".join(fragments)
245 +
246 + def test_plain_content_no_tool(self, parser):
247 + full_text = "Hello, world!"
248 + deltas = self._stream(parser, full_text)
249 + content = "".join(d.content for d in deltas if d.content is not None)
250 + assert "Hello, world!" in content
251 + assert all(not d.tool_calls for d in deltas)
252 +
253 + def test_single_tool_streaming(self, parser):
254 + full_text = build_tool_call("get_weather", {"location": "SF"})
255 + deltas = self._stream(parser, full_text)
256 + args_str = self._reconstruct_args(deltas)
257 + assert json.loads(args_str) == {"location": "SF"}
258 +
259 + def test_tool_name_emitted(self, parser):
260 + full_text = build_tool_call("my_func", {"x": "1"})
261 + deltas = self._stream(parser, full_text)
262 + func_names = [
263 + tc.function.name
264 + for d in deltas
265 + if d.tool_calls
266 + for tc in d.tool_calls
267 + if tc.function and tc.function.name
268 + ]
269 + assert any("my_func" in n for n in func_names)
270 +
271 + def test_content_before_tool_call_streaming(self, parser):
272 + full_text = "Thinking... " + build_tool_call("fn", {"a": "b"})
273 + deltas = self._stream(parser, full_text)
274 + content = "".join(d.content for d in deltas if d.content is not None)
275 + assert "Thinking" in content
276 +
277 + def test_type_conversion_in_streaming(self, parser):
278 + tool = make_tool_param(
279 + "add",
280 + {
281 + "type": "object",
282 + "properties": {
283 + "x": {"type": "integer"},
284 + "y": {"type": "integer"},
285 + },
286 + },
287 + )
288 + request = make_request(tools=[tool])
289 + full_text = build_tool_call("add", {"x": "3", "y": "4"})
290 + deltas = self._stream(parser, full_text, request=request)
291 + args_str = self._reconstruct_args(deltas)
292 + assert json.loads(args_str) == {"x": 3, "y": 4}
293 +
294 + def test_multiple_tools_streaming(self, parser):
295 + full_text = (
296 + f"{FC_START}\n"
297 + f'{INV_START}func_a">\n'
298 + f'{PARAM_START}p" string="true">v1{PARAM_END}\n'
299 + f"{INV_END}\n"
300 + f'{INV_START}func_b">\n'
301 + f'{PARAM_START}q" string="true">v2{PARAM_END}\n'
302 + f"{INV_END}\n"
303 + f"{FC_END}"
304 + )
305 + deltas = self._stream(parser, full_text)
306 +
307 + # Collect function names by index
308 + names_by_index: dict[int, str] = {}
309 + for d in deltas:
310 + if d.tool_calls:
311 + for tc in d.tool_calls:
312 + if tc.function and tc.function.name:
313 + names_by_index[tc.index] = tc.function.name
314 +
315 + assert names_by_index.get(0) == "func_a"
316 + assert names_by_index.get(1) == "func_b"
317 +
318 + assert json.loads(self._reconstruct_args(deltas, tool_index=0)) == {"p": "v1"}
319 + assert json.loads(self._reconstruct_args(deltas, tool_index=1)) == {"q": "v2"}
320 +
321 + def test_state_reset_on_new_stream(self, parser):
322 + """A second stream (previous_text == '') must reset state cleanly."""
323 + full_text = build_tool_call("fn", {"k": "v"})
324 + # First stream
325 + self._stream(parser, full_text)
326 + # Second stream - should produce identical results
327 + deltas2 = self._stream(parser, full_text)
328 + assert json.loads(self._reconstruct_args(deltas2)) == {"k": "v"}
329 +
330 + def test_empty_arguments_streaming(self, parser):
331 + """Invoke block with zero parameters should produce empty JSON."""
332 + full_text = f'{FC_START}\n{INV_START}get_time">\n{INV_END}\n{FC_END}'
333 + deltas = self._stream(parser, full_text)
334 + args_str = self._reconstruct_args(deltas)
335 + assert json.loads(args_str) == {}
336 +
337 + def test_unique_tool_call_ids(self, parser):
338 + """Each tool call in a parallel stream must get a distinct id."""
339 + full_text = (
340 + f"{FC_START}\n"
341 + f'{INV_START}fn_a">\n'
342 + f'{PARAM_START}x" string="true">1{PARAM_END}\n'
343 + f"{INV_END}\n"
344 + f'{INV_START}fn_b">\n'
345 + f'{PARAM_START}y" string="true">2{PARAM_END}\n'
346 + f"{INV_END}\n"
347 + f"{FC_END}"
348 + )
349 + deltas = self._stream(parser, full_text)
350 + ids = [
351 + tc.id
352 + for d in deltas
353 + if d.tool_calls
354 + for tc in d.tool_calls
355 + if tc.id is not None
356 + ]
357 + assert len(ids) == 2
358 + assert ids[0] != ids[1]
359 +
360 + def test_eos_after_tool_calls(self, parser):
361 + """EOS token (empty delta_text, non-empty delta_token_ids) returns
362 + a non-None DeltaMessage so the serving framework can finalize."""
363 + full_text = build_tool_call("fn", {"k": "v"})
364 + # Drive through the full text first
365 + deltas = self._stream(parser, full_text)
366 + assert any(d.tool_calls for d in deltas)
367 + # Now simulate EOS: empty delta_text, but token ids present
368 + prev = full_text
369 + result = parser.extract_tool_calls_streaming(
370 + previous_text=prev,
371 + current_text=prev,
372 + delta_text="",
373 + previous_token_ids=[],
374 + current_token_ids=[],
375 + delta_token_ids=[2], # EOS token id
376 + request=make_request(),
377 + )
378 + assert result is not None
379 +
380 + def test_streaming_matches_non_streaming(self, parser):
381 + """Streaming and non-streaming must produce the same result."""
382 + full_text = build_tool_call(
383 + "get_weather", {"location": "SF", "date": "2024-01-16"}
384 + )
385 + # Non-streaming
386 + non_stream = parser.extract_tool_calls(full_text, None)
387 + assert non_stream.tools_called
388 + ns_name = non_stream.tool_calls[0].function.name
389 + ns_args = json.loads(non_stream.tool_calls[0].function.arguments)
390 + # Streaming
391 + deltas = self._stream(parser, full_text)
392 + s_names = [
393 + tc.function.name
394 + for d in deltas
395 + if d.tool_calls
396 + for tc in d.tool_calls
397 + if tc.function and tc.function.name
398 + ]
399 + s_args = json.loads(self._reconstruct_args(deltas))
400 + assert s_names[0] == ns_name
401 + assert s_args == ns_args
402 +
403 + def _stream_chunked(self, parser, full_text: str, chunk_size: int, request=None):
404 + """Drive the parser with fixed-size chunks (simulates stream interval).
405 +
406 + Unlike ``_stream`` which splits on newlines, this splits the text
407 + into ``chunk_size``-character pieces so the start token can be
408 + split across chunks — exactly what happens with stream interval > 1.
409 + """
410 + if request is None:
411 + request = make_request()
412 + chunks = [
413 + full_text[i : i + chunk_size] for i in range(0, len(full_text), chunk_size)
414 + ]
415 + deltas = []
416 + prev = ""
417 + for chunk in chunks:
418 + curr = prev + chunk
419 + result = parser.extract_tool_calls_streaming(
420 + previous_text=prev,
421 + current_text=curr,
422 + delta_text=chunk,
423 + previous_token_ids=[],
424 + current_token_ids=[],
425 + delta_token_ids=[1],
426 + request=request,
427 + )
428 + prev = curr
429 + if result is not None:
430 + deltas.append(result)
431 + return deltas
432 +
433 + def test_single_tool_chunked_stream_interval(self, parser):
434 + """Start token split across chunks (stream interval > 1)."""
435 + full_text = build_tool_call("get_weather", {"location": "SF"})
436 + # Use a chunk size that splits the start token
437 + deltas = self._stream_chunked(parser, full_text, chunk_size=5)
438 + args_str = self._reconstruct_args(deltas)
439 + assert json.loads(args_str) == {"location": "SF"}
440 +
441 + def test_content_before_tool_chunked(self, parser):
442 + """Content before tool call with chunked streaming."""
443 + full_text = "Thinking... " + build_tool_call("fn", {"a": "b"})
444 + deltas = self._stream_chunked(parser, full_text, chunk_size=7)
445 + content = "".join(d.content for d in deltas if d.content is not None)
446 + assert "Thinking" in content
447 + args_str = self._reconstruct_args(deltas)
448 + assert json.loads(args_str) == {"a": "b"}
449 +
450 + def test_multiple_tools_chunked(self, parser):
451 + """Multiple tools with chunked streaming."""
452 + full_text = (
453 + f"{FC_START}\n"
454 + f'{INV_START}func_a">\n'
455 + f'{PARAM_START}p" string="true">v1{PARAM_END}\n'
456 + f"{INV_END}\n"
457 + f'{INV_START}func_b">\n'
458 + f'{PARAM_START}q" string="true">v2{PARAM_END}\n'
459 + f"{INV_END}\n"
460 + f"{FC_END}"
461 + )
462 + deltas = self._stream_chunked(parser, full_text, chunk_size=10)
463 + assert json.loads(self._reconstruct_args(deltas, tool_index=0)) == {"p": "v1"}
464 + assert json.loads(self._reconstruct_args(deltas, tool_index=1)) == {"q": "v2"}
465 +
466 + def test_no_emission_while_incomplete(self, parser):
467 + """No tool calls should be emitted until an invoke block completes."""
468 + # Stream only a partial invoke (no closing tag)
469 + partial_text = (
470 + f"{FC_START}\n"
471 + f'{INV_START}fn">\n'
472 + f'{PARAM_START}k" string="true">val{PARAM_END}\n'
473 + )
474 + deltas = self._stream(parser, partial_text)
475 + # Should have no tool call deltas yet
476 + assert all(not d.tool_calls for d in deltas)
vllm/tool_parsers/deepseekv32_tool_parser.py
+146 -437
@@ -48,41 +48,12 @@ class DeepSeekV32ToolParser(ToolParser):
48
49 self.prev_tool_call_arr: list[dict] = []
50
51 - # Sentinel tokens
52 - self.dsml_token: str = "｜DSML｜"
53 - self.dsml_start_check: str = "<" + self.dsml_token
51 + # Sentinel token
52 self.tool_call_start_token: str = "<｜DSML｜function_calls>"
55 - self.tool_call_end_token: str = "</｜DSML｜function_calls>"
56 - self.invoke_start_prefix: str = "<｜DSML｜invoke name="
57 - self.invoke_end_token: str = "</｜DSML｜invoke>"
58 - self.parameter_prefix: str = "<｜DSML｜parameter name="
59 - self.parameter_end_token: str = "</｜DSML｜parameter>"
60 -
61 - # Streaming state variables
62 - self.current_tool_name_sent: bool = False
63 - # Override base class type - we use string IDs for tool calls
64 - self.current_tool_id: str | None = None # type: ignore
65 - self.streamed_args_for_tool: list[str] = []
66 - self.is_tool_call_started: bool = False
67 - self.failed_count: int = 0
53
69 - # Initialize streaming state variables
54 + # Streaming state
55 + self.is_tool_call_started: bool = False
56 self.current_tool_index: int = 0
71 - self.invoke_index: int = 0
72 - self.header_sent: bool = False
73 - self.current_function_name: str | None = None
74 - self.current_param_name: str | None = None
75 - self.current_param_value: str = ""
76 - self.param_count: int = 0
77 - self.in_param: bool = False
78 - self.in_function: bool = False
79 - self.json_started: bool = False
80 - self.json_closed: bool = False
81 - self.accumulated_params: dict = {}
82 - self.streaming_request: ChatCompletionRequest | None = None
83 -
84 - # Enhanced streaming state - reset for each new message
85 - self._reset_streaming_state()
57
58 # Regex patterns for complete parsing
59 self.tool_call_complete_regex = re.compile(
@@ -106,10 +77,6 @@ class DeepSeekV32ToolParser(ToolParser):
77 "vLLM Successfully import tool parser %s !", self.__class__.__name__
78 )
79
109 - def _generate_tool_call_id(self) -> str:
110 - """Generate a unique tool call ID."""
111 - return f"call_{uuid.uuid4().hex[:24]}"
112 -
80 def adjust_request(self, request):
81 request = super().adjust_request(request)
82 if request.tools and request.tool_choice != "none":
@@ -122,33 +89,77 @@ class DeepSeekV32ToolParser(ToolParser):
89 request.skip_special_tokens = False
90 return request
91
125 - def _reset_streaming_state(self):
126 - """Reset all streaming state."""
127 - self.current_tool_index = 0
128 - self.invoke_index = 0
129 - self.is_tool_call_started = False
130 - self.header_sent = False
131 - self.current_tool_id = None
132 - self.current_function_name = None
133 - self.current_param_name = None
134 - self.current_param_value = ""
135 - self.param_count = 0
136 - self.in_param = False
137 - self.in_function = False
138 - self.json_started = False
139 - self.json_closed = False
140 - # Store accumulated parameters for type conversion
141 - self.accumulated_params = {}
142 - self.streaming_request = None
143 - # Clear previous tool call history to avoid state pollution
144 - self.prev_tool_call_arr.clear()
92 + def _generate_tool_call_id(self) -> str:
93 + """Generate a unique tool call ID."""
94 + return f"call_{uuid.uuid4().hex[:24]}"
95
146 - def _parse_invoke_params(self, invoke_str: str) -> dict | None:
96 + def _parse_invoke_params(self, invoke_str: str) -> dict:
97 param_dict = dict()
98 for param_name, param_val in self.parameter_complete_regex.findall(invoke_str):
99 param_dict[param_name] = param_val
100 return param_dict
101
102 + def _convert_param_value(self, value: str, param_type: str) -> Any:
103 + """Convert parameter value to the correct type."""
104 + if value.lower() == "null":
105 + return None
106 +
107 + param_type = param_type.lower()
108 + if param_type in ["string", "str", "text"]:
109 + return value
110 + elif param_type in ["integer", "int"]:
111 + try:
112 + return int(value)
113 + except (ValueError, TypeError):
114 + return value
115 + elif param_type in ["number", "float"]:
116 + try:
117 + val = float(value)
118 + return val if val != int(val) else int(val)
119 + except (ValueError, TypeError):
120 + return value
121 + elif param_type in ["boolean", "bool"]:
122 + return value.lower() in ["true", "1"]
123 + elif param_type in ["object", "array"]:
124 + try:
125 + return json.loads(value)
126 + except json.JSONDecodeError:
127 + return value
128 + else:
129 + # Try JSON parse first, fallback to string
130 + try:
131 + return json.loads(value)
132 + except json.JSONDecodeError:
133 + return value
134 +
135 + def _convert_params_with_schema(
136 + self,
137 + function_name: str,
138 + param_dict: dict[str, str],
139 + request: ChatCompletionRequest | None,
140 + ) -> dict[str, Any]:
141 + """Convert raw string param values using the tool schema types."""
142 + param_config: dict = {}
143 + if request and request.tools:
144 + for tool in request.tools:
145 + if (
146 + hasattr(tool, "function")
147 + and tool.function.name == function_name
148 + and hasattr(tool.function, "parameters")
149 + ):
150 + schema = tool.function.parameters
151 + if isinstance(schema, dict) and "properties" in schema:
152 + param_config = schema["properties"]
153 + break
154 +
155 + converted: dict[str, Any] = {}
156 + for name, value in param_dict.items():
157 + param_type = "string"
158 + if name in param_config and isinstance(param_config[name], dict):
159 + param_type = param_config[name].get("type", "string")
160 + converted[name] = self._convert_param_value(value, param_type)
161 + return converted
162 +
163 def extract_tool_calls(
164 self,
165 model_output: str,
@@ -200,56 +211,55 @@ class DeepSeekV32ToolParser(ToolParser):
211 tools_called=False, tool_calls=[], content=model_output
212 )
213
203 - def _extract_name(self, name_str: str) -> str:
204 - """Extract name from quoted string."""
205 - name_str = name_str.strip()
206 - if (
207 - name_str.startswith('"')
208 - and name_str.endswith('"')
209 - or name_str.startswith("'")
210 - and name_str.endswith("'")
211 - ):
212 - return name_str[1:-1]
213 - return name_str
214 -
215 - def _extract_param_name(self, input_str: str) -> str:
216 - """Extract param name"""
217 - start = input_str.find('"') + 1
218 - end = input_str.find('"', start)
219 - return input_str[start:end] if start > 0 and end > start else input_str
214 + def _reset_streaming_state(self):
215 + """Reset all streaming state."""
216 + self.current_tool_index = 0
217 + self.is_tool_call_started = False
218 + self.prev_tool_call_arr.clear()
219 + self.streamed_args_for_tool.clear()
220
221 - def _convert_param_value(self, value: str, param_type: str) -> Any:
222 - """Convert parameter value to the correct type."""
223 - if value.lower() == "null":
224 - return None
221 + def _extract_delta_tool_calls(
222 + self,
223 + current_text: str,
224 + request: ChatCompletionRequest | None,
225 + ) -> list[DeltaToolCall]:
226 + """Extract DeltaToolCalls from newly completed <invoke> blocks.
227 +
228 + Tracks progress via ``current_tool_index`` so each block is
229 + extracted exactly once across successive streaming calls.
230 + """
231 + complete_invokes = self.invoke_complete_regex.findall(current_text)
232 + delta_tool_calls: list[DeltaToolCall] = []
233 +
234 + while len(complete_invokes) > self.current_tool_index:
235 + invoke_name, invoke_body = complete_invokes[self.current_tool_index]
236 + param_dict = self._parse_invoke_params(invoke_body)
237 +
238 + converted = self._convert_params_with_schema(
239 + invoke_name, param_dict, request
240 + )
241 + args_json = json.dumps(converted, ensure_ascii=False)
242 + idx = self.current_tool_index
243 + self.current_tool_index += 1
244
226 - param_type = param_type.lower()
227 - if param_type in ["string", "str", "text"]:
228 - return value
229 - elif param_type in ["integer", "int"]:
230 - try:
231 - return int(value)
232 - except (ValueError, TypeError):
233 - return value
234 - elif param_type in ["number", "float"]:
235 - try:
236 - val = float(value)
237 - return val if val != int(val) else int(val)
238 - except (ValueError, TypeError):
239 - return value
240 - elif param_type in ["boolean", "bool"]:
241 - return value.lower() in ["true", "1"]
242 - elif param_type in ["object", "array"]:
243 - try:
244 - return json.loads(value)
245 - except json.JSONDecodeError:
246 - return value
247 - else:
248 - # Try JSON parse first, fallback to string
249 - try:
250 - return json.loads(value)
251 - except json.JSONDecodeError:
252 - return value
245 + self.prev_tool_call_arr.append(
246 + {"name": invoke_name, "arguments": converted}
247 + )
248 + self.streamed_args_for_tool.append(args_json)
249 +
250 + delta_tool_calls.append(
251 + DeltaToolCall(
252 + index=idx,
253 + id=self._generate_tool_call_id(),
254 + function=DeltaFunctionCall(
255 + name=invoke_name,
256 + arguments=args_json,
257 + ),
258 + type="function",
259 + )
260 + )
261 +
262 + return delta_tool_calls
263
264 def extract_tool_calls_streaming(
265 self,
@@ -261,345 +271,44 @@ class DeepSeekV32ToolParser(ToolParser):
271 delta_token_ids: Sequence[int],
272 request: ChatCompletionRequest,
273 ) -> DeltaMessage | None:
264 - """Extract tool calls from streaming model output."""
274 + """Extract tool calls from streaming model output.
275 +
276 + Uses a buffer-until-complete-invoke strategy: tokens are buffered
277 + until a complete invoke block is available, then parsed and emitted
278 + in one shot.
279 + """
280
266 - # Store request for type conversion
281 + # First chunk of a new stream — reset state from prior request.
282 if not previous_text:
283 self._reset_streaming_state()
269 - self.streaming_request = request
270 -
271 - # If no delta text, return None unless it's an EOS token after tools
272 - if not delta_text:
273 - # Check if this is an EOS token after all tool calls are complete
274 - if delta_token_ids:
275 - # Count complete tool calls
276 - complete_calls = len(
277 - self.tool_call_complete_regex.findall(current_text)
278 - )
279 -
280 - # If we have completed tool calls and populated prev_tool_call_arr
281 - if complete_calls > 0 and len(self.prev_tool_call_arr) > 0:
282 - # Check if all tool calls are closed
283 - open_calls = current_text.count(
284 - self.tool_call_start_token
285 - ) - current_text.count(self.tool_call_end_token)
286 - if open_calls == 0:
287 - # Return empty delta for finish_reason processing
288 - return DeltaMessage(content="")
289 - elif not self.is_tool_call_started and current_text:
290 - # This is a regular content response that's now complete
291 - return DeltaMessage(content="")
292 - return None
293 -
294 - # Check if we need to advance to next tool
295 - if self.json_closed and not self.in_function:
296 - # Check if this tool call has ended
297 - invoke_ends = current_text.count(self.invoke_end_token)
298 - if invoke_ends > self.current_tool_index:
299 - # This tool has ended, advance to next
300 - self.current_tool_index += 1
301 - self.header_sent = False
302 - self.param_count = 0
303 - self.json_started = False
304 - self.json_closed = False
305 - self.in_function = False # Now we can safely set this to False
306 - self.accumulated_params = {}
307 - # Continue processing next tool
308 - return None
309 -
310 - # Handle normal content before tool calls
311 - if not self.is_tool_call_started:
312 - # Check if tool call is starting
313 - if self.dsml_token in current_text:
314 - self.is_tool_call_started = True
315 - # Return any content before the tool call
316 - if self.dsml_start_check in delta_text:
317 - content_before = delta_text[
318 - : delta_text.index(self.dsml_start_check)
319 - ]
320 - if content_before:
321 - return DeltaMessage(content=content_before)
322 - return None
323 - else:
324 - # Check if we're between tool calls - skip whitespace
325 - if (
326 - current_text.rstrip().endswith(self.tool_call_end_token)
327 - and delta_text.strip() == ""
328 - ):
329 - # We just ended a tool call, skip whitespace
330 - return None
331 - # Normal content, no tool call
332 - if delta_text.endswith("<"):
333 - return DeltaMessage(content=delta_text[:-1])
334 - if previous_text and previous_text.endswith("<"):
335 - return DeltaMessage(content="<" + delta_text)
336 - return DeltaMessage(content=delta_text)
337 -
338 - # Check if we're between tool calls (waiting for next one)
339 - invoke_starts_count = current_text.count(self.invoke_start_prefix)
340 - if self.current_tool_index >= invoke_starts_count:
341 - # We're past all tool calls, shouldn't be here
342 - return None
343 -
344 - # Find the current tool call portion
345 - invoke_start_positions: list[int] = []
346 - idx = 0
347 - while True:
348 - idx = current_text.find(self.invoke_start_prefix, idx)
349 - if idx == -1:
350 - break
351 - invoke_start_positions.append(idx)
352 - idx += len(self.invoke_start_prefix)
353 -
354 - if self.current_tool_index >= len(invoke_start_positions):
355 - # No more tool calls to process yet
356 - return None
284
358 - invoke_start_idx = invoke_start_positions[self.current_tool_index]
359 - # Find where this tool call ends (or current position if not ended yet)
360 - invoke_end_idx = current_text.find(self.invoke_end_token, invoke_start_idx)
361 - if invoke_end_idx == -1:
362 - tool_text = current_text[invoke_start_idx:]
285 + # Detect whether we've entered the tool-call region.
286 + # Use current_text (not delta_text) since the start token may
287 + # be split across chunks.
288 + content_before = None
289 + if self.is_tool_call_started:
290 + pass
291 + elif self.tool_call_start_token in current_text:
292 + # Tool-call region found, capture any plain text before it.
293 + self.is_tool_call_started = True
294 + start_idx = current_text.index(self.tool_call_start_token)
295 + content_before = current_text[len(previous_text) : start_idx] or None
296 else:
364 - tool_text = current_text[
365 - invoke_start_idx : invoke_end_idx + len(self.invoke_end_token)
366 - ]
367 -
368 - # Looking for function header
369 - if not self.header_sent:
370 - if self.invoke_start_prefix in tool_text:
371 - func_start = tool_text.find(self.invoke_start_prefix) + len(
372 - self.invoke_start_prefix
373 - )
374 - # Find the end quote for the function name
375 - func_end = tool_text.find(">", func_start)
376 -
377 - if func_end != -1:
378 - # Found complete function name
379 - function_name_raw = tool_text[func_start:func_end]
380 - self.current_function_name = self._extract_name(function_name_raw)
381 - self.current_tool_id = self._generate_tool_call_id()
382 - self.header_sent = True
383 - self.in_function = True
384 -
385 - # Add to prev_tool_call_arr immediately when we detect a tool call
386 - # Each tool call should be recorded regardless of function name
387 - # Ensure we don't add the same tool call index multiple times
388 - if len(self.prev_tool_call_arr) <= self.current_tool_index:
389 - self.prev_tool_call_arr.append(
390 - {
391 - "name": self.current_function_name,
392 - "arguments": "{}", # Placeholder, will be updated later
393 - }
394 - )
297 + # Still in plain-text region, forward as content.
298 + return DeltaMessage(content=delta_text) if delta_text else None
299
396 - # Send header with function info
397 - return DeltaMessage(
398 - tool_calls=[
399 - DeltaToolCall(
400 - index=self.current_tool_index,
401 - id=self.current_tool_id,
402 - function=DeltaFunctionCall(
403 - name=self.current_function_name, arguments=""
404 - ),
405 - type="function",
406 - )
407 - ]
408 - )
409 - return None
300 + # Inside tool-call region: emit any newly completed invokes.
301 + delta_tool_calls = self._extract_delta_tool_calls(current_text, request)
302
411 - # We've sent header, now handle function body
412 - if self.in_function:
413 - # Send opening brace if not sent yet
414 - if self.in_function and not self.json_started:
415 - self.json_started = True
416 - return DeltaMessage(
417 - tool_calls=[
418 - DeltaToolCall(
419 - index=self.current_tool_index,
420 - function=DeltaFunctionCall(arguments="{"),
421 - )
422 - ]
423 - )
424 -
425 - # Make sure json_started is set if we're processing parameters
426 - if not self.json_started:
427 - self.json_started = True
428 -
429 - # Check for function end in accumulated text
430 - if not self.json_closed and self.invoke_end_token in tool_text:
431 - # Count total parameters in the tool text
432 - total_param_count = tool_text.count(self.parameter_prefix)
433 -
434 - # Only close JSON if all parameters have been processed
435 - if self.param_count >= total_param_count:
436 - # Close JSON
437 - self.json_closed = True
438 -
439 - # Extract complete tool call
440 - # Find the invoke content
441 - invoke_start = tool_text.find(self.invoke_start_prefix) + len(
442 - self.invoke_start_prefix
443 - )
444 - invoke_content_end = tool_text.find(
445 - self.invoke_end_token, invoke_start
446 - )
447 - if invoke_content_end != -1:
448 - invoke_content = tool_text[invoke_start:invoke_content_end]
449 - # Parse to get the complete arguments
450 - try:
451 - invoke_params = self._parse_invoke_params(invoke_content)
452 - if invoke_params and self.current_tool_index < len(
453 - self.prev_tool_call_arr
454 - ):
455 - # Update existing entry in prev_tool_call_arr
456 - self.prev_tool_call_arr[self.current_tool_index][
457 - "arguments"
458 - ] = json.dumps(invoke_params, ensure_ascii=False)
459 - except Exception:
460 - pass # Ignore parsing errors during streaming
461 -
462 - result = DeltaMessage(
463 - tool_calls=[
464 - DeltaToolCall(
465 - index=self.current_tool_index,
466 - function=DeltaFunctionCall(arguments="}"),
467 - )
468 - ]
469 - )
470 -
471 - # Reset state for next tool
472 - self.json_closed = True
473 - self.in_function = False
474 - self.accumulated_params = {}
475 -
476 - logger.debug("[M2_STREAMING] Tool call completed")
477 -
478 - return result
479 - else:
480 - # Don't close JSON yet, continue processing parameters
481 - return None
482 -
483 - # Look for parameters
484 - # Find all parameter starts
485 - param_starts = []
486 - idx = 0
487 - while True:
488 - idx = tool_text.find(self.parameter_prefix, idx)
489 - if idx == -1:
490 - break
491 - param_starts.append(idx)
492 - idx += len(self.parameter_prefix)
493 -
494 - # Check if we should start a new parameter
495 - if (
496 - not self.in_param
497 - and self.param_count < len(param_starts)
498 - and len(param_starts) > self.param_count
499 - ):
500 - # Process the next parameter
501 - param_idx = param_starts[self.param_count]
502 - param_start = param_idx + len(self.parameter_prefix)
503 - remaining = tool_text[param_start:]
504 -
505 - if ">" in remaining:
506 - # We have the complete parameter name
507 - name_end = remaining.find(">")
508 - param_name_raw = remaining[:name_end]
509 - self.current_param_name = self._extract_param_name(param_name_raw)
510 -
511 - # Find the parameter value
512 - value_start = param_start + name_end + 1
513 - value_text = tool_text[value_start:]
514 - if value_text.startswith("\n"):
515 - value_text = value_text[1:]
516 -
517 - # Find where this parameter ends
518 - param_end_idx = value_text.find(self.parameter_end_token)
519 - if param_end_idx == -1:
520 - # No closing tag, look for next parameter or function end
521 - next_param_idx = value_text.find(self.parameter_prefix)
522 - func_end_idx = value_text.find(self.invoke_end_token)
523 -
524 - if next_param_idx != -1 and (
525 - func_end_idx == -1 or next_param_idx < func_end_idx
526 - ):
527 - param_end_idx = next_param_idx
528 - elif func_end_idx != -1:
529 - param_end_idx = func_end_idx
530 - else:
531 - # Neither found, check if tool call is complete
532 - if self.invoke_end_token in tool_text:
533 - # Tool call and parameter is complete
534 - param_end_idx = len(value_text)
535 - else:
536 - # Still streaming, wait for more content
537 - return None
538 -
539 - if param_end_idx != -1:
540 - # Complete parameter found
541 - param_value = value_text[:param_end_idx]
542 - if param_value.endswith("\n"):
543 - param_value = param_value[:-1]
544 -
545 - # Store raw value for later processing
546 - self.accumulated_params[self.current_param_name] = param_value
547 -
548 - # Get parameter configuration for type conversion
549 - param_config = {}
550 - if self.streaming_request and self.streaming_request.tools:
551 - for tool in self.streaming_request.tools:
552 - if (
553 - hasattr(tool, "function")
554 - and tool.function.name == self.current_function_name
555 - and hasattr(tool.function, "parameters")
556 - ):
557 - params = tool.function.parameters
558 - if (
559 - isinstance(params, dict)
560 - and "properties" in params
561 - ):
562 - param_config = params["properties"]
563 - break
564 -
565 - # Get parameter type
566 - param_type = "string"
567 - if (
568 - self.current_param_name in param_config
569 - and isinstance(param_config[self.current_param_name], dict)
570 - and "type" in param_config[self.current_param_name]
571 - ):
572 - param_type = param_config[self.current_param_name]["type"]
573 -
574 - # Convert param value to appropriate type
575 - converted_value = self._convert_param_value(
576 - param_value, param_type
577 - )
578 -
579 - # Build JSON fragment based on the converted type
580 - # Use json.dumps to properly serialize the value
581 - serialized_value = json.dumps(
582 - converted_value, ensure_ascii=False
583 - )
303 + if delta_tool_calls or content_before:
304 + return DeltaMessage(
305 + content=content_before,
306 + tool_calls=delta_tool_calls,
307 + )
308
585 - if self.param_count == 0:
586 - json_fragment = (
587 - f'"{self.current_param_name}": {serialized_value}'
588 - )
589 - else:
590 - json_fragment = (
591 - f', "{self.current_param_name}": {serialized_value}'
592 - )
593 -
594 - self.param_count += 1
595 -
596 - return DeltaMessage(
597 - tool_calls=[
598 - DeltaToolCall(
599 - index=self.current_tool_index,
600 - function=DeltaFunctionCall(arguments=json_fragment),
601 - )
602 - ]
603 - )
309 + # Empty delta with token ids means EOS or closing tag; return
310 + # non-None so the serving framework can finalize finish_reason.
311 + if not delta_text and delta_token_ids and self.prev_tool_call_arr:
312 + return DeltaMessage(content="")
313
314 return None