From 5899674504af7a12109332a0d14a96f1c5d5bb9d Mon Sep 17 00:00:00 2001 From: Liangsheng Yin Date: Tue, 11 Aug 2026 20:03:28 -0700 Subject: [PATCH] [Fix] Make DeepSeek-V4 reasoning and tool-call streaming parsing chunk-invariant (#34458) Co-authored-by: hao-cyber <89575785+hao-cyber@users.noreply.github.com> Co-authored-by: Enrico Falco Co-authored-by: Svyatoslav <85786374+slivanovich@users.noreply.github.com> Co-authored-by: Andreas Hassellof Co-authored-by: Leoyzen Co-authored-by: Chenglun Hu Co-authored-by: robellliu-dev Co-authored-by: Gavin.Zhu Co-authored-by: Xinyuan Tong Co-authored-by: tancheng33 Co-authored-by: dineshx29 Co-authored-by: Kangyan Zhou --- .../srt/function_call/deepseekv32_detector.py | 54 +++-- python/sglang/srt/parser/reasoning_parser.py | 59 ++++-- .../function_call/test_deepseekv4_detector.py | 129 ++++++++++++ .../unit/parser/test_reasoning_parser.py | 189 ++++++++++++++++-- 4 files changed, 383 insertions(+), 48 deletions(-) create mode 100644 test/registered/unit/function_call/test_deepseekv4_detector.py diff --git a/python/sglang/srt/function_call/deepseekv32_detector.py b/python/sglang/srt/function_call/deepseekv32_detector.py index a4c1b9dab..4a84a4e33 100644 --- a/python/sglang/srt/function_call/deepseekv32_detector.py +++ b/python/sglang/srt/function_call/deepseekv32_detector.py @@ -2,6 +2,7 @@ import json import logging import re +from partial_json_parser.core.exceptions import MalformedJSON from partial_json_parser.core.options import Allow from sglang.srt.entrypoints.openai.protocol import Tool @@ -179,7 +180,7 @@ class DeepSeekV32Detector(BaseFormatDetector): parameters[param_name] = _partial_json_loads( param_value, Allow.ALL )[0] - except json.JSONDecodeError: + except (json.JSONDecodeError, MalformedJSON, ValueError): parameters[param_name] = param_value.strip() return json.dumps(parameters, ensure_ascii=False) @@ -199,26 +200,25 @@ class DeepSeekV32Detector(BaseFormatDetector): 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: + sections = re.findall(self.function_calls_regex, text, re.DOTALL) + if not sections: 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)) + for function_calls_content in sections: + 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: @@ -259,6 +259,9 @@ class DeepSeekV32Detector(BaseFormatDetector): return StreamingParseResult(normal_text=current_text) all_calls: list[ToolCallItem] = [] + # Only recovered for the first call: the DSML guard above never releases a + # buffer that still holds a marker, so later prose stays buffered. + preamble = "" try: # Loop to handle multiple consecutive invoke blocks while True: @@ -280,6 +283,12 @@ class DeepSeekV32Detector(BaseFormatDetector): self.current_tool_id = 0 self.prev_tool_call_arr = [] self.streamed_args_for_tool = [""] + call_start = invoke_match.start() + bot_pos = current_text.rfind(self.bot_token, 0, call_start) + if bot_pos != -1: + call_start = bot_pos + # Same trailing-newline trim as detect_and_parse, so both agree. + preamble = current_text[:call_start].removesuffix("\n\n") # Ensure arrays are large enough for current tool while len(self.prev_tool_call_arr) <= self.current_tool_id: @@ -355,10 +364,17 @@ class DeepSeekV32Detector(BaseFormatDetector): break # No more invoke blocks found - return StreamingParseResult(normal_text="", calls=all_calls) + return StreamingParseResult(normal_text=preamble, calls=all_calls) except Exception as e: logger.error(f"Error in parse_streaming_increment: {e}") + # Re-emit verbatim rather than swallowing the turn; the preamble is + # still inside current_text unless a completed call advanced past it. + # Calls are dropped on purpose: the failure can land between a tool's + # name and its arguments, and a half-formed call is worse than none. + self._buffer = "" + if not current_text.startswith(preamble): + current_text = preamble + current_text return StreamingParseResult(normal_text=current_text) def structure_info(self) -> _GetInfoFunc: diff --git a/python/sglang/srt/parser/reasoning_parser.py b/python/sglang/srt/parser/reasoning_parser.py index 50b60f543..af57e94e8 100644 --- a/python/sglang/srt/parser/reasoning_parser.py +++ b/python/sglang/srt/parser/reasoning_parser.py @@ -205,6 +205,8 @@ class BaseReasoningFormatDetector: # Strip `` token if present if not self.stripped_think_start and think_start_text in current_text: current_text = current_text.replace(think_start_text, "", 1) + # Write back, or stream_reasoning=False carries the token into finish(). + self._buffer = current_text self.stripped_think_start = True self._in_reasoning = True @@ -224,7 +226,8 @@ class BaseReasoningFormatDetector: # Continue with reasoning content if self._in_reasoning: - # Check for tool_start_token interruption + # Check for tool_start_token interruption. Streaming cannot see a + # think_end_token that has not arrived yet; see the chunk_dependent test. if self.tool_start_token and self.tool_start_token in current_text: tool_idx = current_text.find(self.tool_start_token) reasoning_text = current_text[:tool_idx] @@ -236,9 +239,21 @@ class BaseReasoningFormatDetector: normal_text=normal_text, reasoning_text=reasoning_text ) if self.stream_reasoning: - # Stream the content immediately - self._buffer = "" - return StreamingParseResult(reasoning_text=current_text) + # Minus any trailing slice that could be a token split across chunks. + holdback_tokens = [self.think_end_token] + if self.tool_start_token: + holdback_tokens.append(self.tool_start_token) + if not self.stripped_think_start: + # force_reasoning never saw the opening token; it can still split. + holdback_tokens.append(think_start_text) + holdback = max( + self._ends_with_partial_token(current_text, token) + for token in holdback_tokens + ) + self._buffer = current_text[len(current_text) - holdback :] + return StreamingParseResult( + reasoning_text=current_text[: len(current_text) - holdback] + ) else: return StreamingParseResult() @@ -255,15 +270,30 @@ class BaseReasoningFormatDetector: return text[len(think_start_text) :] return text + @staticmethod + def _ends_with_partial_token(buffer: str, token: str) -> int: + """Length of the longest trailing slice of `buffer` that is a strict prefix + of `token`. Longest, so a token whose prefix repeats inside itself does not + get cut short and leak the rest of the marker.""" + for i in range(min(len(buffer), len(token) - 1), 0, -1): + if token.startswith(buffer[-i:]): + return i + return 0 + def finish(self) -> StreamingParseResult: - """Flush reasoning buffered under stream_reasoning=False when the stream ends - before the end token (e.g. max_tokens cut it short), instead of dropping it. + """Flush reasoning still buffered when the stream ends before the end token + (e.g. max_tokens cut it short), instead of dropping it: the whole block under + stream_reasoning=False, the held-back token suffix under stream_reasoning=True. force_nonempty_content emits it as normal_text, else as reasoning_text.""" if not self._in_reasoning: - return StreamingParseResult() + # Same as the reasoning-side flush below: a held-back slice that never + # became a token is content. + leftover = self._buffer + self._buffer = "" + return StreamingParseResult(normal_text=leftover) - # stream_reasoning=False never clears _buffer, so the opening think token - # (stripped only from the base class's local view) survives here. + # Defensive: subclasses that fill _buffer themselves may not have stripped + # the opening think token that _parse_streaming_increment_impl removes. buffer = self._strip_leading_think_start(self._buffer) self._buffer = "" @@ -274,7 +304,7 @@ class BaseReasoningFormatDetector: return StreamingParseResult(normal_text=normal_text) return StreamingParseResult() - if not self.stream_reasoning and buffer: + if buffer: return StreamingParseResult(reasoning_text=buffer) return StreamingParseResult() @@ -1127,6 +1157,8 @@ class DeepSeekV4Detector(BaseReasoningFormatDetector): dsv4_thinking_start_token, dsv4_thinking_end_token, think_excluded_tokens=[dsv4_eos_token, dsv4_dsml_token], + # Leading "<" included: has_tool_call() matches on it. + tool_start_token=f"<{dsv4_dsml_token}", force_reasoning=force_reasoning, stream_reasoning=stream_reasoning, continue_final_message=continue_final_message, @@ -1185,13 +1217,6 @@ class Apertus2509Detector(BaseReasoningFormatDetector): self._reasoning_acc: str = "" self._in_inner_tool: bool = False - @staticmethod - def _ends_with_partial_token(buffer: str, token: str) -> int: - for i in range(1, min(len(buffer) + 1, len(token))): - if token.startswith(buffer[-i:]): - return i - return 0 - def detect_and_parse(self, text: str) -> StreamingParseResult: blocks = self.detect_and_parse_block_sequence(text) reasoning_parts = [t for k, t in blocks if k == "reasoning"] diff --git a/test/registered/unit/function_call/test_deepseekv4_detector.py b/test/registered/unit/function_call/test_deepseekv4_detector.py new file mode 100644 index 000000000..f6d0ceea5 --- /dev/null +++ b/test/registered/unit/function_call/test_deepseekv4_detector.py @@ -0,0 +1,129 @@ +"""Unit tests for DeepSeekV4Detector DSML streaming — no server, no model loading.""" + +from unittest.mock import patch + +from sglang.srt.entrypoints.openai.protocol import Function, Tool +from sglang.srt.function_call.deepseekv4_detector import DeepSeekV4Detector +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase + +register_cpu_ci(1.0, "base-a-test-cpu") + +DSML = "｜DSML｜" + + +def _wrapped(invoke: str) -> str: + return f"<{DSML}tool_calls>\n{invoke}\n" + + +def _invoke(name: str, params: str = "") -> str: + return f'<{DSML}invoke name="{name}">\n{params}\n' + + +def _param(name: str, is_string: str, value: str) -> str: + return ( + f'<{DSML}parameter name="{name}" string="{is_string}">{value}' + ) + + +def _weather_call(city: str = "SF") -> str: + return _wrapped(_invoke("get_weather", _param("city", "true", city))) + + +class TestDeepSeekV4Streaming(CustomTestCase): + def setUp(self): + self.tools = [ + Tool( + type="function", + function=Function( + name="get_weather", + description="Get weather information", + parameters={ + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + ), + ) + ] + + def _feed(self, chunks): + """Returns (normal_text, calls) accumulated over the chunks.""" + detector = DeepSeekV4Detector() + normal, calls = "", [] + for chunk in chunks: + result = detector.parse_streaming_increment(chunk, self.tools) + normal += result.normal_text + calls.extend(result.calls) + return normal, calls + + def test_preamble_in_same_delta_as_tool_call(self): + """Prose sharing a delta with the tool call must not be dropped, and the + streaming and one-shot paths must agree on it.""" + text = "Let me check.\n" + _weather_call() + normal, calls = self._feed([text]) + + self.assertEqual([c.name for c in calls if c.name], ["get_weather"]) + self.assertEqual( + normal, DeepSeekV4Detector().detect_and_parse(text, self.tools).normal_text + ) + + def test_preamble_before_bare_invoke_without_wrapper(self): + """The bare `<｜DSML｜invoke …>` form has no tool_calls wrapper to walk + back to, so the preamble is computed from the invoke itself.""" + text = "Checking.\n" + _invoke("get_weather", _param("city", "true", "SF")) + normal, calls = self._feed([text]) + + self.assertIn("Checking.", normal) + self.assertEqual([c.name for c in calls if c.name], ["get_weather"]) + + def test_no_dsml_markers_leak_into_normal_text(self): + text = "Prose.\n" + _weather_call() + normal, _ = self._feed([text[i : i + 4] for i in range(0, len(text), 4)]) + + self.assertNotIn(DSML, normal) + + def test_malformed_partial_json_falls_back_to_raw_value(self): + """A partial non-string parameter must not escape as MalformedJSON.""" + detector = DeepSeekV4Detector() + result = detector.parse_streaming_increment( + f'<{DSML}tool_calls>\n<{DSML}invoke name="get_weather">\n' + f'<{DSML}parameter name="city" string="false">{{"a"', + self.tools, + ) + + self.assertEqual([c.name for c in result.calls if c.name], ["get_weather"]) + + def test_non_streaming_parses_every_tool_calls_section(self): + """A turn with two tool_calls sections must yield both calls.""" + result = DeepSeekV4Detector().detect_and_parse( + f"{_weather_call('SF')}\n{_weather_call('NY')}", self.tools + ) + + self.assertEqual(len(result.calls), 2) + + def test_parse_error_neither_swallows_nor_duplicates(self): + """An unexpected parse error must not empty the turn, and the dropped + buffer must not come back on the next delta.""" + detector = DeepSeekV4Detector() + + with patch.object( + DeepSeekV4Detector, + "_parse_parameters_from_xml", + side_effect=RuntimeError("boom"), + ): + first = detector.parse_streaming_increment(_weather_call(), self.tools) + self.assertEqual(detector._buffer, "") + second = detector.parse_streaming_increment(" tail", self.tools) + + self.assertIn("get_weather", first.normal_text) + self.assertNotIn("get_weather", second.normal_text) + # No half-formed call: the failure can land between a tool's name and its + # arguments, so an argument-less named call must not reach the client. + self.assertEqual(first.calls, []) + + +if __name__ == "__main__": + import unittest + + unittest.main() diff --git a/test/registered/unit/parser/test_reasoning_parser.py b/test/registered/unit/parser/test_reasoning_parser.py index 37313ae51..879391af1 100644 --- a/test/registered/unit/parser/test_reasoning_parser.py +++ b/test/registered/unit/parser/test_reasoning_parser.py @@ -156,11 +156,14 @@ class TestBaseReasoningFormatDetector(CustomTestCase): self.assertEqual(detector._buffer, "") self.assertEqual(detector.finish().reasoning_text, "") - def test_finish_drops_partial_end_tag_when_streaming_reasoning(self): - """With stream_reasoning=True the reasoning is emitted chunk by chunk, so - finish() must not re-emit. Only a partial end-tag fragment can linger in - _buffer; that fragment is an incomplete token, not content, and must be - dropped rather than surfaced as reasoning.""" + def test_finish_flushes_partial_end_tag_when_streaming_reasoning(self): + """With stream_reasoning=True everything in _buffer at the end of the + stream is a trailing slice that was held back precisely because it could + still have grown into ``, so it was never emitted. Since the + stream ended it never became a token, and dropping it would lose content + whose only crime is looking like the start of one -- reasoning ending in + a literal `<` is the common case. Flushing matches stream_reasoning=False + and the non-streaming path, which both keep it.""" detector = BaseReasoningFormatDetector( "", "", stream_reasoning=True ) @@ -170,8 +173,11 @@ class TestBaseReasoningFormatDetector(CustomTestCase): ) self.assertEqual(detector.parse_streaming_increment("pick a tool<｜DSML｜tool_calls><｜DSML｜invoke name="s">' + ) + self.assertEqual(result.reasoning_text, "pick a tool") + self.assertTrue(result.normal_text.startswith("<｜DSML｜tool_calls>")) + class TestInklingDetector(CustomTestCase): def test_streaming_routes_blocks_across_all_string_boundaries(self): @@ -416,13 +434,12 @@ class TestGlm45Detector(CustomTestCase): self.assertEqual(result.reasoning_text, "") self.assertEqual(result.normal_text, "") - # Tool interruption should still work - flushes buffered reasoning. - # Note: when stream_reasoning=False, the tag is stripped from the - # local `current_text` variable but NOT from `self._buffer` (which is never - # cleared in the non-streaming path). So the flushed reasoning content - # includes the raw tag. + # Tool interruption should still work - flushes buffered reasoning. The + # opening tag is stripped from `self._buffer` as well as from the local + # view, so the flush matches detect_and_parse instead of carrying the raw + # tag into reasoning_content. result = detector.parse_streaming_increment("tool call") - self.assertEqual(result.reasoning_text, "thinking") + self.assertEqual(result.reasoning_text, "thinking") self.assertEqual(result.normal_text, "tool call") def test_streaming_empty_reasoning_with_tool(self): @@ -961,6 +978,154 @@ class TestBufferLossBugFix(CustomTestCase): self.assertTrue(detector.stripped_think_start) +class TestStreamingChunkSizeInvariance(CustomTestCase): + """Accumulated (reasoning, normal) output must not depend on how the decode + steps happen to batch tokens, and must match one-shot detect_and_parse. + + Speculative decoding and stream_interval > 1 deliver multiple tokens per + step, which splits multi-character tokens like `` across chunk + boundaries. The two `_is_chunk_dependent` tests pin known exceptions. + """ + + CHUNK_SIZES = [1, 2, 3, 5, 7, 11, 23, 1000] + DSML = "｜DSML｜" + + def _feed(self, detector, text, chunk_size): + reasoning = normal = "" + for i in range(0, len(text), chunk_size): + result = detector.parse_streaming_increment(text[i : i + chunk_size]) + reasoning += result.reasoning_text + normal += result.normal_text + result = detector.finish() + return reasoning + result.reasoning_text, normal + result.normal_text + + def _assert_invariant(self, make_detector, text, expected): + for chunk_size in self.CHUNK_SIZES: + with self.subTest(chunk_size=chunk_size): + self.assertEqual( + self._feed(make_detector(), text, chunk_size), expected + ) + one_shot = make_detector().detect_and_parse(text) + self.assertEqual((one_shot.reasoning_text, one_shot.normal_text), expected) + + def test_think_end_split_across_chunks(self): + """`` straddling a chunk boundary must still end the block.""" + self._assert_invariant( + DeepSeekR1Detector, + "abc reasoningnormal text", + ("abc reasoning", "normal text"), + ) + + def test_think_end_split_buffered_mode(self): + self._assert_invariant( + lambda: DeepSeekR1Detector(stream_reasoning=False), + "abc reasoningnormal text", + ("abc reasoning", "normal text"), + ) + + def test_literal_angle_bracket_in_reasoning_is_not_swallowed(self): + self._assert_invariant( + DeepSeekR1Detector, + "a < btail", + ("a < b", "tail"), + ) + + def test_reasoning_truncated_mid_partial_token(self): + """Reasoning that happens to end in a `` prefix must keep those + characters: the holdback exists to recombine them with the next chunk, so + a stream that ends first must flush rather than swallow them.""" + for chunk_size in self.CHUNK_SIZES: + with self.subTest(chunk_size=chunk_size): + self.assertEqual( + self._feed(DeepSeekR1Detector(), "compare a <", chunk_size), + ("compare a <", ""), + ) + + def test_normal_text_ending_in_token_prefix_survives(self): + """Content after the reasoning block that happens to end in a `` + prefix is buffered by the prefix check; the stream ending must flush it.""" + for text in ("ab<", "ab", 1)[0].removeprefix(""), + text.split("", 1)[1], + ) + for chunk_size in self.CHUNK_SIZES: + with self.subTest(text=text, chunk_size=chunk_size): + self.assertEqual( + self._feed(DeepSeekR1Detector(), text, chunk_size), expected + ) + + def test_text_before_think_token_is_chunk_dependent(self): + """Accepted divergence, inherited from main: text before `` lands + in reasoning or content depending on where the chunk boundary falls.""" + text = "leadrtail" + variants = { + self._feed(Qwen3Detector(), text, chunk_size) + for chunk_size in self.CHUNK_SIZES + } + + self.assertEqual( + variants, + {("r", "leadtail"), ("", text), ("leadr", "tail")}, + ) + # And the non-streaming path produces yet a fourth split. + one_shot = Qwen3Detector().detect_and_parse(text) + self.assertEqual( + (one_shot.reasoning_text, one_shot.normal_text), ("leadr", "tail") + ) + + def test_dsv4_reasoning_quoting_dsml_is_chunk_dependent(self): + """Accepted divergence: streaming ends the block at the DSML marker, while + one-shot waits to see whether a `` follows. Reachable because the + DSV4 system prompt shows that marker to the model.""" + text = f"format is <{self.DSML}tool_calls>answer" + by_output = {} + for chunk_size in self.CHUNK_SIZES: + by_output.setdefault( + self._feed(DeepSeekV4Detector(), text, chunk_size), [] + ).append(chunk_size) + + self.assertEqual(len(by_output), 2, f"expected two variants, got {by_output}") + early_cut = ("format is ", f"<{self.DSML}tool_calls>answer") + whole_buffer = (f"format is <{self.DSML}tool_calls>", "answer") + self.assertIn(early_cut, by_output) + self.assertIn(whole_buffer, by_output) + + one_shot = DeepSeekV4Detector().detect_and_parse(text) + self.assertEqual((one_shot.reasoning_text, one_shot.normal_text), whole_buffer) + + def test_dsv4_tool_block_after_think_end(self): + tool_call = ( + f"<{self.DSML}tool_calls>" + f'<{self.DSML}invoke name="s">' + f"" + ) + self._assert_invariant( + DeepSeekV4Detector, + f"my reasoning{tool_call}", + ("my reasoning", tool_call), + ) + + def test_dsv4_tool_block_without_think_end(self): + """DSML directly after reasoning must still be routed to normal_text so + the tool call detector can see it.""" + tool_call = ( + f"<{self.DSML}tool_calls>" + f'<{self.DSML}invoke name="s">' + f"" + ) + for chunk_size in self.CHUNK_SIZES: + with self.subTest(chunk_size=chunk_size): + self.assertEqual( + self._feed( + DeepSeekV4Detector(), + f"my reasoning{tool_call}", + chunk_size, + ), + ("my reasoning", tool_call), + ) + + class TestGptOssDetector(CustomTestCase): """Test cases for GptOssDetector which delegates to HarmonyParser."""