# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import pytest
from transformers import AutoTokenizer
from tests.reasoning.utils import run_reasoning_extraction
from vllm.reasoning import ReasoningParser, ReasoningParserManager
parser_name = "qwen3"
start_token = ""
end_token = ""
REASONING_MODEL_NAME = "Qwen/Qwen3-0.6B"
@pytest.fixture(scope="module")
def qwen3_tokenizer():
return AutoTokenizer.from_pretrained(REASONING_MODEL_NAME)
# 带 ,非stream
WITH_THINK = {
"output": "This is a reasoning sectionThis is the rest",
"reasoning_content": "This is a reasoning section",
"content": "This is the rest",
}
# 带 ,stream
WITH_THINK_STREAM = {
"output": "This is a reasoning sectionThis is the rest",
"reasoning_content": "This is a reasoning section",
"content": "This is the rest",
}
# 不带 ,非stream
WITHOUT_THINK = {
"output": "This is the rest",
"reasoning_content": None,
"content": "This is the rest",
}
# 不带 ,stream
WITHOUT_THINK_STREAM = {
"output": "This is the rest",
"reasoning_content": None,
"content": "This is the rest",
}
COMPLETE_REASONING = {
"output": "This is a reasoning section",
"reasoning_content": "This is a reasoning section",
"content": None,
}
MULTILINE_REASONING = {
"output":
"This is a reasoning\nsectionThis is the rest\nThat",
"reasoning_content": "This is a reasoning\nsection",
"content": "This is the rest\nThat",
}
ONLY_OPEN_TAG = {
"output": "This is a reasoning section",
"reasoning_content": None,
"content": "This is a reasoning section",
}
ONLY_OPEN_TAG_STREAM = {
"output": "This is a reasoning section",
"reasoning_content": "This is a reasoning section",
"content": None,
}
TEST_CASES = [
pytest.param(
False,
WITH_THINK,
id="with_think",
),
pytest.param(
True,
WITH_THINK_STREAM,
id="with_think_stream",
),
pytest.param(
False,
WITHOUT_THINK,
id="without_think",
),
pytest.param(
True,
WITHOUT_THINK_STREAM,
id="without_think_stream",
),
pytest.param(
False,
COMPLETE_REASONING,
id="complete_reasoning",
),
pytest.param(
True,
COMPLETE_REASONING,
id="complete_reasoning_stream",
),
pytest.param(
False,
MULTILINE_REASONING,
id="multiline_reasoning",
),
pytest.param(
True,
MULTILINE_REASONING,
id="multiline_reasoning_stream",
),
pytest.param(
False,
ONLY_OPEN_TAG,
id="only_open_tag",
),
pytest.param(
True,
ONLY_OPEN_TAG_STREAM,
id="only_open_tag_stream",
),
]
@pytest.mark.parametrize("streaming, param_dict", TEST_CASES)
def test_reasoning(
streaming: bool,
param_dict: dict,
qwen3_tokenizer,
):
output = qwen3_tokenizer.tokenize(param_dict["output"])
output_tokens: list[str] = [
qwen3_tokenizer.convert_tokens_to_string([token]) for token in output
]
parser: ReasoningParser = ReasoningParserManager.get_reasoning_parser(
parser_name)(qwen3_tokenizer)
reasoning, content = run_reasoning_extraction(parser,
output_tokens,
streaming=streaming)
assert reasoning == param_dict["reasoning_content"]
assert content == param_dict["content"]