mirror of
https://github.com/PaddlePaddle/FastDeploy.git
synced 2025-11-01 20:32:52 +08:00
175391389f
* Initial plan * Add comprehensive unit tests for limit_thinking_content_length functions Co-authored-by: yuanlehome <23653004+yuanlehome@users.noreply.github.com> * fix (#4514) --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: yuanlehome <23653004+yuanlehome@users.noreply.github.com> Co-authored-by: Yuanle Liu <yuanlehome@163.com>
331 lines
14 KiB
Python
331 lines
14 KiB
Python
# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Unit tests for limit_thinking_content_length_v1 and limit_thinking_content_length_v2"""
|
|
|
|
import unittest
|
|
|
|
import paddle
|
|
|
|
from fastdeploy.model_executor.ops.gpu import (
|
|
limit_thinking_content_length_v1,
|
|
limit_thinking_content_length_v2,
|
|
)
|
|
|
|
|
|
class TestLimitThinkingContentLengthV1(unittest.TestCase):
|
|
"""Tests for limit_thinking_content_length_v1 operator (</think> strategy)"""
|
|
|
|
def test_normal_thinking_phase_no_limit_reached(self):
|
|
"""Test normal thinking phase when step < max_think_len"""
|
|
next_tokens = paddle.to_tensor([[100], [200]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([10, 15], dtype="int32")
|
|
step_idx = paddle.to_tensor([[5], [8]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([0, 0], dtype="int32")
|
|
think_end_id = 999
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v1(next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id)
|
|
|
|
# Verify: tokens unchanged, status unchanged
|
|
assert next_tokens.numpy()[0, 0] == 100
|
|
assert next_tokens.numpy()[1, 0] == 200
|
|
assert limit_think_status.numpy()[0] == 0
|
|
assert limit_think_status.numpy()[1] == 0
|
|
|
|
def test_force_truncation_when_max_think_len_exceeded(self):
|
|
"""Test force truncation when step >= max_think_len"""
|
|
next_tokens = paddle.to_tensor([[100], [200]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([5, 8], dtype="int32")
|
|
step_idx = paddle.to_tensor([[5], [10]], dtype="int64") # Both exceed or equal limit
|
|
limit_think_status = paddle.to_tensor([0, 0], dtype="int32")
|
|
think_end_id = 999
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v1(next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id)
|
|
|
|
# Verify: tokens replaced with think_end_id, status changed to 2
|
|
assert next_tokens.numpy()[0, 0] == 999 # Replaced
|
|
assert next_tokens.numpy()[1, 0] == 999 # Replaced
|
|
assert limit_think_status.numpy()[0] == 2 # Status updated
|
|
assert limit_think_status.numpy()[1] == 2 # Status updated
|
|
|
|
def test_model_naturally_generates_think_end_id(self):
|
|
"""Test when model naturally generates think_end_id"""
|
|
next_tokens = paddle.to_tensor([[999]], dtype="int64") # Model generated think_end_id
|
|
max_think_lens = paddle.to_tensor([10], dtype="int32")
|
|
step_idx = paddle.to_tensor([[3]], dtype="int64") # Still within limit
|
|
limit_think_status = paddle.to_tensor([0], dtype="int32")
|
|
think_end_id = 999
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v1(next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id)
|
|
|
|
# Verify: token unchanged (already think_end_id), status changed to 2
|
|
assert next_tokens.numpy()[0, 0] == 999
|
|
assert limit_think_status.numpy()[0] == 2 # Move to response phase
|
|
|
|
def test_status_1_to_status_2_transition(self):
|
|
"""Test transition from status 1 (injected) to status 2 (confirmed)"""
|
|
next_tokens = paddle.to_tensor([[999]], dtype="int64") # think_end_id from previous injection
|
|
max_think_lens = paddle.to_tensor([5], dtype="int32")
|
|
step_idx = paddle.to_tensor([[6]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([1], dtype="int32") # Status is 1
|
|
think_end_id = 999
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v1(next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id)
|
|
|
|
# Verify: status changed to 2
|
|
assert limit_think_status.numpy()[0] == 2
|
|
|
|
def test_disabled_feature_negative_max_think_len(self):
|
|
"""Test that negative max_think_len disables the feature"""
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([-1], dtype="int32") # Disabled
|
|
step_idx = paddle.to_tensor([[100]], dtype="int64") # Would exceed limit if enabled
|
|
limit_think_status = paddle.to_tensor([0], dtype="int32")
|
|
think_end_id = 999
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v1(next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id)
|
|
|
|
# Verify: nothing changed
|
|
assert next_tokens.numpy()[0, 0] == 100
|
|
assert limit_think_status.numpy()[0] == 0
|
|
|
|
def test_already_in_response_phase_status_2(self):
|
|
"""Test that status 2 (response phase) is terminal"""
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([5], dtype="int32")
|
|
step_idx = paddle.to_tensor([[10]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([2], dtype="int32") # Already in response phase
|
|
think_end_id = 999
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v1(next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id)
|
|
|
|
# Verify: nothing changed
|
|
assert next_tokens.numpy()[0, 0] == 100
|
|
assert limit_think_status.numpy()[0] == 2
|
|
|
|
def test_mixed_batch(self):
|
|
"""Test batch with different sequences in different states"""
|
|
next_tokens = paddle.to_tensor([[100], [200], [999], [300]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([10, 5, 8, -1], dtype="int32")
|
|
step_idx = paddle.to_tensor([[3], [5], [4], [100]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([0, 0, 0, 0], dtype="int32")
|
|
think_end_id = 999
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v1(next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id)
|
|
|
|
# Verify each sequence
|
|
# Seq 0: step 3 < max 10, status 0, token unchanged
|
|
assert next_tokens.numpy()[0, 0] == 100
|
|
assert limit_think_status.numpy()[0] == 0
|
|
|
|
# Seq 1: step 5 >= max 5, force inject think_end_id, status -> 2
|
|
assert next_tokens.numpy()[1, 0] == 999
|
|
assert limit_think_status.numpy()[1] == 2
|
|
|
|
# Seq 2: step 4 < max 8, but token is think_end_id, status -> 2
|
|
assert next_tokens.numpy()[2, 0] == 999
|
|
assert limit_think_status.numpy()[2] == 2
|
|
|
|
# Seq 3: disabled (max -1), unchanged
|
|
assert next_tokens.numpy()[3, 0] == 300
|
|
assert limit_think_status.numpy()[3] == 0
|
|
|
|
|
|
class TestLimitThinkingContentLengthV2(unittest.TestCase):
|
|
"""Tests for limit_thinking_content_length_v2 operator (\n</think>\n\n strategy)"""
|
|
|
|
def test_normal_thinking_phase_no_limit_reached(self):
|
|
"""Test normal thinking phase when step < max_think_len"""
|
|
next_tokens = paddle.to_tensor([[100], [200]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([10, 15], dtype="int32")
|
|
step_idx = paddle.to_tensor([[5], [8]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([0, 0], dtype="int32")
|
|
think_end_id = 999
|
|
line_break_id = 888
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
|
|
# Verify: tokens unchanged, status unchanged
|
|
assert next_tokens.numpy()[0, 0] == 100
|
|
assert next_tokens.numpy()[1, 0] == 200
|
|
assert limit_think_status.numpy()[0] == 0
|
|
assert limit_think_status.numpy()[1] == 0
|
|
|
|
def test_force_truncation_sequence_injection(self):
|
|
"""Test force truncation with \n</think>\n\n sequence injection"""
|
|
# Test step == max_think_len (inject first \n)
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([5], dtype="int32")
|
|
step_idx = paddle.to_tensor([[5]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([0], dtype="int32")
|
|
think_end_id = 999
|
|
line_break_id = 888
|
|
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
assert next_tokens.numpy()[0, 0] == 888 # line_break_id
|
|
assert limit_think_status.numpy()[0] == 1
|
|
|
|
# Test step == max_think_len + 1 (inject </think>)
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
step_idx = paddle.to_tensor([[6]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([1], dtype="int32")
|
|
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
assert next_tokens.numpy()[0, 0] == 999 # think_end_id
|
|
assert limit_think_status.numpy()[0] == 1
|
|
|
|
# Test step == max_think_len + 2 (inject second \n)
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
step_idx = paddle.to_tensor([[7]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([1], dtype="int32")
|
|
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
assert next_tokens.numpy()[0, 0] == 888 # line_break_id
|
|
assert limit_think_status.numpy()[0] == 1
|
|
|
|
# Test step == max_think_len + 3 (inject third \n and finish)
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
step_idx = paddle.to_tensor([[8]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([1], dtype="int32")
|
|
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
assert next_tokens.numpy()[0, 0] == 888 # line_break_id
|
|
assert limit_think_status.numpy()[0] == 3 # Move to status 3
|
|
|
|
def test_model_naturally_generates_think_end_id(self):
|
|
"""Test when model naturally generates think_end_id"""
|
|
next_tokens = paddle.to_tensor([[999]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([10], dtype="int32")
|
|
step_idx = paddle.to_tensor([[3]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([0], dtype="int32")
|
|
think_end_id = 999
|
|
line_break_id = 888
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
|
|
# Verify: status changed to 3 (response phase)
|
|
assert next_tokens.numpy()[0, 0] == 999
|
|
assert limit_think_status.numpy()[0] == 3
|
|
|
|
def test_status_2_to_status_3_transition(self):
|
|
"""Test transition from status 2 (replacement done) to status 3 (thinking ended)"""
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([5], dtype="int32")
|
|
step_idx = paddle.to_tensor([[9]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([2], dtype="int32")
|
|
think_end_id = 999
|
|
line_break_id = 888
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
|
|
# Verify: status changed to 3
|
|
assert limit_think_status.numpy()[0] == 3
|
|
|
|
def test_disabled_feature_negative_max_think_len(self):
|
|
"""Test that negative max_think_len disables the feature"""
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([-1], dtype="int32")
|
|
step_idx = paddle.to_tensor([[100]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([0], dtype="int32")
|
|
think_end_id = 999
|
|
line_break_id = 888
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
|
|
# Verify: nothing changed
|
|
assert next_tokens.numpy()[0, 0] == 100
|
|
assert limit_think_status.numpy()[0] == 0
|
|
|
|
def test_already_in_response_phase_status_3(self):
|
|
"""Test that status 3 (response phase) is terminal"""
|
|
next_tokens = paddle.to_tensor([[100]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([5], dtype="int32")
|
|
step_idx = paddle.to_tensor([[10]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([3], dtype="int32")
|
|
think_end_id = 999
|
|
line_break_id = 888
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
|
|
# Verify: nothing changed
|
|
assert next_tokens.numpy()[0, 0] == 100
|
|
assert limit_think_status.numpy()[0] == 3
|
|
|
|
def test_mixed_batch_various_states(self):
|
|
"""Test batch with sequences in different states"""
|
|
next_tokens = paddle.to_tensor([[100], [200], [999], [300], [400]], dtype="int64")
|
|
max_think_lens = paddle.to_tensor([10, 5, 8, -1, 6], dtype="int32")
|
|
step_idx = paddle.to_tensor([[3], [5], [4], [100], [9]], dtype="int64")
|
|
limit_think_status = paddle.to_tensor([0, 0, 0, 0, 2], dtype="int32")
|
|
think_end_id = 999
|
|
line_break_id = 888
|
|
|
|
# Run operator
|
|
limit_thinking_content_length_v2(
|
|
next_tokens, max_think_lens, step_idx, limit_think_status, think_end_id, line_break_id
|
|
)
|
|
|
|
# Seq 0: step 3 < max 10, status 0, unchanged
|
|
assert next_tokens.numpy()[0, 0] == 100
|
|
assert limit_think_status.numpy()[0] == 0
|
|
|
|
# Seq 1: step 5 == max 5, inject line_break_id, status -> 1
|
|
assert next_tokens.numpy()[1, 0] == 888
|
|
assert limit_think_status.numpy()[1] == 1
|
|
|
|
# Seq 2: token is think_end_id, status 0 -> 3
|
|
assert next_tokens.numpy()[2, 0] == 999
|
|
assert limit_think_status.numpy()[2] == 3
|
|
|
|
# Seq 3: disabled, unchanged
|
|
assert next_tokens.numpy()[3, 0] == 300
|
|
assert limit_think_status.numpy()[3] == 0
|
|
|
|
# Seq 4: status 2 (replacement done), transition to 3
|
|
assert limit_think_status.numpy()[4] == 3
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|