feat: DeepSeek-V3.2 Streaming tool call output (#15278)

Signed-off-by: Xinyuan Tong <xinyuantong.cs@gmail.com>
Co-authored-by: momaek <momaek17@gmail.com>
Co-authored-by: Muqi Li <muqi1029@gmail.com>

python/sglang/srt/function_call/deepseekv32_detector.py

@@ -1,7 +1,6 @@
 import json
 import logging
 import re
-from typing import List
 
 from sglang.srt.entrypoints.openai.protocol import Tool
 from sglang.srt.function_call.base_format_detector import BaseFormatDetector
@@ -11,6 +10,7 @@ from sglang.srt.function_call.core_types import (
     ToolCallItem,
     _GetInfoFunc,
 )
+from sglang.srt.function_call.utils import _find_common_prefix
 
 logger = logging.getLogger(__name__)
 
@@ -71,17 +71,26 @@ class DeepSeekV32Detector(BaseFormatDetector):
         super().__init__()
         self.bot_token = "<｜DSML｜function_calls>"
         self.eot_token = "</｜DSML｜function_calls>"
-        self.invoke_begin_regex = r'<｜DSML｜invoke\s+name="([^"]+)"\s*>'
         self.invoke_end_token = "</｜DSML｜invoke>"
         self.parameter_regex = r'<｜DSML｜parameter\s+name="([^"]+)"\s+string="([^"]+)"\s*>(.*?)</｜DSML｜parameter>'
-        self._last_arguments = ""
+        self.partial_parameter_regex = (
+            r'<｜DSML｜parameter\s+name="([^"]+)"\s+string="([^"]+)"\s*>(.*)$'
+        )
+        self.function_calls_regex = (
+            r"<｜DSML｜function_calls>(.*?)</｜DSML｜function_calls>"
+        )
+        self.invoke_regex = (
+            r'<｜DSML｜invoke\s+name="([^"]+)"\s*>(.*?)(</｜DSML｜invoke>|$)'
+        )
         self.current_tool_id = -1
 
     def has_tool_call(self, text: str) -> bool:
         """Check if the text contains a deepseek v32 format tool call."""
         return self.bot_token in text
 
-    def _parse_parameters_from_xml(self, invoke_content: str) -> dict:
+    def _parse_parameters_from_xml(
+        self, invoke_content: str, allow_partial: bool = False
+    ) -> dict:
         """
         Parse parameters from either XML-like format or JSON format to dict.
 
@@ -105,8 +114,18 @@ class DeepSeekV32Detector(BaseFormatDetector):
 
         # Fall back to XML parameter tag parsing (original format)
         parameters = {}
-        param_matches = re.findall(self.parameter_regex, invoke_content, re.DOTALL)
+        # Find all complete parameter matches
-        for param_name, param_type, param_value in param_matches:
+        param_matches = list(
+            re.finditer(self.parameter_regex, invoke_content, re.DOTALL)
+        )
+
+        last_match_end = 0
+        for match in param_matches:
+            param_name = match.group(1)
+            param_type = match.group(2)
+            param_value = match.group(3)
+            last_match_end = match.end()
+
             # Convert value based on type
             if param_type == "true":  # string type
                 parameters[param_name] = param_value.strip()
@@ -116,9 +135,24 @@ class DeepSeekV32Detector(BaseFormatDetector):
                     parameters[param_name] = json.loads(param_value.strip())
                 except (json.JSONDecodeError, ValueError):
                     parameters[param_name] = param_value.strip()
+
+        # If allowed, try to parse a partial parameter at the end
+        if allow_partial:
+            remaining_content = invoke_content[last_match_end:]
+            # Match start of a parameter tag + value (potentially incomplete)
+            # Regex: <tag name="..." string="...">VALUE... (no end tag)
+            partial_match = re.search(
+                self.partial_parameter_regex, remaining_content, re.DOTALL
+            )
+
+            if partial_match:
+                param_name = partial_match.group(1)
+                param_value = partial_match.group(3)
+                parameters[param_name] = param_value
+
         return parameters
 
-    def detect_and_parse(self, text: str, tools: List[Tool]) -> StreamingParseResult:
+    def detect_and_parse(self, text: str, tools: list[Tool]) -> StreamingParseResult:
         """
         One-time parsing: Detects and parses tool calls in the provided text.
 
@@ -135,7 +169,7 @@ class DeepSeekV32Detector(BaseFormatDetector):
         try:
             # Extract content between function_calls tags
             function_calls_match = re.search(
-                r"<｜DSML｜function_calls>(.*?)</｜DSML｜function_calls>",
+                self.function_calls_regex,
                 text,
                 re.DOTALL,
             )
@@ -145,14 +179,11 @@ class DeepSeekV32Detector(BaseFormatDetector):
             function_calls_content = function_calls_match.group(1)
 
             # Find all invoke blocks
-            invoke_pattern = (
-                r'<｜DSML｜invoke\s+name="([^"]+)"\s*>(.*?)</｜DSML｜invoke>'
-            )
             invoke_matches = re.findall(
-                invoke_pattern, function_calls_content, re.DOTALL
+                self.invoke_regex, function_calls_content, re.DOTALL
             )
 
-            for func_name, invoke_content in invoke_matches:
+            for func_name, invoke_content, _ in invoke_matches:
                 # Parse parameters from XML format
                 func_args = self._parse_parameters_from_xml(invoke_content)
                 # construct match_result for parse_base_json
@@ -166,11 +197,11 @@ class DeepSeekV32Detector(BaseFormatDetector):
             return StreamingParseResult(normal_text=text)
 
     def parse_streaming_increment(
-        self, new_text: str, tools: List[Tool]
+        self, new_text: str, tools: list[Tool]
     ) -> StreamingParseResult:
         """
         Streaming incremental parsing tool calls for DeepSeekV32 format.
-        Supports multiple consecutive invoke blocks.
+        Supports multiple consecutive invoke blocks and argument streaming.
         """
         self._buffer += new_text
         current_text = self._buffer
@@ -196,12 +227,9 @@ class DeepSeekV32Detector(BaseFormatDetector):
         if not has_tool_call and not potentially_dsml and not ends_with_prefix:
             self._buffer = ""
             for e_token in [self.eot_token, self.invoke_end_token]:
-                if e_token in new_text:
+                if e_token in current_text:
-                    new_text = new_text.replace(e_token, "")
+                    current_text = current_text.replace(e_token, "")
-            return StreamingParseResult(normal_text=new_text)
+            return StreamingParseResult(normal_text=current_text)
-
-        if not hasattr(self, "_tool_indices"):
-            self._tool_indices = self._get_tool_indices(tools)
 
         all_calls: list[ToolCallItem] = []
         try:
@@ -209,7 +237,7 @@ class DeepSeekV32Detector(BaseFormatDetector):
             while True:
                 # Try to match an invoke block (may be partial)
                 invoke_match = re.search(
-                    pattern=r'<｜DSML｜invoke\s+name="([^"]+)"\s*>(.*?)(</｜DSML｜invoke>|$)',
+                    pattern=self.invoke_regex,
                     string=current_text,
                     flags=re.DOTALL,
                 )
@@ -228,72 +256,74 @@ class DeepSeekV32Detector(BaseFormatDetector):
                     self.prev_tool_call_arr = []
                     self.streamed_args_for_tool = [""]
 
-                # Don't pre-allocate arrays until we actually complete a tool call
-                # This prevents _check_for_unstreamed_tool_args from sending incomplete calls
-
-                # Parse current parameters from XML/JSON
-                current_params = self._parse_parameters_from_xml(invoke_content)
-                current_args_json = json.dumps(current_params, ensure_ascii=False)
-
-                # Check if tool call is complete (has closing tag)
-                if is_tool_end:
-                    # Only emit the tool call when it's complete (saw </｜DSML｜invoke>)
-                    # This ensures each function returns at most once
-                    calls_for_this_invoke: list[ToolCallItem] = []
-
-                    # Note: invoke_content can be empty for functions with no parameters
-                    # This is valid and should NOT be skipped
-
-                    # Send tool name
-                    calls_for_this_invoke.append(
-                        ToolCallItem(
-                            tool_index=self.current_tool_id,
-                            name=func_name,
-                            parameters="",
-                        )
-                    )
-
-                    # Send parameters as complete JSON
-                    # Always send parameters, even if empty, to maintain consistency
-                    calls_for_this_invoke.append(
-                        ToolCallItem(
-                            tool_index=self.current_tool_id,
-                            name=None,
-                            parameters=current_args_json,
-                        )
-                    )
-
                 # Ensure arrays are large enough for current tool
                 while len(self.prev_tool_call_arr) <= self.current_tool_id:
                     self.prev_tool_call_arr.append({})
                 while len(self.streamed_args_for_tool) <= self.current_tool_id:
                     self.streamed_args_for_tool.append("")
 
+                # 1. Send tool name if not sent yet
+                if not self.current_tool_name_sent:
+                    all_calls.append(
+                        ToolCallItem(
+                            tool_index=self.current_tool_id,
+                            name=func_name,
+                            parameters="",
+                        )
+                    )
+                    self.current_tool_name_sent = True
+
+                # 2. Parse current parameters (partial or complete)
+                current_params = self._parse_parameters_from_xml(
+                    invoke_content, allow_partial=not is_tool_end
+                )
+                current_args_json = json.dumps(current_params, ensure_ascii=False)
+
+                # 3. Calculate and send incremental arguments
+                sent_len = len(self.streamed_args_for_tool[self.current_tool_id])
+                prev_params = self.prev_tool_call_arr[self.current_tool_id].get(
+                    "arguments"
+                )
+
+                argument_diff = None
+
+                if is_tool_end:
+                    # If complete, send everything remaining
+                    argument_diff = current_args_json[sent_len:]
+                elif prev_params is not None:
+                    # If partial, send stable prefix diff
+                    prev_args_json = json.dumps(prev_params, ensure_ascii=False)
+                    if current_args_json != prev_args_json:
+                        prefix = _find_common_prefix(prev_args_json, current_args_json)
+                        if len(prefix) > sent_len:
+                            argument_diff = prefix[sent_len:]
+
+                if argument_diff:
+                    all_calls.append(
+                        ToolCallItem(
+                            tool_index=self.current_tool_id,
+                            name=None,
+                            parameters=argument_diff,
+                        )
+                    )
+                    self.streamed_args_for_tool[self.current_tool_id] += argument_diff
+
                 # Update the stored arguments
                 self.prev_tool_call_arr[self.current_tool_id] = {
                     "name": func_name,
                     "arguments": current_params,
                 }
-                    self.streamed_args_for_tool[self.current_tool_id] = (
-                        current_args_json
-                    )
 
+                # Check if tool call is complete (has closing tag)
+                if is_tool_end:
                     # Remove the completed tool call from buffer
                     self._buffer = current_text[invoke_match.end() :]
                     current_text = self._buffer  # Update for next iteration
 
-                    # Add calls for this invoke to all_calls
-                    all_calls.extend(calls_for_this_invoke)
-
                     # Move to next tool call
                     self.current_tool_id += 1
-                    self._last_arguments = ""
                     self.current_tool_name_sent = False
 
-                    # Don't pre-allocate arrays for the next tool
-                    # Only allocate when we actually complete a tool call
-                    # This prevents _check_for_unstreamed_tool_args from sending incomplete calls
-
                     # Continue loop to check for more invoke blocks
                     continue
                 else:

test/registered/function_call/test_function_call_parser.py

@@ -1158,6 +1158,9 @@ class TestDeepSeekV32Detector(unittest.TestCase):
             ),
         ]
         self.detector = DeepSeekV32Detector()
