fix(function_call): skip undefined tools in DeepSeek V3.2/V4 streams

lightllm/server/function_call_parser.py

@@ -1524,8 +1524,15 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
             # Try to find complete invoke blocks first. An invoke is only complete once
             # its end tag is in the buffer, so skip the regex scan until then.
             complete_invoke_match = None
-            if self.invoke_end_token in current_text:
+            while self.invoke_end_token in current_text:
                 complete_invoke_match = self.invoke_regex.search(current_text)
+                if complete_invoke_match is None or complete_invoke_match.group(1) in self._tool_indices:
+                    break
+                logger.warning(f"Model attempted to call undefined function: {complete_invoke_match.group(1)}")
+                self._buffer = current_text[complete_invoke_match.end() :]
+                current_text = self._buffer
+                complete_invoke_match = None
+
             if complete_invoke_match:
                 func_name = complete_invoke_match.group(1)
                 invoke_body = complete_invoke_match.group(2)
@@ -1598,6 +1605,8 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
             if partial_match:
                 func_name = partial_match.group(1)
                 partial_body = partial_match.group(2)
+                if func_name not in self._tool_indices:
+                    return StreamingParseResult()
 
                 if self.current_tool_id == -1:
                     self.current_tool_id = 0
@@ -1611,19 +1620,18 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
                     self.streamed_args_for_tool.append("")
 
                 if not self.current_tool_name_sent:
-                    if func_name in self._tool_indices:
-                        calls.append(
-                            ToolCallItem(
-                                tool_index=self.current_tool_id,
-                                name=func_name,
-                                parameters="",
-                            )
+                    calls.append(
+                        ToolCallItem(
+                            tool_index=self.current_tool_id,
+                            name=func_name,
+                            parameters="",
                         )
-                        self.current_tool_name_sent = True
-                        self.prev_tool_call_arr[self.current_tool_id] = {
-                            "name": func_name,
-                            "arguments": {},
-                        }
+                    )
+                    self.current_tool_name_sent = True
+                    self.prev_tool_call_arr[self.current_tool_id] = {
+                        "name": func_name,
+                        "arguments": {},
+                    }
                 elif partial_body.count(self.param_end_token) > len(self._accumulated_params):
                     # Stream arguments as complete parameters are parsed. Each parameter
                     # ends with its end tag, so only re-parse once a new one has arrived.

unit_tests/server/test_deepseek_tool_parser_stream.py

@@ -82,6 +82,60 @@ def test_dsml_emits_buffered_invoke_on_next_delta():
     assert json.loads(arguments[1]) == {"location": "Paris"}
 
 
+@pytest.mark.parametrize("tool_call_parser,block", [("deepseekv32", "function_calls"), ("deepseekv4", "tool_calls")])
+@pytest.mark.parametrize("chunk_size", [1, 5, 16, None])
+@pytest.mark.parametrize(
+    "tool_names",
+    [
+        ["undefined_tool"],
+        ["undefined_tool", "get_weather"],
+        ["write_file", "undefined_tool", "get_weather"],
+        ["write_file", "get_weather", "undefined_tool"],
+    ],
+)
+def test_dsml_stream_skips_undefined_tools(tool_call_parser, block, chunk_size, tool_names):
+    invokes = [_dsml_invoke(name, "location" if name == "get_weather" else "content", "Paris") for name in tool_names]
+    deltas = [f"<{DSML}{block}>"] + invokes + [f"</{DSML}{block}>"]
+    text = "".join(deltas)
+    if chunk_size is not None:
+        deltas = [text[i : i + chunk_size] for i in range(0, len(text), chunk_size)]
+    elif len(invokes) == 1:
+        deltas = [text]
+
+    _, expected = FunctionCallParser(TOOLS, tool_call_parser).parse_non_stream(text)
+    parser = FunctionCallParser(TOOLS, tool_call_parser)
+    names, arguments = _stream(parser, deltas)
+
+    assert names == [call.name for call in expected]
+    assert list(arguments) == list(range(len(expected)))
+    assert [json.loads(value) for value in arguments.values()] == [json.loads(call.parameters) for call in expected]
+    assert parser.detector.prev_tool_call_arr == [
+        {"name": call.name, "arguments": json.loads(call.parameters)} for call in expected
+    ]
+
+
+@pytest.mark.parametrize("tool_call_parser,block", [("deepseekv32", "function_calls"), ("deepseekv4", "tool_calls")])
+@pytest.mark.parametrize("split_valid_invoke", [False, True])
+def test_dsml_skips_undefined_invokes_before_a_buffered_valid_call(tool_call_parser, block, split_valid_invoke):
+    unknown = _dsml_invoke("undefined_tool", "content", "ignored")
+    valid = _dsml_invoke("get_weather", "location", "Paris")
+    prefix = f"<{DSML}{block}>" + unknown * 3
+    suffix = f"</{DSML}{block}>"
+    if split_valid_invoke:
+        split = valid.index(f"<{DSML}parameter")
+        deltas = [prefix + valid[:split], valid[split:] + suffix]
+    else:
+        deltas = [prefix + valid + suffix]
+
+    parser = FunctionCallParser(TOOLS, tool_call_parser)
+    names, arguments = _stream(parser, deltas)
+
+    assert names == ["get_weather"]
+    assert list(arguments) == [0]
+    assert json.loads(arguments[0]) == {"location": "Paris"}
+    assert parser.detector.prev_tool_call_arr == [{"name": "get_weather", "arguments": {"location": "Paris"}}]
+
+
 def _v3_call(name, payload):
     return "<｜tool▁call▁begin｜>function<｜tool▁sep｜>" + name + "\n```json\n" + payload + "\n```<｜tool▁call▁end｜>"