forked from waybarrios/vllm-mlx
-
Notifications
You must be signed in to change notification settings - Fork 67
Expand file tree
/
Copy pathtest_reasoning_parsers.py
More file actions
984 lines (805 loc) · 35.3 KB
/
test_reasoning_parsers.py
File metadata and controls
984 lines (805 loc) · 35.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
# SPDX-License-Identifier: Apache-2.0
"""Tests for reasoning parsers (base, think_parser, deepseek_r1, gpt_oss)."""
import pytest
from vllm_mlx.reasoning.base import DeltaMessage, ReasoningParser
from vllm_mlx.reasoning.deepseek_r1_parser import DeepSeekR1ReasoningParser
from vllm_mlx.reasoning.gpt_oss_parser import (
_CHANNEL_RE,
_STRUCTURAL_TOKENS,
GptOssReasoningParser,
_extract_channel,
)
from vllm_mlx.reasoning.harmony_parser import HarmonyReasoningParser
from vllm_mlx.reasoning.minimax_parser import MiniMaxReasoningParser
from vllm_mlx.reasoning.qwen3_parser import Qwen3ReasoningParser
# ---------------------------------------------------------------------------
# DeltaMessage
# ---------------------------------------------------------------------------
class TestDeltaMessage:
def test_reasoning_only(self):
dm = DeltaMessage(reasoning="thinking")
assert dm.reasoning == "thinking"
assert dm.content is None
def test_content_only(self):
dm = DeltaMessage(content="answer")
assert dm.content == "answer"
assert dm.reasoning is None
def test_both(self):
dm = DeltaMessage(reasoning="r", content="c")
assert dm.reasoning == "r"
assert dm.content == "c"
def test_reasoning_content_alias(self):
dm = DeltaMessage(reasoning="r")
assert dm.reasoning_content == "r"
def test_defaults(self):
dm = DeltaMessage()
assert dm.role is None
assert dm.content is None
assert dm.reasoning is None
# ---------------------------------------------------------------------------
# ReasoningParser (abstract base)
# ---------------------------------------------------------------------------
class TestReasoningParserBase:
def test_cannot_instantiate(self):
with pytest.raises(TypeError):
ReasoningParser()
def test_reset_state_noop(self):
class Dummy(ReasoningParser):
def extract_reasoning(self, model_output):
return None, model_output
def extract_reasoning_streaming(self, prev, curr, delta):
return None
d = Dummy()
d.reset_state() # should not raise
def test_finalize_streaming_default_none(self):
class Dummy(ReasoningParser):
def extract_reasoning(self, model_output):
return None, model_output
def extract_reasoning_streaming(self, prev, curr, delta):
return None
d = Dummy()
assert d.finalize_streaming("some text") is None
# ---------------------------------------------------------------------------
# BaseThinkingReasoningParser (via DeepSeek-R1 as concrete subclass)
# ---------------------------------------------------------------------------
class TestBaseThinkExtractReasoning:
"""Tests for extract_reasoning using DeepSeekR1ReasoningParser."""
def setup_method(self):
self.parser = DeepSeekR1ReasoningParser()
def test_both_tags(self):
text = "<think>step by step</think>The answer is 42."
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "step by step"
assert content == "The answer is 42."
def test_both_tags_empty_reasoning(self):
text = "<think></think>Just content"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == "Just content"
def test_both_tags_empty_content(self):
text = "<think>reasoning only</think>"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "reasoning only"
assert content is None
def test_both_tags_whitespace_reasoning(self):
text = "<think> </think>content"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == "content"
def test_only_end_tag_implicit(self):
text = "implicit reasoning</think>final answer"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "implicit reasoning"
assert content == "final answer"
def test_only_start_tag(self):
text = "<think>incomplete reasoning without close"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "incomplete reasoning without close"
assert content is None
def test_no_tags_pure_content(self):
text = "Just a simple response with no thinking."
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == text
def test_multiline_reasoning(self):
text = "<think>Line 1\nLine 2\nLine 3</think>Answer"
reasoning, content = self.parser.extract_reasoning(text)
assert "Line 1" in reasoning
assert "Line 3" in reasoning
assert content == "Answer"
def test_multiple_think_tags_uses_first(self):
text = "<think>first</think>middle<think>second</think>end"
reasoning, content = self.parser.extract_reasoning(text)
# partition finds first occurrence
assert reasoning == "first"
assert "middle" in content
# ---------------------------------------------------------------------------
# BaseThinkingReasoningParser streaming
# ---------------------------------------------------------------------------
class TestBaseThinkStreaming:
def setup_method(self):
self.parser = DeepSeekR1ReasoningParser()
self.parser.reset_state()
def test_skip_start_token(self):
result = self.parser.extract_reasoning_streaming("", "<think>", "<think>")
assert result is None
def test_skip_end_token(self):
result = self.parser.extract_reasoning_streaming(
"<think>reasoning", "<think>reasoning</think>", "</think>"
)
assert result is None
def test_reasoning_after_start(self):
prev = "<think>"
delta = "step 1"
curr = prev + delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result.reasoning == "step 1"
assert result.content is None
def test_content_after_end(self):
prev = "<think>reasoning</think>"
delta = "content"
curr = prev + delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result.content == "content"
assert result.reasoning is None
def test_transition_in_delta(self):
prev = "<think>reasoning"
delta = " more</think>content"
curr = prev + delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result.reasoning == " more"
assert result.content == "content"
def test_both_tags_in_single_delta(self):
prev = ""
delta = "<think>reason</think>content"
curr = delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result.reasoning == "reason"
assert result.content == "content"
def test_start_tag_only_in_delta(self):
prev = ""
delta = "<think>beginning"
curr = delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result.reasoning == "beginning"
def test_no_tags_early_defaults_to_reasoning(self):
"""Before any tags seen, base class defaults to reasoning."""
prev = ""
delta = "hello"
curr = "hello"
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
# DeepSeek has threshold logic, but under threshold defaults to reasoning
assert result.reasoning == "hello" or result.content == "hello"
def test_implicit_end_only(self):
"""Implicit mode: </think> without <think>."""
prev = "some reasoning"
delta = "</think>answer"
curr = prev + delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
# Should transition from reasoning to content
assert result is not None
def test_reset_state(self):
self.parser._saw_any_tag = True
self.parser.reset_state()
assert self.parser._saw_any_tag is False
# ---------------------------------------------------------------------------
# DeepSeekR1ReasoningParser specifics
# ---------------------------------------------------------------------------
class TestDeepSeekR1:
def setup_method(self):
self.parser = DeepSeekR1ReasoningParser()
def test_tokens(self):
assert self.parser.start_token == "<think>"
assert self.parser.end_token == "</think>"
def test_no_tag_threshold_constant(self):
assert self.parser.NO_TAG_CONTENT_THRESHOLD == 64
def test_no_start_only_end(self):
"""DeepSeek-R1 handles implicit start tag."""
text = "thinking about it</think>42"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "thinking about it"
assert content == "42"
def test_no_tags_returns_content(self):
text = "direct answer"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == "direct answer"
def test_standard_both_tags(self):
text = "<think>r</think>c"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "r"
assert content == "c"
def test_streaming_no_tag_past_threshold(self):
"""After threshold chars without tags, treat as content."""
self.parser.reset_state()
long_text = "x" * 100
result = self.parser.extract_reasoning_streaming("", long_text, long_text)
assert result.content == long_text
def test_streaming_no_tag_under_threshold(self):
"""Under threshold without tags, delegates to base (reasoning)."""
self.parser.reset_state()
short = "hi"
result = self.parser.extract_reasoning_streaming("", short, short)
assert result.reasoning == short
def test_finalize_short_no_tag_correction(self):
"""Short output without tags gets corrected from reasoning to content."""
self.parser.reset_state()
self.parser._saw_any_tag = False
result = self.parser.finalize_streaming("short answer")
assert result is not None
assert result.content == "short answer"
def test_finalize_long_no_tag_no_correction(self):
"""Long output without tags: no correction (already handled by threshold)."""
self.parser.reset_state()
self.parser._saw_any_tag = False
result = self.parser.finalize_streaming("x" * 100)
assert result is None
def test_finalize_with_tags_no_correction(self):
"""Output with tags: no correction needed."""
self.parser.reset_state()
self.parser._saw_any_tag = True
result = self.parser.finalize_streaming("<think>r</think>c")
assert result is None
def test_finalize_empty_no_correction(self):
self.parser.reset_state()
result = self.parser.finalize_streaming("")
assert result is None
# ---------------------------------------------------------------------------
# GptOssReasoningParser
# ---------------------------------------------------------------------------
class TestGptOssHelpers:
def test_extract_channel_analysis(self):
text = "<|channel|>analysis<|message|>my reasoning<|start|>assistant"
result = _extract_channel(text, "analysis")
assert result == "my reasoning"
def test_extract_channel_final(self):
text = "<|channel|>final<|message|>the answer<|return|>"
result = _extract_channel(text, "final")
assert result == "the answer"
def test_extract_channel_not_found(self):
text = "<|channel|>analysis<|message|>reasoning"
result = _extract_channel(text, "final")
assert result is None
def test_extract_channel_empty_content(self):
text = "<|channel|>analysis<|message|><|start|>"
result = _extract_channel(text, "analysis")
assert result is None
def test_extract_channel_with_constrain(self):
text = "<|channel|>final <|constrain|>JSON<|message|>content here<|return|>"
result = _extract_channel(text, "final")
assert result == "content here"
def test_channel_regex_matches_analysis(self):
text = "<|channel|>analysis<|message|>"
m = _CHANNEL_RE.search(text)
assert m is not None
assert m.group(1) == "analysis"
def test_channel_regex_matches_final(self):
text = "<|channel|>final<|message|>"
m = _CHANNEL_RE.search(text)
assert m is not None
assert m.group(1) == "final"
def test_channel_regex_matches_constrain(self):
text = "<|channel|>final <|constrain|>JSON<|message|>"
m = _CHANNEL_RE.search(text)
assert m is not None
assert m.group(1) == "final"
def test_structural_tokens_regex(self):
for tok in [
"<|start|>",
"<|end|>",
"<|channel|>",
"<|return|>",
"<|call|>",
"<|constrain|>",
]:
assert _STRUCTURAL_TOKENS.search(tok) is not None
class TestGptOssExtractReasoning:
def setup_method(self):
self.parser = GptOssReasoningParser()
def test_full_format(self):
text = (
"<|channel|>analysis<|message|>Step by step reasoning"
"<|start|>assistant<|channel|>final<|message|>The answer is 42<|return|>"
)
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "Step by step reasoning"
assert content == "The answer is 42"
def test_analysis_only(self):
text = "<|channel|>analysis<|message|>just reasoning"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "just reasoning"
assert content is None
def test_final_only(self):
text = "<|channel|>final<|message|>just content<|return|>"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == "just content"
def test_no_channels(self):
text = "plain text without channels"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == text
def test_empty_input(self):
reasoning, content = self.parser.extract_reasoning("")
assert reasoning is None
assert content is None
def test_none_like_empty(self):
reasoning, content = self.parser.extract_reasoning("")
assert reasoning is None
def test_constrain_format(self):
text = (
"<|channel|>analysis<|message|>thinking"
'<|start|>assistant<|channel|>final <|constrain|>JSON<|message|>{"key": "val"}<|return|>'
)
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "thinking"
assert content == '{"key": "val"}'
def test_structural_tokens_stripped(self):
text = (
"<|channel|>analysis<|message|>reason<|start|>"
"<|channel|>final<|message|>answer<|return|>"
)
reasoning, content = self.parser.extract_reasoning(text)
assert "<|" not in (reasoning or "")
assert "<|" not in (content or "")
class TestGptOssStreaming:
def setup_method(self):
self.parser = GptOssReasoningParser()
def test_detect_phase_init(self):
assert GptOssReasoningParser._detect_phase("") == "init"
assert GptOssReasoningParser._detect_phase("random text") == "init"
def test_detect_phase_analysis(self):
text = "<|channel|>analysis<|message|>reasoning"
assert GptOssReasoningParser._detect_phase(text) == "analysis"
def test_detect_phase_final(self):
text = "<|channel|>analysis<|message|>r<|start|>assistant<|channel|>final<|message|>c"
assert GptOssReasoningParser._detect_phase(text) == "final"
def test_detect_phase_transition(self):
text = "<|channel|>analysis<|message|>reason<|start|>"
assert GptOssReasoningParser._detect_phase(text) == "transition"
def test_streaming_analysis_phase(self):
prev = "<|channel|>analysis<|message|>part1"
delta = " part2"
curr = prev + delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result is not None
assert result.reasoning == " part2"
def test_streaming_final_phase(self):
prev = "<|channel|>analysis<|message|>r<|start|>assistant<|channel|>final<|message|>part1"
delta = " part2"
curr = prev + delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result is not None
assert result.content == " part2"
def test_streaming_phase_transition_to_analysis(self):
prev = ""
delta = "<|channel|>analysis<|message|>reasoning start"
curr = delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result is not None
assert result.reasoning is not None
assert "reasoning start" in result.reasoning
def test_streaming_phase_transition_to_final(self):
prev = "<|channel|>analysis<|message|>reason<|start|>assistant"
delta = "<|channel|>final<|message|>content start"
curr = prev + delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result is not None
assert result.content is not None
assert "content start" in result.content
def test_streaming_init_phase_skips(self):
prev = ""
delta = "<|start|>"
curr = delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
assert result is None
def test_streaming_structural_token_stripped(self):
prev = "<|channel|>analysis<|message|>reasoning"
delta = "<|start|>"
curr = prev + delta
result = self.parser.extract_reasoning_streaming(prev, curr, delta)
# Phase transitions to "transition", delta is structural → skip
assert result is None or (
result and "<|start|>" not in (result.reasoning or "")
)
def test_strip_return(self):
assert GptOssReasoningParser._strip_return("text<|return|>") == "text"
assert GptOssReasoningParser._strip_return("no return") == "no return"
def test_extract_content_after_marker(self):
text = "<|channel|>analysis<|message|>the content"
result = GptOssReasoningParser._extract_content_after_marker_in_delta(
text, "analysis"
)
assert result == "the content"
def test_extract_content_after_marker_not_found(self):
text = "<|channel|>analysis<|message|>content"
result = GptOssReasoningParser._extract_content_after_marker_in_delta(
text, "final"
)
assert result is None
# ---------------------------------------------------------------------------
# Full streaming simulation tests
# ---------------------------------------------------------------------------
class TestFullStreamingSimulation:
"""Simulate realistic streaming token-by-token delivery."""
def test_think_parser_full_stream(self):
"""Simulate: <think>step 1\nstep 2</think>The answer."""
parser = DeepSeekR1ReasoningParser()
parser.reset_state()
chunks = ["<think>", "step ", "1\n", "step 2", "</think>", "The ", "answer."]
accumulated = ""
reasoning_parts = []
content_parts = []
for chunk in chunks:
prev = accumulated
accumulated += chunk
result = parser.extract_reasoning_streaming(prev, accumulated, chunk)
if result:
if result.reasoning:
reasoning_parts.append(result.reasoning)
if result.content:
content_parts.append(result.content)
assert "".join(reasoning_parts) == "step 1\nstep 2"
assert "".join(content_parts) == "The answer."
def test_deepseek_implicit_stream(self):
"""Simulate implicit mode: reasoning</think>content (no <think>)."""
parser = DeepSeekR1ReasoningParser()
parser.reset_state()
chunks = ["reas", "oning", "</think>", "content"]
accumulated = ""
reasoning_parts = []
content_parts = []
for chunk in chunks:
prev = accumulated
accumulated += chunk
result = parser.extract_reasoning_streaming(prev, accumulated, chunk)
if result:
if result.reasoning:
reasoning_parts.append(result.reasoning)
if result.content:
content_parts.append(result.content)
assert "reas" in "".join(reasoning_parts)
assert "content" in "".join(content_parts)
def test_gpt_oss_full_stream(self):
"""Simulate GPT-OSS channel-based streaming."""
parser = GptOssReasoningParser()
chunks = [
"<|channel|>analysis<|message|>",
"reasoning ",
"here",
"<|start|>",
"assistant",
"<|channel|>final<|message|>",
"the ",
"answer",
"<|return|>",
]
accumulated = ""
reasoning_parts = []
content_parts = []
for chunk in chunks:
prev = accumulated
accumulated += chunk
result = parser.extract_reasoning_streaming(prev, accumulated, chunk)
if result:
if result.reasoning:
reasoning_parts.append(result.reasoning)
if result.content:
content_parts.append(result.content)
reasoning_text = "".join(reasoning_parts)
content_text = "".join(content_parts)
assert "reasoning" in reasoning_text
assert "answer" in content_text
# ---------------------------------------------------------------------------
# Qwen3ReasoningParser
# ---------------------------------------------------------------------------
class TestQwen3:
def setup_method(self):
self.parser = Qwen3ReasoningParser()
def test_tokens(self):
assert self.parser.start_token == "<think>"
assert self.parser.end_token == "</think>"
def test_both_tags(self):
reasoning, content = self.parser.extract_reasoning(
"<think>analysis</think>answer"
)
assert reasoning == "analysis"
assert content == "answer"
def test_only_end_tag(self):
reasoning, content = self.parser.extract_reasoning(
"implicit reasoning</think>answer"
)
assert reasoning == "implicit reasoning"
assert content == "answer"
def test_no_end_tag_pure_content(self):
"""Qwen3 overrides: if no end token at all, return as content."""
reasoning, content = self.parser.extract_reasoning("just content")
assert reasoning is None
assert content == "just content"
def test_only_start_tag_no_end(self):
"""Start tag without end tag: Qwen3 checks end_token first → pure content."""
reasoning, content = self.parser.extract_reasoning("<think>incomplete")
assert reasoning is None
assert content == "<think>incomplete"
def test_empty_tags(self):
reasoning, content = self.parser.extract_reasoning("<think></think>content")
assert reasoning is None
assert content == "content"
# ---------------------------------------------------------------------------
# MiniMaxReasoningParser
# ---------------------------------------------------------------------------
class TestMiniMaxExtractReasoning:
def setup_method(self):
self.parser = MiniMaxReasoningParser()
def test_direct_content_code_block(self):
text = "```python\nprint('hello')\n```"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == text
def test_direct_content_json(self):
text = '{"key": "value"}'
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == text
def test_direct_content_tool_call(self):
text = "<minimax:tool_call>some tool call"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == text
def test_reasoning_pattern_english(self):
text = "The user asks about Python.\n\nHere is the answer: Python is great."
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is not None
assert "user asks" in reasoning
assert content is not None
def test_reasoning_pattern_i_need(self):
text = "I need to analyze this code.\n\nThe answer is 42."
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is not None
assert content is not None
assert "answer" in content.lower()
def test_reasoning_pattern_let_me(self):
text = "Let me think about this.\n\nHere is the solution."
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is not None
assert content is not None
def test_reasoning_pattern_chinese(self):
text = "用户想知道Python怎么用。\n\n以下是答案。"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is not None
def test_no_reasoning_pattern(self):
text = "Python is a great language for beginners."
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == text
def test_explicit_think_tags(self):
text = "<think>reasoning</think>content"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "reasoning"
assert content == "content"
def test_short_reasoning_not_stripped(self):
"""Very short 'reasoning' (<10 chars) treated as false positive."""
text = "The user\n\nanswer"
reasoning, content = self.parser.extract_reasoning(text)
# "The user" is < 10 chars reasoning → returned as pure content
assert content is not None
def test_double_newline_split(self):
text = "The user asks a question about Python.\n\nPython was created by Guido."
reasoning, content = self.parser.extract_reasoning(text)
# First part matches reasoning pattern, double newline splits
assert reasoning is not None or content is not None
class TestMiniMaxStreaming:
def setup_method(self):
self.parser = MiniMaxReasoningParser()
self.parser.reset_state()
def test_reset_state(self):
self.parser._decided = True
self.parser._buffer = "stuff"
self.parser.reset_state()
assert self.parser._decided is False
assert self.parser._buffer == ""
assert self.parser._is_reasoning is False
def test_explicit_think_tag_in_delta(self):
result = self.parser.extract_reasoning_streaming("", "<think>", "<think>")
assert result is None # tag stripped, nothing left
def test_explicit_think_tag_with_content(self):
result = self.parser.extract_reasoning_streaming(
"", "<think>reasoning", "<think>reasoning"
)
assert result.reasoning == "reasoning"
def test_end_think_tag_transition(self):
self.parser._decided = True
self.parser._is_reasoning = True
result = self.parser.extract_reasoning_streaming(
"thinking", "thinking</think>answer", "</think>answer"
)
assert result.content == "answer"
def test_buffering_phase(self):
"""Short text should be buffered (returns None)."""
result = self.parser.extract_reasoning_streaming("", "hi", "hi")
assert result is None
def test_direct_content_detected_early(self):
"""Code blocks detected immediately as content."""
result = self.parser.extract_reasoning_streaming(
"", "```python\n", "```python\n"
)
assert result is not None
assert result.content is not None
def test_content_phase_passthrough(self):
self.parser._decided = True
self.parser._is_reasoning = False
result = self.parser.extract_reasoning_streaming("prev", "prev more", " more")
assert result.content == " more"
def test_finalize_undecided(self):
self.parser._decided = False
result = self.parser.finalize_streaming("some short text")
assert result is not None
assert result.content == "some short text"
def test_finalize_undecided_empty(self):
self.parser._decided = False
result = self.parser.finalize_streaming("")
assert result is None
def test_finalize_content_phase(self):
self.parser._decided = True
self.parser._is_reasoning = False
result = self.parser.finalize_streaming("content")
assert result is None
def test_finalize_reasoning_reclassifies(self):
self.parser._decided = True
self.parser._is_reasoning = True
result = self.parser.finalize_streaming("Just a simple answer")
assert result is not None
assert result.content is not None
# ---------------------------------------------------------------------------
# HarmonyReasoningParser
# ---------------------------------------------------------------------------
class TestHarmonyExtractReasoning:
def setup_method(self):
self.parser = HarmonyReasoningParser()
def test_full_format(self):
text = (
"<|channel|>analysis<|message|>My reasoning here<|end|>"
"<|channel|>final<|message|>The answer<|return|>"
)
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "My reasoning here"
assert content == "The answer"
def test_analysis_only(self):
text = "<|channel|>analysis<|message|>reasoning only<|end|>"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning == "reasoning only"
assert content is None
def test_final_only(self):
text = "<|channel|>final<|message|>answer only<|return|>"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content == "answer only"
def test_no_channels(self):
text = "plain text"
reasoning, content = self.parser.extract_reasoning(text)
assert reasoning is None
assert content is None
def test_multiple_analysis_blocks(self):
text = (
"<|channel|>analysis<|message|>Block 1<|end|>"
"<|channel|>analysis<|message|>Block 2<|end|>"
"<|channel|>final<|message|>Answer<|return|>"
)
reasoning, content = self.parser.extract_reasoning(text)
assert "Block 1" in reasoning
assert "Block 2" in reasoning
assert content == "Answer"
class TestHarmonyStreaming:
def setup_method(self):
self.parser = HarmonyReasoningParser()
self.parser.reset_state()
def test_reset_state(self):
self.parser._current_channel = "analysis"
self.parser._in_message = True
self.parser.reset_state()
assert self.parser._current_channel is None
assert self.parser._in_message is False
def test_analysis_channel_switch(self):
result = self.parser.extract_reasoning_streaming(
"", "<|channel|>analysis", "<|channel|>analysis"
)
assert result is None
assert self.parser._current_channel == "analysis"
def test_final_channel_switch(self):
result = self.parser.extract_reasoning_streaming(
"", "<|channel|>final", "<|channel|>final"
)
assert result is None
assert self.parser._current_channel == "final"
def test_commentary_channel_switch(self):
result = self.parser.extract_reasoning_streaming(
"", "<|channel|>commentary", "<|channel|>commentary"
)
# Commentary passes through as content for tool parser
assert result is not None
assert result.content == "<|channel|>commentary"
assert self.parser._current_channel == "commentary"
def test_message_start_skipped(self):
self.parser._current_channel = "analysis"
result = self.parser.extract_reasoning_streaming(
"<|channel|>analysis", "<|channel|>analysis<|message|>", "<|message|>"
)
assert result is None
assert self.parser._in_message is True
def test_analysis_content_emitted(self):
self.parser._current_channel = "analysis"
self.parser._in_message = True
result = self.parser.extract_reasoning_streaming(
"<|channel|>analysis<|message|>",
"<|channel|>analysis<|message|>reasoning",
"reasoning",
)
assert result.reasoning == "reasoning"
def test_final_content_emitted(self):
self.parser._current_channel = "final"
self.parser._in_message = True
result = self.parser.extract_reasoning_streaming(
"<|channel|>final<|message|>", "<|channel|>final<|message|>answer", "answer"
)
assert result.content == "answer"
def test_end_token_stops_message(self):
self.parser._current_channel = "analysis"
self.parser._in_message = True
result = self.parser.extract_reasoning_streaming(
"<|channel|>analysis<|message|>r",
"<|channel|>analysis<|message|>r<|end|>",
"<|end|>",
)
assert result is None
assert self.parser._in_message is False
def test_return_token_stops_message(self):
self.parser._current_channel = "final"
self.parser._in_message = True
result = self.parser.extract_reasoning_streaming(
"<|channel|>final<|message|>c",
"<|channel|>final<|message|>c<|return|>",
"<|return|>",
)
assert result is None
assert self.parser._in_message is False
def test_commentary_passed_through(self):
self.parser._current_channel = "commentary"
self.parser._in_message = True
result = self.parser.extract_reasoning_streaming(
"prev", "prev tool_call", " tool_call"
)
# Commentary passes through as content for tool parser
assert result is not None
assert result.content == " tool_call"
def test_control_tokens_skipped(self):
result = self.parser.extract_reasoning_streaming("", "<|start|>", "<|start|>")
assert result is None
def test_full_streaming_simulation(self):
parser = HarmonyReasoningParser()
parser.reset_state()
chunks = [
"<|channel|>analysis",
"<|message|>",
"thinking ",
"step 1",
"<|end|>",
"<|channel|>final",
"<|message|>",
"the ",
"answer",
"<|return|>",
]
accumulated = ""
reasoning_parts = []
content_parts = []
for chunk in chunks:
prev = accumulated
accumulated += chunk
result = parser.extract_reasoning_streaming(prev, accumulated, chunk)
if result:
if result.reasoning:
reasoning_parts.append(result.reasoning)
if result.content:
content_parts.append(result.content)
assert "thinking" in "".join(reasoning_parts)
assert "answer" in "".join(content_parts)