+        from transformers import AutoTokenizer
+
+        self.tokenizer = AutoTokenizer.from_pretrained("deepseek-ai/DeepSeek-V3.2")
 
     def test_detect_and_parse_xml_format(self):
         """Test parsing standard XML format (DSML)"""
@@ -1235,12 +1238,16 @@ class TestDeepSeekV32Detector(unittest.TestCase):
         text = """<｜DSML｜function_calls>
             <｜DSML｜invoke name="get_favorite_tourist_spot">
                 <｜DSML｜parameter name="city" string="true">San Francisco</｜DSML｜parameter>
+                <｜DSML｜parameter name="second" string="true">London</｜DSML｜parameter>
+                <｜DSML｜parameter name="topn" string="false">10</｜DSML｜parameter>
+                <｜DSML｜parameter name="obj" string="false">{"name": "John", "age": 30}</｜DSML｜parameter>
             </｜DSML｜invoke>
         </｜DSML｜function_calls>"""
 
-        chunks = [text[i : i + 5] for i in range(0, len(text), 5)]
+        input_ids = self.tokenizer.encode(text, add_special_tokens=False)
+        chunk_ids = [input_ids[i : i + 5] for i in range(0, len(input_ids), 5)]
+        chunks = [self.tokenizer.decode(chunk_id) for chunk_id in chunk_ids]
 
