[#21004][fix] DeepSeek V3 and V3.1 tool parsers: complete a tool call that ends in the last delta

tensorrt_llm/serve/tool_parser/deepseekv31_parser.py

@@ -60,6 +60,39 @@ def has_tool_call(self, text: str) -> bool:
         """Check if the text contains a deepseek format tool call."""
         return self.bot_token in text
 
+    def finish(self, tools: List[Tool]) -> StreamingParseResult:
+        """Send the arguments of a tool call whose end came in the last delta.
+
+        An increment sends at most one tool name or the arguments of one call,
+        and the arguments go out one increment after the name. So when the last
+        delta holds the end of a call, its name has been sent and its arguments
+        have not. They are sent here when the buffer holds only that call, with
+        its end token, the same non-empty name and a JSON object that continues
+        what was already sent. In any other case nothing is sent.
+        """
+        func_detail = None
+        if self.current_tool_name_sent and self._buffer.count("<｜tool▁call▁begin｜>") == 1:
+            match = re.search(self.func_call_regex, self._buffer, re.DOTALL)
+            func_detail = match and re.search(self.func_detail_regex, match.group(), re.DOTALL)
+        if not func_detail:
+            return StreamingParseResult()
+        func_name = func_detail.group(1)
+        func_args = func_detail.group(2).strip()
+        if (
+            not func_name
+            or func_name != self.prev_tool_call_arr[self.current_tool_id]["name"]
+            or not (func_args.startswith("{") and is_complete_json(func_args))
+            or not func_args.startswith(self._last_arguments)
+            or func_args == self._last_arguments
+        ):
+            return StreamingParseResult()
+        argument_diff = func_args[len(self._last_arguments) :]
+        self._last_arguments = func_args
+        self.streamed_args_for_tool[self.current_tool_id] += argument_diff
+        return StreamingParseResult(
+            calls=[ToolCallItem(tool_index=self.current_tool_id, parameters=argument_diff)]
+        )
+
     def detect_and_parse(self, text: str, tools: List[Tool]) -> StreamingParseResult:
         """One-time parsing: Detects and parses tool calls in the provided text.
 

tensorrt_llm/serve/tool_parser/deepseekv3_parser.py

@@ -63,6 +63,39 @@ def has_tool_call(self, text: str) -> bool:
         """Check if the text contains a deepseek format tool call."""
         return self.bot_token in text
 
+    def finish(self, tools: List[Tool]) -> StreamingParseResult:
+        """Send the arguments of a tool call whose end came in the last delta.
+
+        An increment sends at most one tool name or the arguments of one call,
+        and the arguments go out one increment after the name. So when the last
+        delta holds the end of a call, its name has been sent and its arguments
+        have not. They are sent here when the buffer holds only that call, with
+        its end token, the same non-empty name and a JSON object that continues
+        what was already sent. In any other case nothing is sent.
+        """
+        func_detail = None
+        if self.current_tool_name_sent and self._buffer.count("<｜tool▁call▁begin｜>") == 1:
+            match = re.search(self.func_call_regex, self._buffer, re.DOTALL)
+            func_detail = match and re.search(self.func_detail_regex, match.group(), re.DOTALL)
+        if not func_detail:
+            return StreamingParseResult()
+        func_name = func_detail.group(2)
+        func_args = func_detail.group(3).strip()
+        if (
+            not func_name
+            or func_name != self.prev_tool_call_arr[self.current_tool_id]["name"]
+            or not (func_args.startswith("{") and is_complete_json(func_args))
+            or not func_args.startswith(self._last_arguments)
+            or func_args == self._last_arguments
+        ):
+            return StreamingParseResult()
+        argument_diff = func_args[len(self._last_arguments) :]
+        self._last_arguments = func_args
+        self.streamed_args_for_tool[self.current_tool_id] += argument_diff
+        return StreamingParseResult(
+            calls=[ToolCallItem(tool_index=self.current_tool_id, parameters=argument_diff)]
+        )
+
     def detect_and_parse(self, text: str, tools: List[Tool]) -> StreamingParseResult:
         """One-time parsing: Detects and parses tool calls in the provided text.
 

tests/unittest/llmapi/apps/test_tool_parsers.py

@@ -3160,6 +3160,111 @@ def test_deepseek_streaming_keeps_markdown_fences(
     assert parser_cls().detect_and_parse(text, sample_tools).normal_text == text
 
 
+_DS_V3_CALL = ("<｜tool▁calls▁begin｜><｜tool▁call▁begin｜>function<｜tool▁sep｜>"
+               'get_weather\n```json\n{"location": "Tokyo"}\n```'
+               "<｜tool▁call▁end｜><｜tool▁calls▁end｜>")
+_DS_V31_CALL = ("<｜tool▁calls▁begin｜><｜tool▁call▁begin｜>get_weather<｜tool▁sep｜>"
+                '{"location": "Tokyo"}<｜tool▁call▁end｜><｜tool▁calls▁end｜>')
+
+
+@pytest.mark.parametrize(
+    "parser_cls, deltas",
+    [
+        # The closing fence and both end tokens arrive in the last delta.
+        (DeepSeekV3Parser, [
+            "<｜tool▁calls▁begin｜><｜tool▁call▁begin｜>function<｜tool▁sep｜>"
+            'get_weather\n```json\n{"location": "Tokyo"}',
+            "\n```<｜tool▁call▁end｜><｜tool▁calls▁end｜>"
+        ]),
+        (DeepSeekV3Parser, [_DS_V3_CALL]),
+        (DeepSeekV31Parser, [_DS_V31_CALL]),
+        # A call without arguments: the separator, the empty arguments and
+        # the end tokens arrive together.
+        (DeepSeekV31Parser, [
+            "<｜tool▁calls▁begin｜><｜tool▁call▁begin｜>get_weather",
+            "<｜tool▁sep｜>{}<｜tool▁call▁end｜><｜tool▁calls▁end｜>"
+        ]),
+        # A stray fence after the call is not part of the arguments.
+        (DeepSeekV3Parser, [
+            _DS_V3_CALL.replace("<｜tool▁calls▁end｜>",
+                                "\n```json\n<｜tool▁calls▁end｜>")
+        ]),
+    ],
+    ids=[
+        "v3_fence_in_last_delta", "v3_whole_call", "v31_whole_call",
+        "v31_no_arguments", "v3_stray_fence_after_call"
+    ],
+)
+def test_deepseek_finish_completes_a_call_from_the_last_delta(
+        sample_tools: list[ChatCompletionToolsParam],
+        parser_cls: type[BaseToolParser], deltas: list[str]) -> None:
+    """finish() completes a tool call whose end came in the last delta.
+
+    Several tokens per delta (speculative decoding, stream_interval) can put
+    the end of a call, or the whole call, in the last delta.
+    """
+    parser = parser_cls()
+    results = [
+        parser.parse_streaming_increment(delta, sample_tools)
+        for delta in deltas
+    ]
+    results.append(parser.finish(sample_tools))
+
+    streamed = {}
+    for result in results:
+        for call in result.calls:
+            name, arguments = streamed.get(call.tool_index, (None, ""))
+            streamed[call.tool_index] = (call.name
+                                         or name, arguments + call.parameters)
+    whole = parser_cls().detect_and_parse("".join(deltas), sample_tools)
+    assert [
+        (name, json.loads(arguments)) for name, arguments in streamed.values()
+    ] == [(call.name, json.loads(call.parameters)) for call in whole.calls]
+
+
+@pytest.mark.parametrize(
+    "parser_cls, deltas",
+    [
+        # Two calls in one delta: the streamed arguments of the first call
+        # already come from the second, so sending the second would repeat it.
+        (DeepSeekV3Parser, [
+            "<｜tool▁calls▁begin｜><｜tool▁call▁begin｜>function<｜tool▁sep｜>"
+            'get_weather\n```json\n{"location": "a"}\n```<｜tool▁call▁end｜>'
+            "<｜tool▁call▁begin｜>function<｜tool▁sep｜>"
+            'get_weather\n```json\n{"loc',
+            'ation": "Tokyo"}\n```<｜tool▁call▁end｜><｜tool▁calls▁end｜>'
+        ]),
+        # The second call's separator arrived while the first call was open.
+        (DeepSeekV31Parser, [
+            "<｜tool▁calls▁begin｜><｜tool▁call▁begin｜>get_weather<｜tool▁sep｜>"
+            '{"location": "a"', ', "unit"',
+            ': "c"}<｜tool▁call▁end｜><｜tool▁call▁begin｜>get_weather<｜tool▁sep｜>',
+            '{"location": "P'
+        ]),
+        # The call has no end token.
+        (DeepSeekV3Parser,
+         [_DS_V3_CALL.replace("<｜tool▁call▁end｜><｜tool▁calls▁end｜>", "")]),
+        (DeepSeekV31Parser,
+         [_DS_V31_CALL.replace("<｜tool▁call▁end｜>", "<｜tool▁call▁en")]),
+    ],
+    ids=[
+        "v3_two_calls_in_one_delta", "v31_second_call_open", "v3_no_end_token",
+        "v31_broken_end_token"
+    ],
+)
+def test_deepseek_finish_sends_nothing_when_the_buffer_is_ambiguous(
+        sample_tools: list[ChatCompletionToolsParam],
+        parser_cls: type[BaseToolParser], deltas: list[str]) -> None:
+    """finish() sends nothing unless the buffer holds exactly one whole call."""
+    parser = parser_cls()
+    for delta in deltas:
+        parser.parse_streaming_increment(delta, sample_tools)
+
+    result = parser.finish(sample_tools)
+
+    assert result.calls == []
+
+
 @pytest.mark.parametrize(
     "parser_cls, tool_call_text",
     [