[#17580][fix] Emit text preceding a tool call in streaming tool parsers

tensorrt_llm/serve/tool_parser/base_tool_parser.py

@@ -1,7 +1,7 @@
 # Adapted from https://github.com/sgl-project/sglang/blob/083629c23564e1a64deaa052f1df5c5d914358d8/python/sglang/srt/function_call/base_format_detector.py
 import json
 from abc import ABC, abstractmethod
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Tuple
 
 from partial_json_parser.core.exceptions import MalformedJSON
 from partial_json_parser.core.options import Allow
@@ -110,6 +110,34 @@ def _ends_with_partial_token(self, buffer: str, bot_token: str) -> int:
                 return i
         return 0
 
+    def _split_leading_normal_text(self, buffer: str,
+                                   start_tokens: List[str]) -> Tuple[str, str]:
+        """Split a buffer at the earliest tool call start token.
+
+        A streaming increment can carry ordinary content and the opening of a
+        tool call at once. That content precedes the tool call and is not part
+        of it, so it has to be returned to the client as normal text instead of
+        being consumed along with the markup.
+
+        The split is verbatim: no whitespace is trimmed. Trimming here would
+        make the streamed content depend on where increment boundaries happen
+        to fall, since a prefix that arrives in its own increment is emitted by
+        the no-tool-call path, which does not trim either. Non-streaming
+        ``detect_and_parse`` strips the prefix in most parsers, so accumulated
+        streaming content can carry whitespace that the one-shot path drops.
+
+        Returns a ``(normal_text, remainder)`` pair. ``normal_text`` is empty
+        when the buffer starts with a tool call or contains none of the tokens.
+        """
+        indices = [
+            index for index in (buffer.find(token) for token in start_tokens)
+            if index != -1
+        ]
+        if not indices:
+            return "", buffer
+        split_at = min(indices)
+        return buffer[:split_at], buffer[split_at:]
+
     def parse_streaming_increment(self, new_text: str,
                                   tools: List[Tool]) -> StreamingParseResult:
         """
@@ -151,6 +179,14 @@ def parse_streaming_increment(self, new_text: str,
         if not hasattr(self, "_tool_indices"):
             self._tool_indices = self._get_tool_indices(tools)
 
+        # This increment may hold ordinary text ahead of the tool call. Hand
+        # that text back as content and keep only the markup in the buffer,
+        # otherwise it is consumed along with the tool call and never reaches
+        # the client.
+        normal_text, current_text = self._split_leading_normal_text(
+            current_text, [self.bot_token])
+        self._buffer = current_text
+
         flags = Allow.ALL if self.current_tool_name_sent else Allow.ALL & ~Allow.STR
 
         try:
@@ -165,7 +201,7 @@ def parse_streaming_increment(self, new_text: str,
                     start_idx = 0
 
                 if start_idx >= len(current_text):
-                    return StreamingParseResult()
+                    return StreamingParseResult(normal_text=normal_text)
 
                 (obj, end_idx) = partial_json_loads(current_text[start_idx:],
                                                     flags)
@@ -183,10 +219,10 @@ def parse_streaming_increment(self, new_text: str,
                 current_tool_call = obj
 
             except MalformedJSON:
-                return StreamingParseResult()
+                return StreamingParseResult(normal_text=normal_text)
 
             if not current_tool_call:
-                return StreamingParseResult()
+                return StreamingParseResult(normal_text=normal_text)
 
             # Case 1: Handle tool name streaming
             # This happens when we encounter a tool but haven't sent its name yet
@@ -284,11 +320,12 @@ def parse_streaming_increment(self, new_text: str,
                 self.prev_tool_call_arr[
                     self.current_tool_id] = current_tool_call
 
+            res.normal_text = normal_text + res.normal_text
             return res
 
         except Exception as e:
             logger.error(f"Error in parse_streaming_increment: {e}")
-            return StreamingParseResult()
+            return StreamingParseResult(normal_text=normal_text)
 
     @abstractmethod
     def has_tool_call(self, text: str) -> bool:

tensorrt_llm/serve/tool_parser/deepseekv31_parser.py

@@ -112,6 +112,14 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
         if not hasattr(self, "_tool_indices"):
             self._tool_indices = self._get_tool_indices(tools)
 
+        # This increment may hold ordinary text ahead of the tool call. Hand that
+        # text back as content and keep only the markup in the buffer, otherwise
+        # it is consumed along with the tool call and never reaches the client.
+        normal_text, current_text = self._split_leading_normal_text(
+            current_text, [self.bot_token, "<｜tool▁call▁begin｜>"]
+        )
+        self._buffer = current_text
+
         calls: list[ToolCallItem] = []
         try:
             partial_match = re.search(
@@ -183,17 +191,17 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
                         else:
                             self._buffer = ""
 
-                        result = StreamingParseResult(normal_text="", calls=calls)
+                        result = StreamingParseResult(normal_text=normal_text, calls=calls)
                         self.current_tool_id += 1
                         self._last_arguments = ""
                         self.current_tool_name_sent = False
                         return result
 
-            return StreamingParseResult(normal_text="", calls=calls)
+            return StreamingParseResult(normal_text=normal_text, calls=calls)
 
         except Exception as e:
             logger.error(f"Error in parse_streaming_increment: {e}")
-            return StreamingParseResult(normal_text=current_text)
+            return StreamingParseResult(normal_text=normal_text + current_text)
 
     def structure_info(self) -> _GetInfoFunc:
         return lambda name: StructureInfo(

tensorrt_llm/serve/tool_parser/deepseekv32_parser.py

@@ -191,6 +191,14 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
         if not hasattr(self, "_tool_indices"):
             self._tool_indices = self._get_tool_indices(tools)
 
+        # This increment may hold ordinary text ahead of the tool call. Hand that
+        # text back as content and keep only the markup in the buffer, otherwise
+        # it is consumed along with the tool call and never reaches the client.
+        normal_text, current_text = self._split_leading_normal_text(
+            current_text, [self.bot_token, "<｜DSML｜invoke"]
+        )
+        self._buffer = current_text
+
         all_calls: list[ToolCallItem] = []
         try:
             # Loop to handle multiple consecutive invoke blocks
@@ -288,11 +296,11 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
                     break
 
             # No more invoke blocks found
-            return StreamingParseResult(normal_text="", calls=all_calls)
+            return StreamingParseResult(normal_text=normal_text, calls=all_calls)
 
         except Exception as e:
             logger.error(f"Error in parse_streaming_increment: {e}")
-            return StreamingParseResult(normal_text=current_text)
+            return StreamingParseResult(normal_text=normal_text + current_text)
 
     def structure_info(self) -> _GetInfoFunc:
         return lambda name: StructureInfo(

tensorrt_llm/serve/tool_parser/deepseekv3_parser.py

@@ -115,6 +115,14 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
         if not hasattr(self, "_tool_indices"):
             self._tool_indices = self._get_tool_indices(tools)
 
+        # This increment may hold ordinary text ahead of the tool call. Hand that
+        # text back as content and keep only the markup in the buffer, otherwise
+        # it is consumed along with the tool call and never reaches the client.
+        normal_text, current_text = self._split_leading_normal_text(
+            current_text, [self.bot_token, "<｜tool▁call▁begin｜>"]
+        )
+        self._buffer = current_text
+
         calls: list[ToolCallItem] = []
         try:
             partial_match = re.search(
@@ -187,17 +195,17 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
                         else:
                             self._buffer = ""
 
-                        result = StreamingParseResult(normal_text="", calls=calls)
+                        result = StreamingParseResult(normal_text=normal_text, calls=calls)
                         self.current_tool_id += 1
                         self._last_arguments = ""
                         self.current_tool_name_sent = False
                         return result
 
-            return StreamingParseResult(normal_text="", calls=calls)
+            return StreamingParseResult(normal_text=normal_text, calls=calls)
 
         except Exception as e:
             logger.error(f"Error in parse_streaming_increment: {e}")
-            return StreamingParseResult(normal_text=current_text)
+            return StreamingParseResult(normal_text=normal_text + current_text)
 
     def structure_info(self) -> _GetInfoFunc:
         return lambda name: StructureInfo(

tensorrt_llm/serve/tool_parser/glm4_parser.py

@@ -338,6 +338,12 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
         if not hasattr(self, "_tool_indices"):
             self._tool_indices = self._get_tool_indices(tools)
 
+        # This increment may hold ordinary text ahead of the tool call. Hand that
+        # text back as content and keep only the markup in the buffer, otherwise
+        # it is consumed along with the tool call and never reaches the client.
+        normal_text, current_text = self._split_leading_normal_text(current_text, [self.bot_token])
+        self._buffer = current_text
+
         calls: list[ToolCallItem] = []
         try:
             partial_match = re.search(
@@ -351,7 +357,7 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
                 is_tool_end = partial_match.group(3)
 
                 if func_name_raw is None or not func_name_raw.strip():
-                    return StreamingParseResult(normal_text="", calls=[])
+                    return StreamingParseResult(normal_text=normal_text, calls=[])
 
                 func_name = func_name_raw.strip()
                 func_args_raw = func_args_raw.strip() if func_args_raw else ""
@@ -442,19 +448,19 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
 
                         self._buffer = current_text[partial_match.end(3) :]
 
-                        result = StreamingParseResult(normal_text="", calls=calls)
+                        result = StreamingParseResult(normal_text=normal_text, calls=calls)
                         self.current_tool_id += 1
                         self._last_arguments = ""
                         self.current_tool_name_sent = False
                         self._streamed_raw_length = 0
                         self._reset_streaming_state()
                         return result
 
-            return StreamingParseResult(normal_text="", calls=calls)
+            return StreamingParseResult(normal_text=normal_text, calls=calls)
 
         except Exception as e:
             logger.error(f"Error in parse_streaming_increment: {e}")
-            return StreamingParseResult(normal_text=current_text)
+            return StreamingParseResult(normal_text=normal_text + current_text)
 
     def _parse_argument_pairs(
         self, pairs: List[Tuple[str, str]], func_name: str, tools: List[Tool]

tensorrt_llm/serve/tool_parser/kimi_k2_tool_parser.py

@@ -118,6 +118,14 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
         if not hasattr(self, "_tool_indices"):
             self._tool_indices = self._get_tool_indices(tools)
 
+        # This increment may hold ordinary text ahead of the tool call. Hand that
+        # text back as content and keep only the markup in the buffer, otherwise
+        # it is consumed along with the tool call and never reaches the client.
+        normal_text, current_text = self._split_leading_normal_text(
+            current_text, [self.bot_token, self.tool_call_start_token]
+        )
+        self._buffer = current_text
+
         calls: list[ToolCallItem] = []
         try:
             match = self.stream_tool_call_portion_regex.search(current_text)
@@ -128,7 +136,7 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
                 m = self.tool_call_id_regex.match(function_id)
                 if not m:
                     logger.warning("Unexpected tool_call_id format: %s", function_id)
-                    return StreamingParseResult(normal_text="", calls=calls)
+                    return StreamingParseResult(normal_text=normal_text, calls=calls)
                 function_name = m.group("name")
 
                 # Initialize state if this is the first tool call
@@ -191,19 +199,19 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
                         else:
                             self._buffer = ""
 
-                        result = StreamingParseResult(normal_text="", calls=calls)
+                        result = StreamingParseResult(normal_text=normal_text, calls=calls)
                         self.current_tool_id += 1
                         self._last_arguments = ""
                         self.current_tool_name_sent = False
                         return result
                     except json.JSONDecodeError:
                         pass
 
-            return StreamingParseResult(normal_text="", calls=calls)
+            return StreamingParseResult(normal_text=normal_text, calls=calls)
 
         except Exception as e:
             logger.error(f"Error in parse_streaming_increment: {e}")
-            return StreamingParseResult(normal_text=current_text)
+            return StreamingParseResult(normal_text=normal_text + current_text)
 
     def structure_info(self) -> _GetInfoFunc:
         """Return function that creates StructureInfo for guided generation."""