From 68d442945f9c5e10eced40a1a3da254efe7e96be Mon Sep 17 00:00:00 2001 From: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:25:54 +0800 Subject: [PATCH] Flush dropped reasoning at stream end when stream_reasoning=False (#32225) --- python/sglang/srt/parser/reasoning_parser.py | 51 ++++++++--- .../unit/parser/test_reasoning_parser.py | 86 +++++++++++++++++++ 2 files changed, 124 insertions(+), 13 deletions(-) diff --git a/python/sglang/srt/parser/reasoning_parser.py b/python/sglang/srt/parser/reasoning_parser.py index d7f1e21fa..091581afc 100644 --- a/python/sglang/srt/parser/reasoning_parser.py +++ b/python/sglang/srt/parser/reasoning_parser.py @@ -227,24 +227,34 @@ class BaseReasoningFormatDetector: return StreamingParseResult() + def _strip_leading_think_start(self, text: str) -> str: + think_start_text = self.think_start_token + self.think_start_self_label + if text.startswith(think_start_text): + return text[len(think_start_text) :] + return text + def finish(self) -> StreamingParseResult: - """ - Called once when the stream ends. If force_nonempty_content is set - and the stream ended mid-reasoning, reclassifies the accumulated - reasoning (plus any partial token still buffered) as normal text. - """ - if self._force_nonempty_content and self._in_reasoning: - # stream_reasoning=False never clears _buffer, so the opening think - # token (stripped only from the base class's local view) survives here. - buffer = self._buffer - think_start_text = self.think_start_token + self.think_start_self_label - if buffer.startswith(think_start_text): - buffer = buffer[len(think_start_text) :] + """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. + force_nonempty_content emits it as normal_text, else as reasoning_text.""" + if not self._in_reasoning: + return StreamingParseResult() + + # stream_reasoning=False never clears _buffer, so the opening think token + # (stripped only from the base class's local view) survives here. + buffer = self._strip_leading_think_start(self._buffer) + self._buffer = "" + + if self._force_nonempty_content: normal_text = self._accumulated_reasoning + buffer self._accumulated_reasoning = "" - self._buffer = "" if normal_text: return StreamingParseResult(normal_text=normal_text) + return StreamingParseResult() + + if not self.stream_reasoning and buffer: + return StreamingParseResult(reasoning_text=buffer) + return StreamingParseResult() @@ -1401,6 +1411,21 @@ class CohereCommand4Detector(BaseReasoningFormatDetector): return StreamingParseResult() + def finish(self) -> StreamingParseResult: + # _in_reasoning stays pinned True here (phase tracked via _reasoning_done), so + # the base finish() would misfile a truncated answer tail as reasoning. + buffer = self._buffer + self._buffer = "" + if not self._reasoning_done: + ret = StreamingParseResult( + reasoning_text=self._strip_leading_think_start(buffer) + ) + elif self._saw_text_start and not self._saw_text_end: + ret = StreamingParseResult(normal_text=buffer) + else: + return StreamingParseResult() + return self._maybe_apply_force_nonempty_content(ret) + class ReasoningParser: """ diff --git a/test/registered/unit/parser/test_reasoning_parser.py b/test/registered/unit/parser/test_reasoning_parser.py index 2ee881701..37313ae51 100644 --- a/test/registered/unit/parser/test_reasoning_parser.py +++ b/test/registered/unit/parser/test_reasoning_parser.py @@ -5,6 +5,7 @@ import unittest from sglang.srt.parser.reasoning_parser import ( Apertus2509Detector, BaseReasoningFormatDetector, + CohereCommand4Detector, DeepSeekR1Detector, DeepSeekV4Detector, Gemma4Detector, @@ -135,6 +136,43 @@ class TestBaseReasoningFormatDetector(CustomTestCase): self.assertEqual(result.reasoning_text, "reasoning") self.assertEqual(result.normal_text, "normal") + def test_finish_flushes_truncated_reasoning_no_stream_reasoning(self): + """Bug regression: with stream_reasoning=False the base detector buffers + the whole thinking block and only emits it on . A stream cut + short (e.g. max_tokens) before left the trace stuck in _buffer, + and finish() dropped it. finish() must now emit it as reasoning, with the + opening think token stripped, matching the non-streaming path.""" + detector = BaseReasoningFormatDetector( + "", "", stream_reasoning=False + ) + detector.parse_streaming_increment("") + self.assertEqual( + detector.parse_streaming_increment("half a thought").reasoning_text, "" + ) + end = detector.finish() + self.assertEqual(end.reasoning_text, "half a thought") + self.assertEqual(end.normal_text, "") + # State is cleared, so a second finish() is a no-op (no duplicate flush). + 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.""" + detector = BaseReasoningFormatDetector( + "", "", stream_reasoning=True + ) + self.assertEqual( + detector.parse_streaming_increment("thought").reasoning_text, + "thought", + ) + self.assertEqual(detector.parse_streaming_increment(" start + token, so the whole output is reasoning until . With + stream_reasoning=False a stream cut before buffered the trace; + finish() must flush it as reasoning instead of dropping it.""" + detector = DeepSeekR1Detector(stream_reasoning=False) + self.assertEqual( + detector.parse_streaming_increment( + "reasoning with no end token" + ).reasoning_text, + "", + ) + end = detector.finish() + self.assertEqual(end.reasoning_text, "reasoning with no end token") + self.assertEqual(end.normal_text, "") + class TestQwen3Detector(CustomTestCase): def setUp(self): @@ -1277,5 +1331,37 @@ class TestPoolsideV1Registered(CustomTestCase): self.assertTrue(rp.detector.thinks_internally) +class TestCohereCommand4DetectorFinish(CustomTestCase): + """finish() flush for Cohere's custom streaming state machine. + + This detector pins _in_reasoning True for its whole run and tracks phase via + _reasoning_done, so it overrides finish() rather than inheriting the base + one, which keys on _in_reasoning.""" + + def test_finish_flushes_truncated_reasoning(self): + """Stream cut mid-thinking (stream_reasoning=False, no <|END_THINKING|>) + must flush the buffered trace as reasoning instead of dropping it.""" + detector = CohereCommand4Detector(stream_reasoning=False) + self.assertEqual( + detector.parse_streaming_increment("partial thinking").reasoning_text, + "", + ) + end = detector.finish() + self.assertEqual(end.reasoning_text, "partial thinking") + self.assertEqual(end.normal_text, "") + + def test_finish_flushes_answer_tail_as_normal_text(self): + """Regression guard for the base-class fix: once reasoning has closed, a + truncated answer tail (stream ended before <|END_TEXT|>) must be flushed + as normal_text. The base finish() keyed on _in_reasoning would misfile it + as reasoning because this detector never clears _in_reasoning.""" + detector = CohereCommand4Detector(stream_reasoning=False) + detector.parse_streaming_increment("thinking<|END_THINKING|>") + detector.parse_streaming_increment("<|START_TEXT|>the answer") + end = detector.finish() + self.assertEqual(end.normal_text, "the answer") + self.assertEqual(end.reasoning_text, "") + + if __name__ == "__main__": unittest.main()