-        accumulated_calls = []
         tool_calls_by_index = {}
 
         for chunk in chunks:
@@ -1282,7 +1289,9 @@ class TestDeepSeekV32Detector(unittest.TestCase):
             </｜DSML｜invoke>
         </｜DSML｜function_calls>"""
 
-        chunks = [text[i : i + 5] for i in range(0, len(text), 5)]
+        input_ids = self.tokenizer.encode(text, add_special_tokens=False)
+        chunk_ids = [input_ids[i : i + 5] for i in range(0, len(input_ids), 5)]
+        chunks = [self.tokenizer.decode(chunk_id) for chunk_id in chunk_ids]
 
         tool_calls_by_index = {}
 
@@ -1369,7 +1378,10 @@ class TestDeepSeekV32Detector(unittest.TestCase):
         self.detector = DeepSeekV32Detector()
 
         # Simulate streaming by splitting into small chunks
-        chunks = [text[i : i + 5] for i in range(0, len(text), 5)]
+        # chunks = [text[i : i + 5] for i in range(0, len(text), 5)]
+        input_ids = self.tokenizer.encode(text, add_special_tokens=False)
+        chunk_ids = [input_ids[i : i + 5] for i in range(0, len(input_ids), 5)]
+        chunks = [self.tokenizer.decode(chunk_id) for chunk_id in chunk_ids]
 
         tool_calls_by_index = {}