1368 lines
58 KiB
Python
1368 lines
58 KiB
Python
"""Unit tests for srt/parser/reasoning_parser.py"""
|
||
|
||
import unittest
|
||
|
||
from sglang.srt.parser.reasoning_parser import (
|
||
Apertus2509Detector,
|
||
BaseReasoningFormatDetector,
|
||
CohereCommand4Detector,
|
||
DeepSeekR1Detector,
|
||
DeepSeekV4Detector,
|
||
Gemma4Detector,
|
||
Glm45Detector,
|
||
HunyuanDetector,
|
||
InklingDetector,
|
||
KimiDetector,
|
||
KimiK2Detector,
|
||
Nemotron3Detector,
|
||
Qwen3Detector,
|
||
ReasoningParser,
|
||
)
|
||
from sglang.test.ci.ci_register import register_cpu_ci
|
||
from sglang.test.test_utils import CustomTestCase
|
||
|
||
register_cpu_ci(est_time=7, suite="base-a-test-cpu")
|
||
|
||
|
||
class TestBaseReasoningFormatDetector(CustomTestCase):
|
||
def setUp(self):
|
||
self.detector = BaseReasoningFormatDetector(
|
||
think_start_token="<think>",
|
||
think_end_token="</think>",
|
||
force_reasoning=False,
|
||
stream_reasoning=True,
|
||
)
|
||
|
||
def test_detect_and_parse_normal_text(self):
|
||
"""Test parsing normal text without reasoning."""
|
||
text = "This is normal text"
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, text)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_detect_and_parse_with_start_token(self):
|
||
"""Test parsing text starting with think token."""
|
||
text = "<think>This is reasoning"
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "This is reasoning")
|
||
self.assertEqual(result.normal_text, "")
|
||
|
||
def test_detect_and_parse_complete_reasoning(self):
|
||
"""Test parsing complete reasoning block."""
|
||
text = "<think>This is reasoning</think>This is normal"
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "This is reasoning")
|
||
self.assertEqual(result.normal_text, "This is normal")
|
||
|
||
def test_detect_and_parse_force_reasoning(self):
|
||
"""Test forced reasoning mode."""
|
||
detector = BaseReasoningFormatDetector(
|
||
"<think>", "</think>", force_reasoning=True
|
||
)
|
||
text = "This should be reasoning"
|
||
result = detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "This should be reasoning")
|
||
self.assertEqual(result.normal_text, "")
|
||
|
||
def test_parse_streaming_increment_normal(self):
|
||
"""Test streaming parse of normal text."""
|
||
result = self.detector.parse_streaming_increment("Hello world")
|
||
self.assertEqual(result.normal_text, "Hello world")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_parse_streaming_increment_partial_token(self):
|
||
"""Test streaming parse with partial token."""
|
||
# Test partial start token
|
||
result = self.detector.parse_streaming_increment("<thi")
|
||
self.assertEqual(result.normal_text, "")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
# Reset detector and test partial end token when in reasoning mode
|
||
detector = BaseReasoningFormatDetector("<think>", "</think>")
|
||
detector._in_reasoning = True
|
||
result = detector.parse_streaming_increment("</thi")
|
||
self.assertEqual(result.normal_text, "")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_parse_streaming_increment_complete_start(self):
|
||
"""Test streaming parse with complete start token."""
|
||
result = self.detector.parse_streaming_increment("<think>")
|
||
self.assertEqual(result.normal_text, "")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
self.assertTrue(self.detector._in_reasoning)
|
||
self.assertTrue(self.detector.stripped_think_start)
|
||
|
||
def test_parse_streaming_increment_reasoning_content(self):
|
||
"""Test streaming parse of reasoning content."""
|
||
# First add start token
|
||
self.detector.parse_streaming_increment("<think>")
|
||
|
||
# Then add reasoning content
|
||
result = self.detector.parse_streaming_increment("reasoning content")
|
||
self.assertEqual(result.reasoning_text, "reasoning content")
|
||
self.assertEqual(result.normal_text, "")
|
||
|
||
def test_parse_streaming_increment_end_token(self):
|
||
"""Test streaming parse with end token."""
|
||
# Start reasoning mode
|
||
self.detector.parse_streaming_increment("<think>")
|
||
self.detector.parse_streaming_increment("reasoning")
|
||
|
||
# End reasoning - the reasoning content accumulated in previous calls is cleared when end token is found
|
||
result = self.detector.parse_streaming_increment("</think>normal text")
|
||
self.assertEqual(result.reasoning_text, "") # Buffer cleared, returns empty
|
||
self.assertEqual(result.normal_text, "normal text")
|
||
self.assertFalse(self.detector._in_reasoning)
|
||
|
||
def test_parse_streaming_increment_no_stream_reasoning(self):
|
||
"""Test streaming parse without streaming reasoning."""
|
||
detector = BaseReasoningFormatDetector(
|
||
"<think>", "</think>", stream_reasoning=False
|
||
)
|
||
|
||
# Start reasoning mode
|
||
detector.parse_streaming_increment("<think>")
|
||
|
||
# Add reasoning content - should not return content
|
||
result = detector.parse_streaming_increment("reasoning content")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
self.assertEqual(result.normal_text, "")
|
||
|
||
def test_parse_streaming_increment_mixed_content(self):
|
||
"""Test streaming parse with mixed content in one chunk."""
|
||
result = self.detector.parse_streaming_increment(
|
||
"<think>reasoning</think>normal"
|
||
)
|
||
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 </think>. A stream cut
|
||
short (e.g. max_tokens) before </think> 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(
|
||
"<think>", "</think>", stream_reasoning=False
|
||
)
|
||
detector.parse_streaming_increment("<think>")
|
||
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(
|
||
"<think>", "</think>", stream_reasoning=True
|
||
)
|
||
self.assertEqual(
|
||
detector.parse_streaming_increment("<think>thought").reasoning_text,
|
||
"thought",
|
||
)
|
||
self.assertEqual(detector.parse_streaming_increment("</thi").reasoning_text, "")
|
||
end = detector.finish()
|
||
self.assertEqual(end.reasoning_text, "")
|
||
self.assertEqual(end.normal_text, "")
|
||
|
||
|
||
class TestDeepSeekR1Detector(CustomTestCase):
|
||
def setUp(self):
|
||
self.detector = DeepSeekR1Detector()
|
||
|
||
def test_detect_and_parse_with_end_token(self):
|
||
"""Test parsing with end token."""
|
||
text = "I think this is the answer</think>The final answer is 42."
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "I think this is the answer")
|
||
self.assertEqual(result.normal_text, "The final answer is 42.")
|
||
|
||
def test_detect_and_parse_with_start_token(self):
|
||
"""Test parsing deepseek-ai/DeepSeek-R1-0528 format, which generates the <think> token."""
|
||
text = "<think>I need to think about this.</think>The answer is 42."
|
||
result = self.detector.detect_and_parse(text)
|
||
# Should be treated as reasoning because force_reasoning=True
|
||
self.assertEqual(result.reasoning_text, "I need to think about this.")
|
||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||
|
||
def test_finish_flushes_truncated_forced_reasoning(self):
|
||
"""Bug regression: DeepSeek-R1 forces reasoning without a <think> start
|
||
token, so the whole output is reasoning until </think>. With
|
||
stream_reasoning=False a stream cut before </think> 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):
|
||
self.detector = Qwen3Detector()
|
||
|
||
def test_detect_and_parse_without_thinking(self):
|
||
"""Test parsing without thinking (enable_thinking=False case)."""
|
||
text = "Direct answer without thinking."
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, text)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
|
||
class TestDeepSeekV4Detector(CustomTestCase):
|
||
def test_strict_thinking_excludes_deepseek_control_tokens(self):
|
||
detector = ReasoningParser(model_type="deepseek-v4").detector
|
||
self.assertIsInstance(detector, DeepSeekV4Detector)
|
||
self.assertEqual(
|
||
detector.think_excluded_tokens,
|
||
["<｜end▁of▁sentence｜>", "｜DSML｜"],
|
||
)
|
||
|
||
def test_thinking_stays_explicit_opt_in(self):
|
||
detector = ReasoningParser(model_type="deepseek-v4").detector
|
||
self.assertEqual(detector.reasoning_default, "explicit_thinking")
|
||
self.assertTrue(detector.thinks_internally)
|
||
|
||
|
||
class TestInklingDetector(CustomTestCase):
|
||
def test_streaming_routes_blocks_across_all_string_boundaries(self):
|
||
detector = InklingDetector()
|
||
source = (
|
||
"<|message_model|><|content_thinking|>think<|end_message|>"
|
||
"<|message_model|><|content_text|>answer<|end_message|>"
|
||
"<|content_model_end_sampling|>"
|
||
)
|
||
reasoning = ""
|
||
content = ""
|
||
for char in source:
|
||
result = detector.parse_streaming_increment(char)
|
||
reasoning += result.reasoning_text
|
||
content += result.normal_text
|
||
self.assertEqual(reasoning, "think")
|
||
self.assertEqual(content, "answer")
|
||
|
||
def test_tool_header_is_preserved_for_the_tool_parser(self):
|
||
detector = InklingDetector()
|
||
source = (
|
||
"<|message_model|>weather<|content_invoke_tool_json|>"
|
||
'{"name":"weather","args":{"city":"SF"}}<|end_message|>'
|
||
)
|
||
content = ""
|
||
for char in source:
|
||
content += detector.parse_streaming_increment(char).normal_text
|
||
self.assertEqual(content, source)
|
||
|
||
def test_raw_text_tool_framing_is_preserved_for_the_tool_parser(self):
|
||
"""The headerless <|content_invoke_tool_text|> block must survive into
|
||
content so the tool-call detector can surface it, rather than being
|
||
swallowed as header data."""
|
||
detector = InklingDetector()
|
||
source = "<|message_model|><|content_invoke_tool_text|>search<|end_message|>"
|
||
result = detector.detect_and_parse(source)
|
||
self.assertIn("<|content_invoke_tool_text|>", result.normal_text)
|
||
self.assertIn("search", result.normal_text)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_quoted_message_model_token_inside_content_is_preserved(self):
|
||
"""Bug regression: the header branch flipped to header state on ANY
|
||
<|message_model|> occurrence, so a literal token the model wrote
|
||
inside a content block (e.g. quoting the protocol) silently swallowed
|
||
all payload text up to the next control token."""
|
||
detector = InklingDetector()
|
||
source = (
|
||
"<|message_model|><|content_text|>Header token: <|message_model|>"
|
||
" then more text<|end_message|>"
|
||
)
|
||
result = detector.detect_and_parse(source)
|
||
self.assertEqual(
|
||
result.normal_text, "Header token: <|message_model|> then more text"
|
||
)
|
||
|
||
def test_control_token_inside_tool_header_shares_the_full_alphabet(self):
|
||
"""Bug regression: the tool-call detector validated headers against
|
||
INKLING_SPECIAL_TOKENS while the reasoning parser keyed on the larger
|
||
control alphabet (+ <|model_trigger_generation|>), so a control token
|
||
smuggled inside a header passed one machine and not the other."""
|
||
from sglang.srt.function_call.inkling_detector import (
|
||
InklingDetector as ToolDetector,
|
||
)
|
||
|
||
detector = ToolDetector()
|
||
prefix, name = detector._split_trailing_tool_header(
|
||
"<|message_model|>weather<|model_trigger_generation|>"
|
||
)
|
||
self.assertIsNone(name)
|
||
|
||
def test_continuation_stream_text_survives_chunk_boundaries(self):
|
||
"""Bug regression: text arriving with no open block (a
|
||
continue_final_message stream resumes MID text block) was routed to
|
||
content only when a chunk held no control token; a chunk like
|
||
'ld<|end_message|>' silently dropped the 'ld'. All out-of-block text
|
||
must reach content regardless of chunking."""
|
||
source = (
|
||
" world<|end_message|><|message_model|><|content_text|>next<|end_message|>"
|
||
)
|
||
for chunks in (
|
||
[source],
|
||
[
|
||
" wor",
|
||
"ld<|end_message|>",
|
||
"<|message_model|><|content_text|>next<|end_message|>",
|
||
],
|
||
list(source),
|
||
):
|
||
detector = InklingDetector()
|
||
content = ""
|
||
for chunk in chunks:
|
||
content += detector.parse_streaming_increment(chunk).normal_text
|
||
self.assertEqual(content, " worldnext", msg=f"chunks={chunks!r}")
|
||
|
||
def test_finish_flushes_reasoning_truncated_before_end_token(self):
|
||
"""Bug regression: with stream_reasoning=False the detector buffers the
|
||
thinking block and only flushes it on a control/end token. When
|
||
generation is cut mid-block (e.g. max_tokens) the stream ends with no
|
||
end token, so the buffered trace was dropped entirely; finish() must
|
||
emit it, matching the non-streaming detect_and_parse path."""
|
||
detector = InklingDetector(stream_reasoning=False)
|
||
source = "<|message_model|><|content_thinking|>truncated thinking"
|
||
streamed_reasoning = ""
|
||
for char in source:
|
||
streamed_reasoning += detector.parse_streaming_increment(
|
||
char
|
||
).reasoning_text
|
||
# The block never closed, so nothing surfaces mid-stream.
|
||
self.assertEqual(streamed_reasoning, "")
|
||
# finish() flushes the buffered trace instead of dropping it.
|
||
self.assertEqual(detector.finish().reasoning_text, "truncated thinking")
|
||
|
||
|
||
class TestKimiDetector(CustomTestCase):
|
||
def setUp(self):
|
||
self.detector = KimiDetector()
|
||
|
||
def test_detect_and_parse_kimi_no_thinking(self):
|
||
"""Test parsing Kimi format without thinking."""
|
||
text = "Direct answer without thinking tokens."
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, text)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
|
||
class TestKimiK2Detector(CustomTestCase):
|
||
"""Test cases for KimiK2 detector with tool interruption support."""
|
||
|
||
def setUp(self):
|
||
self.detector = KimiK2Detector()
|
||
|
||
def test_init(self):
|
||
"""Test KimiK2Detector initialization."""
|
||
self.assertEqual(self.detector.think_start_token, "<think>")
|
||
self.assertEqual(self.detector.think_end_token, "</think>")
|
||
self.assertEqual(self.detector.tool_start_token, "<|tool_calls_section_begin|>")
|
||
self.assertFalse(self.detector._in_reasoning)
|
||
self.assertTrue(self.detector.stream_reasoning)
|
||
|
||
|
||
class TestGlm45Detector(CustomTestCase):
|
||
"""Test cases for GLM45 detector with tool interruption support."""
|
||
|
||
def setUp(self):
|
||
self.detector = Glm45Detector()
|
||
|
||
def test_detect_and_parse_multiple_tool_calls_find(self):
|
||
"""
|
||
Test that find() finds the FIRST occurrence of tool_start_token.
|
||
|
||
If multiple tool calls exist in buffer, should split at the first one.
|
||
"""
|
||
text = "<think>thinking<tool_call>first tool<tool_call>second tool<tool_call>final tool"
|
||
result = self.detector.detect_and_parse(text)
|
||
# Should split at the first <tool_call>
|
||
self.assertEqual(result.reasoning_text, "thinking")
|
||
self.assertEqual(
|
||
result.normal_text,
|
||
"<tool_call>first tool<tool_call>second tool<tool_call>final tool",
|
||
)
|
||
|
||
def test_detect_and_parse_normal_text_only(self):
|
||
"""Test parsing text without reasoning block."""
|
||
text = "Just the answer without any reasoning."
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, text)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_streaming_no_stream_reasoning(self):
|
||
"""Test streaming without stream_reasoning enabled."""
|
||
detector = Glm45Detector(stream_reasoning=False)
|
||
|
||
# Start reasoning
|
||
detector.parse_streaming_increment("<think>")
|
||
|
||
# Reasoning content is buffered and not returned yet
|
||
result = detector.parse_streaming_increment("thinking")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
self.assertEqual(result.normal_text, "")
|
||
|
||
# Tool interruption should still work - flushes buffered reasoning.
|
||
# Note: when stream_reasoning=False, the <think> 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 <think> tag.
|
||
result = detector.parse_streaming_increment("<tool_call>tool call")
|
||
self.assertEqual(result.reasoning_text, "<think>thinking")
|
||
self.assertEqual(result.normal_text, "<tool_call>tool call")
|
||
|
||
def test_streaming_empty_reasoning_with_tool(self):
|
||
"""Test empty reasoning block followed by tool call."""
|
||
result1 = self.detector.parse_streaming_increment("<think>")
|
||
result2 = self.detector.parse_streaming_increment("<tool_call>tool call")
|
||
self.assertEqual(result2.reasoning_text, "")
|
||
self.assertEqual(result2.normal_text, "<tool_call>tool call")
|
||
|
||
def test_forced_reasoning_mode(self):
|
||
"""Test GLM45 with force_reasoning=True."""
|
||
detector = Glm45Detector(force_reasoning=True)
|
||
|
||
# Without start token, should still be in reasoning mode
|
||
text = "This is reasoning"
|
||
result = detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "This is reasoning")
|
||
self.assertEqual(result.normal_text, "")
|
||
|
||
# Tool interruption should work with forced reasoning
|
||
text = "More reasoning<tool_call>tool call"
|
||
result = detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "More reasoning")
|
||
self.assertEqual(result.normal_text, "<tool_call>tool call")
|
||
|
||
|
||
class TestHunyuanDetector(CustomTestCase):
|
||
"""Test cases for Hunyuan detector with tool interruption support."""
|
||
|
||
def setUp(self):
|
||
self.detector = HunyuanDetector()
|
||
|
||
def test_detect_and_parse_without_thinking(self):
|
||
"""Test parsing without thinking tokens (no_think mode)."""
|
||
text = "Direct answer without thinking."
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, text)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_reasoning_parser_integration(self):
|
||
"""Test Hunyuan through ReasoningParser API."""
|
||
parser = ReasoningParser("hunyuan")
|
||
self.assertIsInstance(parser.detector, HunyuanDetector)
|
||
|
||
# Non-streaming
|
||
reasoning, normal = parser.parse_non_stream(
|
||
"<think>thinking<tool_calls><tool_call>func<tool_sep></tool_call></tool_calls>"
|
||
)
|
||
self.assertEqual(reasoning, "thinking")
|
||
self.assertIn("<tool_calls>", normal)
|
||
|
||
def test_reasoning_parser_streaming(self):
|
||
"""Test Hunyuan streaming through ReasoningParser API."""
|
||
parser = ReasoningParser("hunyuan")
|
||
chunks = ["<think>", "reasoning", "<tool_calls>", "<tool_call>func"]
|
||
all_reasoning = ""
|
||
all_normal = ""
|
||
for chunk in chunks:
|
||
reasoning, normal = parser.parse_stream_chunk(chunk)
|
||
if reasoning:
|
||
all_reasoning += reasoning
|
||
if normal:
|
||
all_normal += normal
|
||
|
||
self.assertEqual(all_reasoning, "reasoning")
|
||
self.assertIn("<tool_calls>", all_normal)
|
||
|
||
|
||
class TestNemotron3Detector(CustomTestCase):
|
||
def setUp(self):
|
||
self.detector = Nemotron3Detector()
|
||
|
||
def test_detect_and_parse_no_thinking(self):
|
||
"""Test parsing without thinking tokens."""
|
||
text = "Direct answer without thinking."
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, text)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_detect_and_parse_reasoning_only(self):
|
||
"""Test parsing when output is all reasoning (no content after </think>)."""
|
||
text = "<think>All reasoning, no answer</think>"
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "All reasoning, no answer")
|
||
self.assertEqual(result.normal_text, "")
|
||
|
||
def test_force_nonempty_content_swaps_when_no_normal_text(self):
|
||
"""Test force_nonempty_content swaps reasoning to content when content is empty."""
|
||
detector = Nemotron3Detector(force_nonempty_content=True)
|
||
text = "<think>All reasoning, no answer</think>"
|
||
result = detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, "All reasoning, no answer")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_force_nonempty_content_no_swap_when_normal_text_exists(self):
|
||
"""Test force_nonempty_content does not swap when content already exists."""
|
||
detector = Nemotron3Detector(force_nonempty_content=True)
|
||
text = "<think>Reasoning here</think>The answer is 42."
|
||
result = detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "Reasoning here")
|
||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||
|
||
def test_force_nonempty_content_truncated_reasoning(self):
|
||
"""Test force_nonempty_content with truncated reasoning (no end token)."""
|
||
detector = Nemotron3Detector(force_nonempty_content=True)
|
||
text = "<think>Truncated reasoning without end token"
|
||
result = detector.detect_and_parse(text)
|
||
# Truncated reasoning has no normal_text, so swap should occur
|
||
self.assertEqual(result.normal_text, "Truncated reasoning without end token")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_streaming_truncated_reasoning_reclassified_on_finish(self):
|
||
"""force_nonempty_content: truncated reasoning (no think_end) is flushed
|
||
as normal_text when the stream ends, so streaming content is non-empty."""
|
||
detector = Nemotron3Detector(force_nonempty_content=True)
|
||
detector.parse_streaming_increment(detector.think_start_token)
|
||
detector.parse_streaming_increment("reasoning part one")
|
||
detector.parse_streaming_increment(" more reasoning")
|
||
end = detector.finish()
|
||
self.assertEqual(end.reasoning_text, "")
|
||
self.assertEqual(end.normal_text, "reasoning part one more reasoning")
|
||
|
||
def test_streaming_tool_start_ends_reasoning_and_noops_finish(self):
|
||
"""tool_start_token interrupts reasoning; finish() then no-ops because
|
||
_in_reasoning is already False."""
|
||
detector = Nemotron3Detector(force_nonempty_content=True)
|
||
detector.parse_streaming_increment(detector.think_start_token)
|
||
detector.parse_streaming_increment("reasoning here")
|
||
result = detector.parse_streaming_increment(
|
||
detector.tool_start_token + "payload"
|
||
)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
self.assertEqual(result.normal_text, detector.tool_start_token + "payload")
|
||
self.assertFalse(detector._in_reasoning)
|
||
end = detector.finish()
|
||
self.assertEqual(end.normal_text, "")
|
||
|
||
def test_streaming_truncated_no_stream_reasoning_strips_think_start(self):
|
||
"""force_nonempty_content + stream_reasoning=False: the opening think
|
||
token must not leak into content when truncation is flushed on finish.
|
||
|
||
Regression: with stream_reasoning=False the base parse_streaming_increment
|
||
never clears _buffer, so the stripped think_start survives in _buffer and
|
||
finish() would prepend it to the reclassified content."""
|
||
detector = Nemotron3Detector(
|
||
force_nonempty_content=True, stream_reasoning=False
|
||
)
|
||
detector.parse_streaming_increment(detector.think_start_token)
|
||
detector.parse_streaming_increment("hidden reasoning")
|
||
end = detector.finish()
|
||
self.assertEqual(end.reasoning_text, "")
|
||
self.assertEqual(end.normal_text, "hidden reasoning")
|
||
self.assertNotIn(detector.think_start_token, end.normal_text)
|
||
|
||
|
||
class TestApertus2509DetectorForceNonempty(CustomTestCase):
|
||
"""force_nonempty_content swap on Apertus2509 (non-streaming, via base helper)."""
|
||
|
||
def test_swap_when_only_reasoning(self):
|
||
detector = Apertus2509Detector(force_nonempty_content=True)
|
||
text = (
|
||
detector.think_start_token
|
||
+ "apertus reasoning only"
|
||
+ detector.think_end_token
|
||
)
|
||
result = detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, "apertus reasoning only")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_no_swap_when_normal_exists(self):
|
||
detector = Apertus2509Detector(force_nonempty_content=True)
|
||
text = (
|
||
detector.think_start_token + "reason" + detector.think_end_token + "answer"
|
||
)
|
||
result = detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "reason")
|
||
self.assertEqual(result.normal_text, "answer")
|
||
|
||
|
||
class TestGemma4Detector(CustomTestCase):
|
||
def setUp(self):
|
||
self.detector = Gemma4Detector()
|
||
|
||
def test_detect_and_parse_complete_reasoning(self):
|
||
"""Test parsing complete Gemma4 reasoning block (think_start_self_label is stripped)."""
|
||
text = "<|channel>thought\nLet me think about this<channel|>The answer is 42."
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.reasoning_text, "Let me think about this")
|
||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||
|
||
def test_detect_and_parse_without_thinking(self):
|
||
"""Test parsing without thinking (enable_thinking=False case)."""
|
||
text = "Direct answer without thinking."
|
||
result = self.detector.detect_and_parse(text)
|
||
self.assertEqual(result.normal_text, text)
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_streaming_end_token_mid_chunk(self):
|
||
"""Test end token arriving in the same chunk as reasoning content."""
|
||
self.detector.parse_streaming_increment("<|channel>thought\n")
|
||
result = self.detector.parse_streaming_increment(
|
||
"some reasoning<channel|>the answer"
|
||
)
|
||
self.assertEqual(result.reasoning_text, "some reasoning")
|
||
self.assertEqual(result.normal_text, "the answer")
|
||
self.assertFalse(self.detector._in_reasoning)
|
||
|
||
def test_streaming_self_label_split_across_chunks(self):
|
||
"""Test self_label ('thought\\n') arriving separately from start token."""
|
||
result1 = self.detector.parse_streaming_increment("<|channel>")
|
||
self.assertEqual(result1.reasoning_text, "")
|
||
self.assertEqual(result1.normal_text, "")
|
||
|
||
result2 = self.detector.parse_streaming_increment("thought\n")
|
||
self.assertTrue(self.detector._in_reasoning)
|
||
|
||
result3 = self.detector.parse_streaming_increment("reasoning here")
|
||
self.assertEqual(result3.reasoning_text, "reasoning here")
|
||
|
||
|
||
class TestReasoningParser(CustomTestCase):
|
||
def test_init_valid_model(self):
|
||
"""Test initialization with valid model types."""
|
||
parser = ReasoningParser("deepseek-r1")
|
||
self.assertIsInstance(parser.detector, DeepSeekR1Detector)
|
||
|
||
parser = ReasoningParser("qwen3")
|
||
self.assertIsInstance(parser.detector, Qwen3Detector)
|
||
|
||
parser = ReasoningParser("kimi")
|
||
self.assertIsInstance(parser.detector, KimiDetector)
|
||
|
||
parser = ReasoningParser("kimi_k2")
|
||
self.assertIsInstance(parser.detector, KimiK2Detector)
|
||
|
||
parser = ReasoningParser("glm45")
|
||
self.assertIsInstance(parser.detector, Glm45Detector)
|
||
|
||
parser = ReasoningParser("hunyuan")
|
||
self.assertIsInstance(parser.detector, HunyuanDetector)
|
||
|
||
parser = ReasoningParser("gemma4")
|
||
self.assertIsInstance(parser.detector, Gemma4Detector)
|
||
|
||
def test_init_invalid_model(self):
|
||
"""Test initialization with invalid model type."""
|
||
with self.assertRaises(ValueError) as context:
|
||
ReasoningParser("invalid-model")
|
||
self.assertIn("Unsupported model type", str(context.exception))
|
||
|
||
def test_init_no_model(self):
|
||
"""Test initialization without model type."""
|
||
with self.assertRaises(ValueError) as context:
|
||
ReasoningParser(None)
|
||
self.assertEqual(str(context.exception), "Model type must be specified")
|
||
|
||
def test_parse_non_stream(self):
|
||
"""Test non-streaming parsing."""
|
||
parser = ReasoningParser("qwen3")
|
||
reasoning, normal = parser.parse_non_stream(
|
||
"<think>Let me think</think>The answer is 42."
|
||
)
|
||
self.assertEqual(reasoning, "Let me think")
|
||
self.assertEqual(normal, "The answer is 42.")
|
||
|
||
def test_parse_non_stream_preserves_payload_whitespace(self):
|
||
"""Non-streaming parsing must not rewrite text inside or after reasoning."""
|
||
parser = ReasoningParser("qwen3")
|
||
reasoning, normal = parser.parse_non_stream(
|
||
"<think>\nLet me think\n</think>\n\nThe answer is 42.\n"
|
||
)
|
||
self.assertEqual(reasoning, "\nLet me think\n")
|
||
self.assertEqual(normal, "\n\nThe answer is 42.\n")
|
||
|
||
def test_parse_non_stream_strips_repeated_leading_start_tokens(self):
|
||
"""Repeated leading start tokens are markers, not reasoning payload."""
|
||
parser = ReasoningParser("qwen3")
|
||
reasoning, normal = parser.parse_non_stream(
|
||
"<think><think>Let me think</think>The answer is 42."
|
||
)
|
||
self.assertEqual(reasoning, "Let me think")
|
||
self.assertEqual(normal, "The answer is 42.")
|
||
|
||
def test_parse_stream_chunk_preserves_payload_whitespace(self):
|
||
"""Streaming parsing preserves the same generated payload whitespace."""
|
||
parser = ReasoningParser("qwen3")
|
||
reasoning, normal = parser.parse_stream_chunk("<think>")
|
||
self.assertEqual(reasoning, "")
|
||
self.assertEqual(normal, "")
|
||
|
||
reasoning, normal = parser.parse_stream_chunk(
|
||
"\nLet me think\n</think>\n\nThe answer is 42.\n"
|
||
)
|
||
self.assertEqual(reasoning, "\nLet me think\n")
|
||
self.assertEqual(normal, "\n\nThe answer is 42.\n")
|
||
|
||
def test_parse_stream_chunk(self):
|
||
"""Test streaming chunk parsing."""
|
||
parser = ReasoningParser("qwen3")
|
||
|
||
# First chunk with start token
|
||
reasoning, normal = parser.parse_stream_chunk("<think>")
|
||
self.assertEqual(reasoning, "")
|
||
self.assertEqual(normal, "")
|
||
|
||
# Second chunk with reasoning content
|
||
reasoning, normal = parser.parse_stream_chunk("thinking...")
|
||
self.assertEqual(reasoning, "thinking...")
|
||
self.assertEqual(normal, "")
|
||
|
||
# Third chunk with end token and normal text
|
||
reasoning, normal = parser.parse_stream_chunk("</think>answer")
|
||
self.assertEqual(reasoning, "") # Buffer cleared when end token processed
|
||
self.assertEqual(normal, "answer")
|
||
|
||
def test_case_insensitive_model_type(self):
|
||
"""Test case insensitive model type matching."""
|
||
parser1 = ReasoningParser("DeepSeek-R1")
|
||
parser2 = ReasoningParser("QWEN3")
|
||
parser3 = ReasoningParser("Kimi")
|
||
|
||
self.assertIsInstance(parser1.detector, DeepSeekR1Detector)
|
||
self.assertIsInstance(parser2.detector, Qwen3Detector)
|
||
self.assertIsInstance(parser3.detector, KimiDetector)
|
||
|
||
def test_stream_reasoning_parameter(self):
|
||
"""Test stream_reasoning parameter is passed correctly."""
|
||
parser = ReasoningParser("qwen3", stream_reasoning=False)
|
||
self.assertFalse(parser.detector.stream_reasoning)
|
||
|
||
parser = ReasoningParser("qwen3", stream_reasoning=True)
|
||
self.assertTrue(parser.detector.stream_reasoning)
|
||
|
||
def test_glm45_tool_interruption(self):
|
||
"""Test GLM45 tool interruption through ReasoningParser API."""
|
||
parser = ReasoningParser("glm45")
|
||
|
||
# Non-streaming: tool interrupt
|
||
reasoning, normal = parser.parse_non_stream(
|
||
"<think>thinking<tool_call>tool call"
|
||
)
|
||
self.assertEqual(reasoning, "thinking")
|
||
self.assertEqual(normal, "<tool_call>tool call")
|
||
|
||
# Streaming: tool interrupt
|
||
parser = ReasoningParser("glm45")
|
||
chunks = ["<think>", "reasoning", "<tool_call>", "tool args"]
|
||
all_reasoning = ""
|
||
all_normal = ""
|
||
for chunk in chunks:
|
||
reasoning, normal = parser.parse_stream_chunk(chunk)
|
||
if reasoning:
|
||
all_reasoning += reasoning
|
||
if normal:
|
||
all_normal += normal
|
||
|
||
self.assertEqual(all_reasoning, "reasoning")
|
||
self.assertEqual(all_normal, "<tool_call>tool args")
|
||
|
||
def test_kimik2_tool_interruption(self):
|
||
"""Test Kimi-K2 tool interruption through ReasoningParser API."""
|
||
parser = ReasoningParser("kimi_k2")
|
||
|
||
# Non-streaming: tool interrupt
|
||
reasoning, normal = parser.parse_non_stream(
|
||
"<think>thinking<|tool_calls_section_begin|><|tool_call_begin|>"
|
||
)
|
||
self.assertEqual(reasoning, "thinking")
|
||
self.assertEqual(normal, "<|tool_calls_section_begin|><|tool_call_begin|>")
|
||
|
||
# Streaming: tool interrupt
|
||
parser = ReasoningParser("kimi_k2")
|
||
chunks = [
|
||
"<think>",
|
||
"reasoning",
|
||
"<|tool_calls_section_begin|>",
|
||
"<|tool_call_begin|>",
|
||
]
|
||
all_reasoning = ""
|
||
all_normal = ""
|
||
for chunk in chunks:
|
||
reasoning, normal = parser.parse_stream_chunk(chunk)
|
||
if reasoning:
|
||
all_reasoning += reasoning
|
||
if normal:
|
||
all_normal += normal
|
||
|
||
self.assertEqual(all_reasoning, "reasoning")
|
||
self.assertEqual(all_normal, "<|tool_calls_section_begin|><|tool_call_begin|>")
|
||
|
||
|
||
class TestIntegrationScenarios(CustomTestCase):
|
||
"""Integration tests for realistic usage scenarios."""
|
||
|
||
def test_kimi_streaming_scenario(self):
|
||
"""Test Kimi streaming scenario."""
|
||
parser = ReasoningParser("kimi")
|
||
chunks = [
|
||
"◁thi",
|
||
"nk▷",
|
||
"Let me analyze this problem.",
|
||
" I need to consider multiple factors.",
|
||
"◁/th",
|
||
"ink▷",
|
||
"The answer is 42.",
|
||
]
|
||
all_reasoning = ""
|
||
all_normal = ""
|
||
for chunk in chunks:
|
||
reasoning, normal = parser.parse_stream_chunk(chunk)
|
||
all_reasoning += reasoning
|
||
all_normal += normal
|
||
|
||
self.assertIn("analyze", all_reasoning)
|
||
self.assertIn("multiple factors", all_reasoning)
|
||
self.assertIn("42", all_normal)
|
||
|
||
def test_empty_reasoning_blocks(self):
|
||
"""Test handling of empty reasoning blocks."""
|
||
parser = ReasoningParser("qwen3")
|
||
text = "<think></think>Just the answer."
|
||
|
||
reasoning, normal = parser.parse_non_stream(text)
|
||
self.assertEqual(reasoning, "")
|
||
self.assertEqual(normal, "Just the answer.")
|
||
|
||
def test_qwen3_forced_reasoning_complete_response(self):
|
||
"""Test complete Qwen3-ForcedReasoning response parsing."""
|
||
parser = ReasoningParser("qwen3", force_reasoning=True)
|
||
text = "Let me solve this step by step. The equation is x + 2 = 5. Subtracting 2 from both sides gives x = 3.</think>The solution is x = 3."
|
||
|
||
reasoning, normal = parser.parse_non_stream(text)
|
||
self.assertIn("step by step", reasoning)
|
||
self.assertIn("x = 3", reasoning)
|
||
self.assertEqual(normal, "The solution is x = 3.")
|
||
|
||
def test_qwen3_forced_reasoning_streaming_scenario(self):
|
||
"""Test Qwen3-ForcedReasoning streaming scenario."""
|
||
parser = ReasoningParser("qwen3", force_reasoning=True)
|
||
|
||
chunks = [
|
||
"I need to analyze",
|
||
" this problem carefully.",
|
||
" Let me break it down.",
|
||
"</think>",
|
||
"The final answer is 42.",
|
||
]
|
||
|
||
all_reasoning = ""
|
||
all_normal = ""
|
||
|
||
for chunk in chunks:
|
||
reasoning, normal = parser.parse_stream_chunk(chunk)
|
||
all_reasoning += reasoning
|
||
all_normal += normal
|
||
|
||
self.assertIn("analyze", all_reasoning)
|
||
self.assertIn("break it down", all_reasoning)
|
||
self.assertIn("final answer", all_normal)
|
||
|
||
|
||
class TestBufferLossBugFix(CustomTestCase):
|
||
"""Test cases for the buffer loss bug fix in parse_streaming_increment."""
|
||
|
||
def test_partial_end_tag_buffer_loss_bug(self):
|
||
"""
|
||
Test the bug where partial end tag fragments are lost when followed by normal text.
|
||
|
||
Bug scenario:
|
||
1. _in_reasoning is False
|
||
2. new_text is "</" (part of closing thinking tag)
|
||
3. Fragment is stored in buffer and empty string is returned
|
||
4. Next step: new_text is "answer", _in_reasoning still False
|
||
5. Buffer is cleared and "answer" is returned directly
|
||
6. The "</" from previous step is lost
|
||
|
||
This test verifies the fix where the return was changed from:
|
||
return StreamingParseResult(normal_text=new_text)
|
||
to:
|
||
return StreamingParseResult(normal_text=current_text)
|
||
"""
|
||
detector = BaseReasoningFormatDetector("<think>", "</think>")
|
||
|
||
# Step 1: Send partial end tag when not in reasoning mode
|
||
# This should be buffered since it could be start of "</think>"
|
||
result1 = detector.parse_streaming_increment("</")
|
||
self.assertEqual(result1.normal_text, "")
|
||
self.assertEqual(result1.reasoning_text, "")
|
||
|
||
# Step 2: Send normal text that doesn't complete the end tag
|
||
# Before fix: would return only "answer", losing the "</"
|
||
# After fix: should return the complete buffered content "</answer"
|
||
result2 = detector.parse_streaming_increment("answer")
|
||
self.assertEqual(result2.normal_text, "</answer")
|
||
self.assertEqual(result2.reasoning_text, "")
|
||
|
||
def test_partial_end_tag_in_reasoning_mode(self):
|
||
"""
|
||
Test partial end tag handling when already in reasoning mode.
|
||
"""
|
||
detector = BaseReasoningFormatDetector("<think>", "</think>")
|
||
|
||
# Enter reasoning mode
|
||
detector.parse_streaming_increment("<think>")
|
||
detector.parse_streaming_increment("some reasoning")
|
||
|
||
# Send partial end tag
|
||
result1 = detector.parse_streaming_increment("</")
|
||
self.assertEqual(result1.normal_text, "")
|
||
self.assertEqual(result1.reasoning_text, "")
|
||
|
||
# Complete the end tag with normal text
|
||
result2 = detector.parse_streaming_increment("think>normal text")
|
||
self.assertEqual(result2.normal_text, "normal text")
|
||
# The reasoning text should be empty since buffer was cleared when end tag was processed
|
||
self.assertEqual(result2.reasoning_text, "")
|
||
|
||
def test_edge_case_exact_token_match(self):
|
||
"""
|
||
Test edge case where buffer content exactly matches a token.
|
||
"""
|
||
detector = BaseReasoningFormatDetector("<think>", "</think>")
|
||
|
||
# Build up the exact start token character by character
|
||
detector.parse_streaming_increment("<")
|
||
detector.parse_streaming_increment("t")
|
||
detector.parse_streaming_increment("h")
|
||
detector.parse_streaming_increment("i")
|
||
detector.parse_streaming_increment("n")
|
||
result = detector.parse_streaming_increment("k>")
|
||
|
||
# Should enter reasoning mode
|
||
self.assertEqual(result.normal_text, "")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
self.assertTrue(detector._in_reasoning)
|
||
self.assertTrue(detector.stripped_think_start)
|
||
|
||
|
||
class TestGptOssDetector(CustomTestCase):
|
||
"""Test cases for GptOssDetector which delegates to HarmonyParser."""
|
||
|
||
def setUp(self):
|
||
from sglang.srt.parser.reasoning_parser import GptOssDetector
|
||
|
||
self.detector = GptOssDetector()
|
||
|
||
def test_streaming_analysis_then_final(self):
|
||
"""Test streaming parse across multiple chunks."""
|
||
chunks = [
|
||
"<|start|><|channel|>analysis<|message|>",
|
||
"reasoning part",
|
||
"<|end|>",
|
||
"<|channel|>final<|message|>answer",
|
||
"<|end|>",
|
||
]
|
||
all_reasoning = ""
|
||
all_normal = ""
|
||
for chunk in chunks:
|
||
result = self.detector.parse_streaming_increment(chunk)
|
||
all_reasoning += result.reasoning_text
|
||
all_normal += result.normal_text
|
||
self.assertIn("reasoning part", all_reasoning)
|
||
self.assertIn("answer", all_normal)
|
||
|
||
|
||
class TestMiniMaxAppendThinkDetector(CustomTestCase):
|
||
"""Test cases for MiniMaxAppendThinkDetector."""
|
||
|
||
def setUp(self):
|
||
from sglang.srt.parser.reasoning_parser import MiniMaxAppendThinkDetector
|
||
|
||
self.detector = MiniMaxAppendThinkDetector()
|
||
|
||
def test_detect_and_parse_prepends_think(self):
|
||
"""Test that detect_and_parse prepends <think> to the text."""
|
||
result = self.detector.detect_and_parse("Hello world")
|
||
self.assertEqual(result.normal_text, "<think>Hello world")
|
||
|
||
def test_streaming_first_chunk_prepends_think(self):
|
||
"""Test that first streaming chunk gets <think> prepended."""
|
||
result = self.detector.parse_streaming_increment("First chunk")
|
||
self.assertEqual(result.normal_text, "<think>First chunk")
|
||
|
||
def test_streaming_second_chunk_no_prepend(self):
|
||
"""Test that subsequent streaming chunks are passed through."""
|
||
self.detector.parse_streaming_increment("First")
|
||
result = self.detector.parse_streaming_increment("Second")
|
||
self.assertEqual(result.normal_text, "Second")
|
||
|
||
|
||
class TestMiniMaxM3Detector(CustomTestCase):
|
||
"""Test cases for MiniMaxM3Detector multi-turn stray-closer handling."""
|
||
|
||
def _detector(self, force_reasoning=False):
|
||
from sglang.srt.parser.reasoning_parser import MiniMaxM3Detector
|
||
|
||
return MiniMaxM3Detector(force_reasoning=force_reasoning)
|
||
|
||
def test_drops_leading_stray_close_non_stream(self):
|
||
"""Non-thinking multi-turn reply opening with a stray </mm:think>."""
|
||
result = self._detector().detect_and_parse("</mm:think>The answer is 42.")
|
||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||
self.assertEqual(result.reasoning_text or "", "")
|
||
|
||
def test_drops_leading_stray_close_with_whitespace(self):
|
||
result = self._detector().detect_and_parse("\n</mm:think>Hello")
|
||
self.assertEqual(result.normal_text, "Hello")
|
||
|
||
def test_plain_reply_untouched(self):
|
||
result = self._detector().detect_and_parse("Just a normal answer.")
|
||
self.assertEqual(result.normal_text, "Just a normal answer.")
|
||
|
||
def test_real_reasoning_block_non_stream(self):
|
||
result = self._detector().detect_and_parse(
|
||
"<mm:think>reasoning here</mm:think>final"
|
||
)
|
||
self.assertEqual(result.reasoning_text, "reasoning here")
|
||
self.assertEqual(result.normal_text, "final")
|
||
|
||
def test_thinking_mode_close_not_dropped(self):
|
||
"""force_reasoning=True means the closer ends reasoning, not a stray drop."""
|
||
result = self._detector(force_reasoning=True).detect_and_parse(
|
||
"reasoning</mm:think>answer"
|
||
)
|
||
self.assertEqual(result.reasoning_text, "reasoning")
|
||
self.assertEqual(result.normal_text, "answer")
|
||
|
||
def test_drops_leading_stray_close_stream_single_token(self):
|
||
detector = self._detector()
|
||
first = detector.parse_streaming_increment("</mm:think>")
|
||
self.assertEqual(first.normal_text or "", "")
|
||
second = detector.parse_streaming_increment("The answer.")
|
||
self.assertEqual(second.normal_text, "The answer.")
|
||
|
||
def test_drops_leading_stray_close_stream_after_whitespace(self):
|
||
"""A leading whitespace token before the atomic </mm:think> is buffered."""
|
||
detector = self._detector()
|
||
self.assertEqual(detector.parse_streaming_increment(" ").normal_text or "", "")
|
||
self.assertEqual(
|
||
detector.parse_streaming_increment("</mm:think>").normal_text or "", ""
|
||
)
|
||
self.assertEqual(
|
||
detector.parse_streaming_increment("Hello").normal_text, "Hello"
|
||
)
|
||
|
||
def test_plain_reply_stream_untouched(self):
|
||
detector = self._detector()
|
||
out = detector.parse_streaming_increment("Hello")
|
||
self.assertEqual(out.normal_text, "Hello")
|
||
|
||
def test_minimax_m3_model_type(self):
|
||
from sglang.srt.parser.reasoning_parser import MiniMaxM3Detector
|
||
|
||
parser = ReasoningParser("minimax-m3")
|
||
self.assertIsInstance(parser.detector, MiniMaxM3Detector)
|
||
|
||
def test_force_nonempty_content_via_chat_template_kwargs(self):
|
||
"""force_nonempty_content must reach the M3 detector without a TypeError."""
|
||
from sglang.srt.entrypoints.openai.protocol import (
|
||
ChatCompletionMessageUserParam,
|
||
ChatCompletionRequest,
|
||
)
|
||
|
||
request = ChatCompletionRequest(
|
||
model="test",
|
||
messages=[ChatCompletionMessageUserParam(role="user", content="Hi")],
|
||
chat_template_kwargs={"force_nonempty_content": True},
|
||
)
|
||
parser = ReasoningParser("minimax-m3", request=request)
|
||
self.assertTrue(parser.detector._force_nonempty_content)
|
||
|
||
def test_force_nonempty_content_swaps_when_no_content(self):
|
||
from sglang.srt.parser.reasoning_parser import MiniMaxM3Detector
|
||
|
||
detector = MiniMaxM3Detector(force_reasoning=True, force_nonempty_content=True)
|
||
result = detector.detect_and_parse("only reasoning, no closer")
|
||
self.assertEqual(result.normal_text, "only reasoning, no closer")
|
||
self.assertEqual(result.reasoning_text or "", "")
|
||
|
||
|
||
class TestReasoningParserAdvanced(CustomTestCase):
|
||
"""Additional tests for ReasoningParser init edge cases."""
|
||
|
||
def test_gpt_oss_model_type(self):
|
||
"""Test that gpt-oss model type creates GptOssDetector."""
|
||
from sglang.srt.parser.reasoning_parser import GptOssDetector
|
||
|
||
parser = ReasoningParser("gpt-oss")
|
||
self.assertIsInstance(parser.detector, GptOssDetector)
|
||
|
||
def test_minimax_append_think_model_type(self):
|
||
"""Test that minimax-append-think creates MiniMaxAppendThinkDetector."""
|
||
from sglang.srt.parser.reasoning_parser import MiniMaxAppendThinkDetector
|
||
|
||
parser = ReasoningParser("minimax-append-think")
|
||
self.assertIsInstance(parser.detector, MiniMaxAppendThinkDetector)
|
||
|
||
def test_qwen3_thinking_forces_reasoning(self):
|
||
"""Test that qwen3-thinking model type forces reasoning mode."""
|
||
parser = ReasoningParser("qwen3-thinking")
|
||
self.assertTrue(parser.detector._in_reasoning)
|
||
|
||
def test_minimax_forces_reasoning(self):
|
||
"""Test that minimax model type forces reasoning mode.
|
||
|
||
minimax maps to Qwen3Detector but ReasoningParser overrides
|
||
force_reasoning=True, unlike the default Qwen3Detector behavior.
|
||
"""
|
||
parser = ReasoningParser("minimax")
|
||
self.assertIsInstance(parser.detector, Qwen3Detector)
|
||
self.assertTrue(parser.detector._in_reasoning)
|
||
|
||
def test_detector_map_aliases(self):
|
||
"""Test that all DetectorMap alias keys create the correct detector type."""
|
||
# These are aliases that map to existing detector classes
|
||
alias_tests = {
|
||
"deepseek-v3": Qwen3Detector,
|
||
"step3": DeepSeekR1Detector,
|
||
"step3p5": DeepSeekR1Detector,
|
||
"interns1": Qwen3Detector,
|
||
}
|
||
for model_type, expected_class in alias_tests.items():
|
||
parser = ReasoningParser(model_type)
|
||
self.assertIsInstance(
|
||
parser.detector,
|
||
expected_class,
|
||
f"{model_type} should create {expected_class.__name__}",
|
||
)
|
||
|
||
def test_continue_final_message_with_request(self):
|
||
"""Test continue_final_message passes previous content to detector."""
|
||
from sglang.srt.entrypoints.openai.protocol import (
|
||
ChatCompletionMessageGenericParam,
|
||
ChatCompletionMessageUserParam,
|
||
ChatCompletionRequest,
|
||
)
|
||
|
||
request = ChatCompletionRequest(
|
||
model="test",
|
||
messages=[
|
||
ChatCompletionMessageUserParam(role="user", content="Hi"),
|
||
ChatCompletionMessageGenericParam(
|
||
role="assistant", content="Let me think..."
|
||
),
|
||
],
|
||
continue_final_message=True,
|
||
)
|
||
parser = ReasoningParser("qwen3", request=request)
|
||
self.assertTrue(parser.detector.continue_final_message)
|
||
|
||
def test_force_nonempty_content_via_chat_template_kwargs(self):
|
||
"""Test that force_nonempty_content is passed via chat_template_kwargs."""
|
||
from sglang.srt.entrypoints.openai.protocol import (
|
||
ChatCompletionMessageUserParam,
|
||
ChatCompletionRequest,
|
||
)
|
||
|
||
request = ChatCompletionRequest(
|
||
model="test",
|
||
messages=[
|
||
ChatCompletionMessageUserParam(role="user", content="Hi"),
|
||
],
|
||
chat_template_kwargs={"force_nonempty_content": True},
|
||
)
|
||
parser = ReasoningParser("nemotron_3", request=request)
|
||
self.assertTrue(parser.detector._force_nonempty_content)
|
||
|
||
|
||
class TestContinueFinalMessage(CustomTestCase):
|
||
"""Test continue_final_message mode for BaseReasoningFormatDetector."""
|
||
|
||
def test_continue_with_think_start_in_previous(self):
|
||
"""Test that previous_content with <think> sets _in_reasoning=True."""
|
||
detector = BaseReasoningFormatDetector(
|
||
"<think>",
|
||
"</think>",
|
||
force_reasoning=False,
|
||
continue_final_message=True,
|
||
previous_content="<think>some reasoning",
|
||
)
|
||
self.assertTrue(detector._in_reasoning)
|
||
self.assertEqual(detector.previous_count, len("<think>some reasoning"))
|
||
|
||
def test_continue_with_think_end_in_previous(self):
|
||
"""Test that previous_content with </think> sets _in_reasoning=False."""
|
||
detector = BaseReasoningFormatDetector(
|
||
"<think>",
|
||
"</think>",
|
||
force_reasoning=True,
|
||
continue_final_message=True,
|
||
previous_content="<think>done</think>normal",
|
||
)
|
||
# think_end_token in previous → _in_reasoning = False
|
||
self.assertFalse(detector._in_reasoning)
|
||
|
||
def test_continue_detect_parse_with_end_in_previous(self):
|
||
"""Test detect_and_parse when think_end_token is in previous_content only.
|
||
This covers the branch where think_end is NOT in current text
|
||
but IS in previous_content, so output is treated as normal_text."""
|
||
detector = BaseReasoningFormatDetector(
|
||
"<think>",
|
||
"</think>",
|
||
force_reasoning=True,
|
||
continue_final_message=True,
|
||
previous_content="<think>reasoning</think>",
|
||
)
|
||
# _in_reasoning is False (think_end in previous)
|
||
# But force_reasoning was True → detect_and_parse still enters the
|
||
# reasoning path because think_start is in previous_content.
|
||
# However, since _in_reasoning=False and no think_start in new text,
|
||
# it returns normal_text directly.
|
||
result = detector.detect_and_parse("new content here")
|
||
self.assertEqual(result.normal_text, "new content here")
|
||
|
||
def test_continue_end_in_previous_new_text_has_start_but_no_end(self):
|
||
"""Test: think_end in previous, new text has think_start but no think_end.
|
||
This produces: in_reasoning=True (from think_start in text),
|
||
think_end NOT in processed_text, think_end IN previous_content,
|
||
so it falls through to the else branch that returns normal_text."""
|
||
detector = BaseReasoningFormatDetector(
|
||
"<think>",
|
||
"</think>",
|
||
force_reasoning=False,
|
||
continue_final_message=True,
|
||
previous_content="earlier <think>old</think>old answer",
|
||
)
|
||
# _in_reasoning = False (think_end in previous overrides)
|
||
self.assertFalse(detector._in_reasoning)
|
||
# New text has <think> (triggers in_reasoning) but no </think>
|
||
# think_end IS in previous_content → skips the truncated-reasoning branch
|
||
# think_end NOT in processed_text → falls to else that returns normal_text
|
||
result = detector.detect_and_parse("<think>continuing reasoning")
|
||
self.assertEqual(result.normal_text, "continuing reasoning")
|
||
self.assertEqual(result.reasoning_text, "")
|
||
|
||
def test_continue_detect_parse_think_start_in_prev_but_end_also_in_prev(self):
|
||
"""Test detect_and_parse where both think tokens are in previous,
|
||
and new text contains <think> to re-enter reasoning."""
|
||
detector = BaseReasoningFormatDetector(
|
||
"<think>",
|
||
"</think>",
|
||
force_reasoning=False,
|
||
continue_final_message=True,
|
||
previous_content="<think>old reasoning</think>old answer",
|
||
)
|
||
# _in_reasoning = False (end token in previous overrides start)
|
||
self.assertFalse(detector._in_reasoning)
|
||
# New text starts a fresh reasoning block
|
||
result = detector.detect_and_parse("<think>new reasoning</think>new answer")
|
||
self.assertEqual(result.reasoning_text, "new reasoning")
|
||
self.assertEqual(result.normal_text, "new answer")
|
||
|
||
|
||
class TestGptOssDetectorToolCall(CustomTestCase):
|
||
"""Test GptOssDetector tool_call raw_text handling."""
|
||
|
||
def test_detect_and_parse_tool_call_raw_text(self):
|
||
"""Test that tool_call events use raw_text when available."""
|
||
from sglang.srt.parser.reasoning_parser import GptOssDetector
|
||
|
||
detector = GptOssDetector()
|
||
# Sequence with CALL...RETURN that produces tool_call events with raw_text
|
||
text = (
|
||
"<|start|><|channel|>analysis<|message|>think<|end|>"
|
||
"<|call|>function_data<|return|>"
|
||
"<|channel|>final<|message|>result<|end|>"
|
||
)
|
||
result = detector.detect_and_parse(text)
|
||
self.assertIn("think", result.reasoning_text)
|
||
# Tool call raw_text and/or final result should be in normal_text
|
||
self.assertIn("result", result.normal_text)
|
||
|
||
def test_streaming_tool_call_raw_text(self):
|
||
"""Test streaming parse with tool_call events preserving raw_text."""
|
||
from sglang.srt.parser.reasoning_parser import GptOssDetector
|
||
|
||
detector = GptOssDetector()
|
||
chunks = [
|
||
"<|start|><|channel|>analysis<|message|>reason<|end|>",
|
||
"<|call|>tool_payload<|return|>",
|
||
"<|channel|>final<|message|>done<|end|>",
|
||
]
|
||
all_reasoning = ""
|
||
all_normal = ""
|
||
for chunk in chunks:
|
||
result = detector.parse_streaming_increment(chunk)
|
||
all_reasoning += result.reasoning_text
|
||
all_normal += result.normal_text
|
||
self.assertIn("reason", all_reasoning)
|
||
self.assertIn("done", all_normal)
|
||
|
||
|
||
class TestPoolsideV1Registered(CustomTestCase):
|
||
"""poolside_v1 (Laguna-XS.2) reuses the Qwen3 `<think>...</think>` envelope.
|
||
Request dispatch differs (Mimo-style explicit `enable_thinking=True`,
|
||
asserted in test_serving_chat.py), driven by
|
||
`reasoning_default = "explicit_enable_thinking"` on the detector."""
|
||
|
||
def test_registered_to_qwen3_subclass(self):
|
||
cls = ReasoningParser.DetectorMap["poolside_v1"]
|
||
self.assertTrue(issubclass(cls, Qwen3Detector))
|
||
|
||
def test_explicit_enable_thinking_default(self):
|
||
rp = ReasoningParser("poolside_v1", stream_reasoning=True)
|
||
self.assertEqual(rp.detector.reasoning_default, "explicit_enable_thinking")
|
||
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()
|