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 2, 14:45 UTC
[Bugfix] Stream DeepSeek DSML tool-call argument deltas incrementally (#42879)
Signed-off-by: QwertyJack <7554089+QwertyJack@users.noreply.github.com> Co-authored-by: QwertyJack <7554089+QwertyJack@users.noreply.github.com> Co-authored-by: Chauncey <chaunceyjiang@gmail.com>
jack committed
May 28, 2026 at 17:50 UTC
b372ad3e9018f032478619adbc7f7fdcc9318212
3 files changed
+445
-63
tests/tool_parsers/test_deepseekv32_tool_parser.py
+85
-4
@@ -10,6 +10,7 @@ import json
10
from unittest.mock import MagicMock
11
12
import pytest
13
+from openai.types.responses.function_tool import FunctionTool
14
15
from tests.tool_parsers.utils import run_tool_extraction_streaming
16
from vllm.entrypoints.openai.chat_completion.protocol import (
@@ -718,6 +719,81 @@ class TestExtractToolCallsStreaming:
719
"patches": [{"op": "replace", "path": "/schedule", "value": "quiet"}],
720
}
721
722
+ def test_responses_function_tool_schema_in_streaming(self):
723
+ """Responses API FunctionTool schemas must drive streaming conversion."""
724
+ tool = FunctionTool(
725
+ type="function",
726
+ name="toggle",
727
+ parameters={
728
+ "type": "object",
729
+ "properties": {
730
+ "enabled": {"type": "boolean"},
731
+ "count": {"type": "integer"},
732
+ },
733
+ },
734
+ )
735
+ parser = make_parser(tools=[tool])
736
+ full_text = (
737
+ f"{FC_START}\n"
738
+ f'{INV_START}toggle">\n'
739
+ f'{PARAM_START}enabled" string="false">true{PARAM_END}\n'
740
+ f'{PARAM_START}count" string="false">42{PARAM_END}\n'
741
+ f"{INV_END}\n"
742
+ f"{FC_END}"
743
+ )
744
+
745
+ deltas = self._stream(parser, full_text)
746
+ args = json.loads(self._reconstruct_args(deltas))
747
+
748
+ assert args == {"enabled": True, "count": 42}
749
+ assert isinstance(args["enabled"], bool)
750
+ assert isinstance(args["count"], int)
751
+
752
+ def test_streaming_matches_non_streaming_conversion_fallbacks(self):
753
+ """Streaming must reuse conversion fallback semantics for string=false."""
754
+ tool = ChatCompletionToolsParam(
755
+ function=FunctionDefinition(
756
+ name="coerce",
757
+ parameters={
758
+ "type": "object",
759
+ "properties": {
760
+ "union_value": {"type": ["null", "string"]},
761
+ "bad_int": {"type": "integer"},
762
+ "nullable_string": {"type": ["null", "string"]},
763
+ "null_string": {"type": "string"},
764
+ "whole_number": {"type": "number"},
765
+ },
766
+ },
767
+ ),
768
+ )
769
+ parser = make_parser(tools=[tool])
770
+ full_text = (
771
+ f"{FC_START}\n"
772
+ f'{INV_START}coerce">\n'
773
+ f'{PARAM_START}union_value" string="false">hello{PARAM_END}\n'
774
+ f'{PARAM_START}bad_int" string="false">abc{PARAM_END}\n'
775
+ f'{PARAM_START}nullable_string" string="false">null{PARAM_END}\n'
776
+ f'{PARAM_START}null_string" string="false">null{PARAM_END}\n'
777
+ f'{PARAM_START}whole_number" string="false">3.0{PARAM_END}\n'
778
+ f"{INV_END}\n"
779
+ f"{FC_END}"
780
+ )
781
+
782
+ non_stream = parser.extract_tool_calls(full_text, None)
783
+ non_stream_args = json.loads(non_stream.tool_calls[0].function.arguments)
784
+
785
+ deltas = self._stream(parser, full_text)
786
+ stream_args = json.loads(self._reconstruct_args(deltas))
787
+
788
+ assert stream_args == non_stream_args
789
+ assert stream_args == {
790
+ "union_value": "hello",
791
+ "bad_int": "abc",
792
+ "nullable_string": None,
793
+ "null_string": "null",
794
+ "whole_number": 3,
795
+ }
796
+
797
def test_multiple_tools_streaming(self, parser):
798
full_text = (
799
f"{FC_START}\n"
@@ -890,8 +966,8 @@ class TestExtractToolCallsStreaming:
966
assert json.loads(self._reconstruct_args(deltas, tool_index=0)) == {"p": "v1"}
967
assert json.loads(self._reconstruct_args(deltas, tool_index=1)) == {"q": "v2"}
968
893
- def test_no_emission_while_incomplete(self, parser):
894
- """No tool calls should be emitted until an invoke block completes."""
969
+ def test_emits_arguments_before_invoke_completes(self, parser):
970
+ """Argument deltas should stream before the invoke block closes."""
971
# Stream only a partial invoke (no closing tag)
972
partial_text = (
973
f"{FC_START}\n"
@@ -899,8 +975,13 @@ class TestExtractToolCallsStreaming:
975
f'{PARAM_START}k" string="true">val{PARAM_END}\n'
976
)
977
deltas = self._stream(parser, partial_text)
902
- # Should have no tool call deltas yet
903
- assert all(not d.tool_calls for d in deltas)
978
+ arg_chunks = [
979
+ tc.function.arguments
980
+ for delta in deltas
981
+ for tc in delta.tool_calls or []
982
+ if tc.function and tc.function.arguments is not None
983
+ ]
984
+ assert "".join(arg_chunks) == '{"k":"val"'
985
986
def test_no_marker_leak_chunked(self, parser):
987
"""Chunked streaming must NOT leak DSML start-marker fragments
tests/tool_parsers/test_deepseekv4_tool_parser.py
+47
@@ -14,6 +14,7 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
14
ChatCompletionNamedToolChoiceParam,
15
ChatCompletionRequest,
16
ChatCompletionToolsParam,
17
+ FunctionDefinition,
18
)
19
from vllm.tool_parsers import ToolParserManager
20
from vllm.tool_parsers.deepseekv4_tool_parser import DeepSeekV4ToolParser
@@ -164,11 +165,57 @@ def test_streaming_extracts_complete_invokes():
165
for delta in deltas
166
if delta.tool_calls
167
for tool_call in delta.tool_calls
168
+ if tool_call.function.name
169
]
170
assert names == ["search"]
171
assert json.loads(reconstruct_args(deltas)) == {"query": "deepseek v4"}
172
173
174
+def test_streaming_emits_incremental_argument_chunks():
175
+ tool = ChatCompletionToolsParam(
176
+ function=FunctionDefinition(
177
+ name="plan_trip",
178
+ parameters={
179
+ "type": "object",
180
+ "properties": {
181
+ "days": {"type": "integer"},
182
+ "flexible": {"type": "boolean"},
183
+ "cities": {"type": "array", "items": {"type": "string"}},
184
+ "notes": {"type": "string"},
185
+ },
186
+ },
187
+ ),
188
+ )
189
+ parser = make_parser(tools=[tool])
190
+ full_text = (
191
+ f"{TC_START}\n"
192
+ f'{INV_START}plan_trip">\n'
193
+ f'{PARAM_START}days" string="false">3{PARAM_END}\n'
194
+ f'{PARAM_START}flexible" string="false">false{PARAM_END}\n'
195
+ f'{PARAM_START}cities" string="false">'
196
+ f'["Beijing","Shanghai","Tokyo","New York"]{PARAM_END}\n'
197
+ f'{PARAM_START}notes" string="true">靠窗座位{PARAM_END}\n'
198
+ f"{INV_END}\n"
199
+ f"{TC_END}"
200
+ )
201
+
202
+ deltas = stream(parser, full_text, chunk_size=4)
203
+ arg_chunks = [
204
+ tool_call.function.arguments
205
+ for delta in deltas
206
+ for tool_call in delta.tool_calls or []
207
+ if tool_call.function and tool_call.function.arguments is not None
208
+ ]
209
+
210
+ assert len([chunk for chunk in arg_chunks if chunk]) > 2
211
+ assert json.loads("".join(arg_chunks)) == {
212
+ "days": 3,
213
+ "flexible": False,
214
+ "cities": ["Beijing", "Shanghai", "Tokyo", "New York"],
215
+ "notes": "靠窗座位",
216
+ }
217
+
218
+
219
def test_get_vllm_registry_structural_tag_returns_structural_tag(
220
sample_tools: list[ChatCompletionToolsParam],
221
) -> None:
vllm/tool_parsers/deepseekv32_tool_parser.py
+313
-59
@@ -4,7 +4,7 @@
4
import json
5
import uuid
6
from collections.abc import Sequence
7
-from typing import Any
7
+from typing import Any, Literal
8
9
import regex as re
10
@@ -62,6 +62,15 @@ class DeepSeekV32ToolParser(ToolParser):
62
# Streaming state
63
self.current_tool_index: int = 0
64
self._sent_content_idx: int = 0
65
+ self._buffer: str = ""
66
+ self._in_tool_calls: bool = False
67
+ self._active_tool_index: int | None = None
68
+ self._active_tool_name: str | None = None
69
+ self._active_param_name: str | None = None
70
+ self._active_param_string_attr: str | None = None
71
+ self._active_param_mode: str | None = None
72
+ self._active_param_parts: list[str] = []
73
+ self._args_started: list[bool] = []
74
75
# Regex patterns for complete parsing
76
self.tool_call_complete_regex = re.compile(
@@ -77,6 +86,10 @@ class DeepSeekV32ToolParser(ToolParser):
86
r'<|DSML|parameter\s+name="([^"]+)"\s+string="(true|false)"\s*>(.*?)</|DSML|parameter>',
87
re.DOTALL,
88
)
89
+ self.invoke_start_regex = re.compile(r'<|DSML|invoke\s+name="([^"]+)"\s*>')
90
+ self.parameter_start_regex = re.compile(
91
+ r'<|DSML|parameter\s+name="([^"]+)"\s+string="(true|false)"\s*>'
92
+ )
93
94
if not self.model_tokenizer:
95
raise ValueError(
@@ -117,7 +130,7 @@ class DeepSeekV32ToolParser(ToolParser):
130
@staticmethod
131
def _repair_param_dict(
132
param_dict: dict[str, Any],
120
- param_config: dict[str, dict],
133
+ param_config: dict[str, Any],
134
) -> dict[str, Any]:
135
"""Unwrap single 'arguments' / 'input' wrappers when the wrapper
136
is not part of the requested tool schema and the wrapped object
@@ -154,6 +167,15 @@ class DeepSeekV32ToolParser(ToolParser):
167
converted[name] = coerce_to_schema_type(value, param_types)
168
return self._repair_param_dict(converted, param_config)
169
170
+ def _get_param_config(self, function_name: str | None) -> dict[str, Any]:
171
+ if not function_name or not self.tools:
172
+ return {}
173
+ return find_tool_properties(self.tools, function_name)
174
+
175
+ @staticmethod
176
+ def _json_escape_string_content(text: str) -> str:
177
+ return json.dumps(text, ensure_ascii=False)[1:-1]
178
+
179
def extract_tool_calls(
180
self,
181
model_output: str,
@@ -210,67 +232,295 @@ class DeepSeekV32ToolParser(ToolParser):
232
"""Reset all streaming state."""
233
self.current_tool_index = 0
234
self._sent_content_idx = 0
235
+ self._buffer = ""
236
+ self._in_tool_calls = False
237
+ self._active_tool_index = None
238
+ self._active_tool_name = None
239
+ self._active_param_name = None
240
+ self._active_param_string_attr = None
241
+ self._active_param_mode = None
242
+ self._active_param_parts.clear()
243
self.prev_tool_call_arr.clear()
244
self.streamed_args_for_tool.clear()
245
+ self._args_started.clear()
246
216
- def _extract_delta_tool_calls(
247
+ def _add_tool_call_delta(
248
self,
218
- current_text: str,
219
- request: ChatCompletionRequest | None,
220
- ) -> list[DeltaToolCall]:
221
- """Extract DeltaToolCalls from newly completed <invoke> blocks.
249
+ tool_call_deltas: dict[int, DeltaToolCall],
250
+ index: int,
251
+ *,
252
+ call_id: str | None = None,
253
+ call_type: Literal["function"] | None = None,
254
+ name: str | None = None,
255
+ arguments: str | None = None,
256
+ ) -> None:
257
+ if arguments:
258
+ self.streamed_args_for_tool[index] += arguments
259
+
260
+ if index not in tool_call_deltas:
261
+ tool_call_deltas[index] = DeltaToolCall(
262
+ index=index,
263
+ id=call_id,
264
+ type=call_type,
265
+ function=DeltaFunctionCall(name=name, arguments=arguments),
266
+ )
267
+ return
268
+
269
+ delta = tool_call_deltas[index]
270
+ if call_id is not None:
271
+ delta.id = call_id
272
+ if call_type is not None:
273
+ delta.type = call_type
274
+ if delta.function is None:
275
+ delta.function = DeltaFunctionCall()
276
+ if name is not None:
277
+ delta.function.name = name
278
+ if arguments is not None:
279
+ delta.function.arguments = (delta.function.arguments or "") + arguments
280
+
281
+ def _begin_streaming_tool_call(
282
+ self,
283
+ name: str,
284
+ tool_call_deltas: dict[int, DeltaToolCall],
285
+ ) -> None:
286
+ index = self.current_tool_index
287
+ self.current_tool_index += 1
288
+ self._active_tool_index = index
289
+ self._active_tool_name = name
290
+ self.prev_tool_call_arr.append({"name": name, "arguments": {}})
291
+ self.streamed_args_for_tool.append("")
292
+ self._args_started.append(False)
293
+ self._add_tool_call_delta(
294
+ tool_call_deltas,
295
+ index,
296
+ call_id=self._generate_tool_call_id(),
297
+ call_type="function",
298
+ name=name,
299
+ arguments="",
300
+ )
301
223
- Tracks progress via ``current_tool_index`` so each block is
224
- extracted exactly once across successive streaming calls.
225
- """
226
- complete_invokes = self.invoke_complete_regex.findall(current_text)
227
- delta_tool_calls: list[DeltaToolCall] = []
302
+ def _append_param_prefix(
303
+ self,
304
+ tool_call_deltas: dict[int, DeltaToolCall],
305
+ index: int,
306
+ key: str,
307
+ *,
308
+ as_string: bool,
309
+ ) -> None:
310
+ prefix = "{" if not self._args_started[index] else ","
311
+ self._args_started[index] = True
312
+ arguments = prefix + json.dumps(key, ensure_ascii=False) + ":"
313
+ if as_string:
314
+ arguments += '"'
315
+ self._add_tool_call_delta(tool_call_deltas, index, arguments=arguments)
316
+
317
+ def _append_json_param_value(
318
+ self,
319
+ tool_call_deltas: dict[int, DeltaToolCall],
320
+ index: int,
321
+ key: str,
322
+ value: Any,
323
+ ) -> None:
324
+ self._append_param_prefix(tool_call_deltas, index, key, as_string=False)
325
+ self._add_tool_call_delta(
326
+ tool_call_deltas,
327
+ index,
328
+ arguments=json.dumps(value, ensure_ascii=False),
329
+ )
330
229
- while len(complete_invokes) > self.current_tool_index:
230
- invoke_name, invoke_body = complete_invokes[self.current_tool_index]
231
- param_dict = self._parse_invoke_params(invoke_body)
331
+ def _param_types_for_name(self, name: str) -> list[str]:
332
+ param_config = self._get_param_config(self._active_tool_name)
333
+ if name in param_config and isinstance(param_config[name], dict):
334
+ return extract_types_from_schema(param_config[name])
335
+ return ["string"]
336
233
- converted = self._convert_params_with_schema(invoke_name, param_dict)
234
- args_json = json.dumps(converted, ensure_ascii=False)
235
- idx = self.current_tool_index
236
- self.current_tool_index += 1
337
+ @staticmethod
338
+ def _can_stream_raw_param(param_types: list[str]) -> bool:
339
+ # Scalars and unions need the complete value so streaming and
340
+ # non-streaming share the same coercion fallback behavior.
341
+ return set(param_types).issubset({"object", "array"})
342
+
343
+ def _should_buffer_wrapper_param(self, name: str) -> bool:
344
+ if (
345
+ self._active_tool_index is None
346
+ or self._args_started[self._active_tool_index]
347
+ ):
348
+ return False
349
+ param_config = self._get_param_config(self._active_tool_name)
350
+ return bool(
351
+ param_config and name in ("arguments", "input") and name not in param_config
352
+ )
353
238
- self.prev_tool_call_arr.append(
239
- {"name": invoke_name, "arguments": converted}
240
- )
241
- self.streamed_args_for_tool.append(args_json)
242
-
243
- delta_tool_calls.append(
244
- DeltaToolCall(
245
- index=idx,
246
- id=self._generate_tool_call_id(),
247
- function=DeltaFunctionCall(
248
- name=invoke_name,
249
- arguments=args_json,
250
- ),
251
- type="function",
252
- )
253
- )
354
+ def _finish_buffered_param(
355
+ self,
356
+ tool_call_deltas: dict[int, DeltaToolCall],
357
+ index: int,
358
+ ) -> None:
359
+ assert self._active_param_name is not None
360
+ assert self._active_param_string_attr is not None
361
+ raw_value = "".join(self._active_param_parts)
362
+ converted = self._convert_params_with_schema(
363
+ self._active_tool_name or "",
364
+ {self._active_param_name: (raw_value, self._active_param_string_attr)},
365
+ )
366
+ for key, value in converted.items():
367
+ self._append_json_param_value(tool_call_deltas, index, key, value)
368
255
- return delta_tool_calls
369
+ def _close_streaming_tool_call(
370
+ self,
371
+ tool_call_deltas: dict[int, DeltaToolCall],
372
+ ) -> None:
373
+ index = self._active_tool_index
374
+ if index is None:
375
+ return
376
+
377
+ suffix = "}" if self._args_started[index] else "{}"
378
+ self._add_tool_call_delta(tool_call_deltas, index, arguments=suffix)
379
+ try:
380
+ self.prev_tool_call_arr[index] = {
381
+ "name": self._active_tool_name,
382
+ "arguments": json.loads(self.streamed_args_for_tool[index]),
383
+ }
384
+ except (json.JSONDecodeError, IndexError):
385
+ logger.exception("Failed to finalize DeepSeek DSML streaming tool call")
386
+
387
+ self._active_tool_index = None
388
+ self._active_tool_name = None
389
+ self._active_param_name = None
390
+ self._active_param_string_attr = None
391
+ self._active_param_mode = None
392
+ self._active_param_parts.clear()
393
+
394
+ def _process_streaming_buffer(
395
+ self,
396
+ content_parts: list[str],
397
+ tool_call_deltas: dict[int, DeltaToolCall],
398
+ ) -> None:
399
+ parameter_end_token = "</|DSML|parameter>"
400
+ invoke_end_token = "</|DSML|invoke>"
401
+
402
+ while True:
403
+ if not self._in_tool_calls:
404
+ start_idx = self._buffer.find(self.tool_call_start_token)
405
+ if start_idx == -1:
406
+ overlap = partial_tag_overlap(
407
+ self._buffer, self.tool_call_start_token
408
+ )
409
+ sendable_idx = len(self._buffer) - overlap
410
+ if sendable_idx > 0:
411
+ content_parts.append(self._buffer[:sendable_idx])
412
+ self._buffer = self._buffer[sendable_idx:]
413
+ return
414
+
415
+ if start_idx > 0:
416
+ content_parts.append(self._buffer[:start_idx])
417
+ self._buffer = self._buffer[start_idx:]
418
+ continue
419
+
420
+ self._buffer = self._buffer[len(self.tool_call_start_token) :]
421
+ self._in_tool_calls = True
422
+ continue
423
257
- def _extract_content(self, current_text: str) -> str | None:
258
- """Return unsent non-tool-call text, or None.
424
+ if self._active_tool_index is None:
425
+ stripped_len = len(self._buffer) - len(self._buffer.lstrip())
426
+ if stripped_len:
427
+ self._buffer = self._buffer[stripped_len:]
428
+ continue
429
260
- Holds back any suffix that could be a partial start marker
261
- so that split markers are never leaked as content.
262
- """
263
- if self.tool_call_start_token not in current_text:
264
- overlap = partial_tag_overlap(current_text, self.tool_call_start_token)
265
- sendable_idx = len(current_text) - overlap
266
- else:
267
- sendable_idx = current_text.index(self.tool_call_start_token)
268
-
269
- if sendable_idx > self._sent_content_idx:
270
- content = current_text[self._sent_content_idx : sendable_idx]
271
- self._sent_content_idx = sendable_idx
272
- return content
273
- return None
430
+ if self._buffer.startswith(self.tool_call_end_token):
431
+ self._buffer = self._buffer[len(self.tool_call_end_token) :]
432
+ self._in_tool_calls = False
433
+ continue
434
+
435
+ match = self.invoke_start_regex.match(self._buffer)
436
+ if match is None:
437
+ return
438
+
439
+ self._buffer = self._buffer[match.end() :]
440
+ self._begin_streaming_tool_call(match.group(1), tool_call_deltas)
441
+ continue
442
+
443
+ index = self._active_tool_index
444
+
445
+ if self._active_param_mode is not None:
446
+ end_pos = self._buffer.find(parameter_end_token)
447
+ if end_pos != -1:
448
+ raw_content = self._buffer[:end_pos]
449
+ self._buffer = self._buffer[end_pos + len(parameter_end_token) :]
450
+ if self._active_param_mode in ("wrapper", "buffered"):
451
+ self._active_param_parts.append(raw_content)
452
+ self._finish_buffered_param(tool_call_deltas, index)
453
+ elif self._active_param_mode == "string":
454
+ arguments = self._json_escape_string_content(raw_content) + '"'
455
+ self._add_tool_call_delta(
456
+ tool_call_deltas, index, arguments=arguments
457
+ )
458
+ else:
459
+ self._add_tool_call_delta(
460
+ tool_call_deltas, index, arguments=raw_content
461
+ )
462
+
463
+ self._active_param_name = None
464
+ self._active_param_string_attr = None
465
+ self._active_param_mode = None
466
+ self._active_param_parts.clear()
467
+ continue
468
+
469
+ overlap = partial_tag_overlap(self._buffer, parameter_end_token)
470
+ safe_len = len(self._buffer) - overlap
471
+ if safe_len > 0:
472
+ raw_content = self._buffer[:safe_len]
473
+ self._buffer = self._buffer[safe_len:]
474
+ if self._active_param_mode in ("wrapper", "buffered"):
475
+ self._active_param_parts.append(raw_content)
476
+ elif self._active_param_mode == "string":
477
+ self._add_tool_call_delta(
478
+ tool_call_deltas,
479
+ index,
480
+ arguments=self._json_escape_string_content(raw_content),
481
+ )
482
+ else:
483
+ self._add_tool_call_delta(
484
+ tool_call_deltas, index, arguments=raw_content
485
+ )
486
+ return
487
+
488
+ stripped_len = len(self._buffer) - len(self._buffer.lstrip())
489
+ if stripped_len:
490
+ self._buffer = self._buffer[stripped_len:]
491
+ continue
492
+
493
+ if self._buffer.startswith(invoke_end_token):
494
+ self._buffer = self._buffer[len(invoke_end_token) :]
495
+ self._close_streaming_tool_call(tool_call_deltas)
496
+ continue
497
+
498
+ match = self.parameter_start_regex.match(self._buffer)
499
+ if match is None:
500
+ return
501
+
502
+ self._buffer = self._buffer[match.end() :]
503
+ name = match.group(1)
504
+ string_attr = match.group(2)
505
+ self._active_param_name = name
506
+ self._active_param_string_attr = string_attr
507
+
508
+ if self._should_buffer_wrapper_param(name):
509
+ self._active_param_mode = "wrapper"
510
+ continue
511
+
512
+ if string_attr == "true":
513
+ self._append_param_prefix(tool_call_deltas, index, name, as_string=True)
514
+ self._active_param_mode = "string"
515
+ continue
516
+
517
+ param_types = self._param_types_for_name(name)
518
+ if not self._can_stream_raw_param(param_types):
519
+ self._active_param_mode = "buffered"
520
+ continue
521
+
522
+ self._append_param_prefix(tool_call_deltas, index, name, as_string=False)
523
+ self._active_param_mode = "raw"
524
525
def extract_tool_calls_streaming(
526
self,
@@ -284,20 +534,24 @@ class DeepSeekV32ToolParser(ToolParser):
534
) -> DeltaMessage | None:
535
"""Extract tool calls from streaming model output.
536
287
- Uses a buffer-until-complete-invoke strategy: tokens are buffered
288
- until a complete invoke block is available, then parsed and emitted
289
- in one shot.
537
+ Buffers DSML markup while streaming tool-call metadata and argument
538
+ JSON fragments as soon as they are complete enough to be valid deltas.
539
"""
540
541
# First chunk of a new stream — reset state from prior request.
542
if not previous_text:
543
self._reset_streaming_state()
544
296
- content = self._extract_content(current_text)
297
- delta_tool_calls = self._extract_delta_tool_calls(current_text, request)
545
+ self._buffer += delta_text
546
+ content_parts: list[str] = []
547
+ tool_call_deltas: dict[int, DeltaToolCall] = {}
548
+ self._process_streaming_buffer(content_parts, tool_call_deltas)
549
299
- if delta_tool_calls or content:
300
- return DeltaMessage(content=content, tool_calls=delta_tool_calls)
550
+ if content_parts or tool_call_deltas:
551
+ content = "".join(content_parts) or None
552
+ return DeltaMessage(
553
+ content=content, tool_calls=list(tool_call_deltas.values())
554
+ )
555
556
# Empty delta with token ids means EOS or closing tag; return
557
# non-None so the serving framework can finalize finish_reason.