import json import logging import re from partial_json_parser.core.options import Allow from sglang.srt.entrypoints.openai.protocol import Tool from sglang.srt.function_call.base_format_detector import BaseFormatDetector from sglang.srt.function_call.core_types import ( StreamingParseResult, StructureInfo, ToolCallItem, _GetInfoFunc, ) from sglang.srt.function_call.utils import _find_common_prefix, _partial_json_loads logger = logging.getLogger(__name__) class DeepSeekV32Detector(BaseFormatDetector): """ Detector for DeepSeek V3.2 model function call format. The DeepSeek V3.2 format uses XML-like DSML tags to delimit function calls. Supports two parameter formats: Format 1 - XML Parameter Tags: ``` <｜DSML｜function_calls> <｜DSML｜invoke name="function_name"> <｜DSML｜parameter name="param_name" string="true">value ... ``` Format 2 - Direct JSON: ``` <｜DSML｜function_calls> <｜DSML｜invoke name="function_name"> { "param_name": "value" } ``` Examples: ``` <｜DSML｜function_calls> <｜DSML｜invoke name="get_favorite_tourist_spot"> <｜DSML｜parameter name="city" string="true">San Francisco <｜DSML｜function_calls> <｜DSML｜invoke name="get_favorite_tourist_spot"> { "city": "San Francisco" } ``` Key Components: - Tool Calls Section: Wrapped between `<｜DSML｜function_calls>` and `` - Individual Tool Call: Wrapped between `<｜DSML｜invoke name="...">` and `` - Parameters: Either XML tags or direct JSON format - Supports multiple tool calls Reference: DeepSeek V3.2 format specification """ def __init__(self): super().__init__() self.bot_token = "<｜DSML｜function_calls>" self.eot_token = "" self.invoke_end_token = "" self.parameter_regex = r'<｜DSML｜parameter\s+name="([^"]+)"\s+string="([^"]+)"\s*>(.*?)' self.partial_parameter_regex = ( r'<｜DSML｜parameter\s+name="([^"]+)"\s+string="([^"]+)"\s*>(.*)$' ) self.function_calls_regex = ( r"<｜DSML｜function_calls>(.*?)" ) # Long-form `<｜DSML｜invoke name="x">...` and the # self-closing `<｜DSML｜invoke name="x"/>` shape V4 emits for zero-arg # tools. The `end` group is empty when the closer hasn't streamed in. self.invoke_regex = ( r'<｜DSML｜invoke\s+name="(?P[^"]+)"\s*' r"(?:(?P/>)" r"|>(?P.*?)(?P(?:|$)))" ) self.prefix_parameter_end_call = [" bool: """Check if the text contains a deepseek v32 format tool call.""" return self.bot_token in text or "<｜DSML｜invoke" in text @staticmethod def _unpack_invoke_match(m: "re.Match[str]") -> tuple[str, str, bool]: """Returns (name, body, is_complete) for an invoke_regex match. Self-closing invokes have empty body and are always complete. Long-form bodies are always strings (possibly empty); they're incomplete when matched against `$` because the closing tag hasn't streamed in yet. """ name = m.group("name").strip() if m.group("self_close"): return name, "", True return name, m.group("body"), bool(m.group("end")) def _parse_parameters_from_xml( self, invoke_content: str, allow_partial: bool = False ) -> str: """ Parse parameters from either XML-like format or JSON format to str. Supports two formats: 1. XML parameter tags: <｜DSML｜parameter name="..." string="...">value 2. Direct JSON: { "key": "value" } """ # First, try to parse as direct JSON (new format) invoke_content_stripped = invoke_content.strip() if invoke_content_stripped.startswith("{"): if allow_partial: # Remove incomplete invoke end call prefix in case they are captured by param for token in reversed(self.prefix_invoke_end_call): invoke_content_stripped = invoke_content_stripped.rstrip(token) return invoke_content_stripped elif invoke_content_stripped.endswith("}"): return invoke_content_stripped # Fall back to XML parameter tag parsing (original format) parameters = {} # Find all complete parameter 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() else: # Try to parse as JSON for other types try: 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:] # Remove incomplete parameter_end_call prefix in case they are captured by param for token in reversed(self.prefix_parameter_end_call): remaining_content = remaining_content.rstrip(token) # Match start of a parameter tag + value (potentially incomplete) # Regex: VALUE... (no end tag) partial_match = re.search( self.partial_parameter_regex, remaining_content, re.DOTALL ) if partial_match and (param_value := partial_match.group(3)): param_name = partial_match.group(1) if partial_match.group(2) == "true": parameters[param_name] = param_value.strip() else: try: parameters[param_name] = _partial_json_loads( param_value, Allow.ALL )[0] except json.JSONDecodeError: parameters[param_name] = param_value.strip() return json.dumps(parameters, ensure_ascii=False) def detect_and_parse(self, text: str, tools: list[Tool]) -> StreamingParseResult: """ One-time parsing: Detects and parses tool calls in the provided text. :param text: The complete text to parse. :param tools: List of available tools. :return: ParseResult indicating success or failure, consumed text, leftover text, and parsed calls. """ idx = text.find(self.bot_token) normal_text = text[:idx].removesuffix("\n\n") if idx != -1 else text if self.bot_token not in text: return StreamingParseResult(normal_text=normal_text, calls=[]) calls = [] try: # Extract content between function_calls tags function_calls_match = re.search( self.function_calls_regex, text, re.DOTALL, ) if not function_calls_match: return StreamingParseResult(normal_text=normal_text, calls=[]) function_calls_content = function_calls_match.group(1) # Find all invoke blocks for invoke_match in re.finditer( self.invoke_regex, function_calls_content, re.DOTALL ): func_name, invoke_content, _ = self._unpack_invoke_match(invoke_match) func_args = self._parse_parameters_from_xml(invoke_content) # construct match_result for parse_base_json match_result = {"name": func_name, "parameters": json.loads(func_args)} calls.extend(self.parse_base_json(match_result, tools)) return StreamingParseResult(normal_text=normal_text, calls=calls) except Exception as e: logger.error(f"Error in detect_and_parse: {e}") # return the normal text if parsing fails return StreamingParseResult(normal_text=text) def parse_streaming_increment( self, new_text: str, tools: list[Tool] ) -> StreamingParseResult: """ Streaming incremental parsing tool calls for DeepSeekV32 format. Supports multiple consecutive invoke blocks and argument streaming. """ self._buffer += new_text current_text = self._buffer # Check if buffer contains any DSML markers or ends with potential tag prefix # This handles partial/streaming DSML content dsml_markers = ["｜DSML｜", "<｜", " 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, } # 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 # Move to next tool call self.current_tool_id += 1 self.current_tool_name_sent = False # Continue loop to check for more invoke blocks continue else: # Tool call not complete yet, don't return anything # Wait for more chunks until we see break # No more invoke blocks found return StreamingParseResult(normal_text="", calls=all_calls) except Exception as e: logger.error(f"Error in parse_streaming_increment: {e}") return StreamingParseResult(normal_text=current_text) def structure_info(self) -> _GetInfoFunc: return lambda name: StructureInfo( begin=f'<｜DSML｜invoke name="{name}">', end="", trigger="<｜DSML｜invoke", ) def get_structural_tag_name(self) -> str: return "deepseek_v3_2"