112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
"""Reasoning field compatibility: vLLM `reasoning` vs older `reasoning_content`."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from langchain_core.messages import AIMessage, AIMessageChunk
|
|
|
|
from deerflow.models.patched_openai import _extract_reasoning_content, _restore_reasoning_content
|
|
from deerflow.models.reasoning_fields import (
|
|
apply_reasoning_from_payload,
|
|
extract_reasoning_text,
|
|
fill_ui_reasoning_content,
|
|
restore_reasoning_fields,
|
|
)
|
|
from deerflow.models.vllm_provider import (
|
|
_convert_delta_to_message_chunk_with_reasoning,
|
|
_restore_reasoning_field,
|
|
)
|
|
|
|
|
|
def test_extract_reasoning_reads_vllm_reasoning_field() -> None:
|
|
assert extract_reasoning_text({"reasoning": "先分析问题"}) == "先分析问题"
|
|
assert extract_reasoning_text({"reasoning_content": "逐步推理"}) == "逐步推理"
|
|
assert extract_reasoning_text({"content": "答案"}) is None
|
|
assert _extract_reasoning_content({"reasoning": "先分析问题"}) == "先分析问题"
|
|
|
|
|
|
def test_extract_reasoning_reads_structured_reasoning_object() -> None:
|
|
assert extract_reasoning_text({"reasoning": {"content": "结构化思考"}}) == "结构化思考"
|
|
assert extract_reasoning_text({"reasoning": {"text": "逐步分析"}}) == "逐步分析"
|
|
|
|
|
|
def test_extract_prefers_reasoning_content_when_both_present() -> None:
|
|
assert (
|
|
extract_reasoning_text({"reasoning_content": "旧字段", "reasoning": "新字段"})
|
|
== "旧字段"
|
|
)
|
|
|
|
|
|
def test_apply_new_vllm_reasoning_keeps_ui_field_and_wire_field() -> None:
|
|
message = apply_reasoning_from_payload(AIMessage(content="答案"), {"reasoning": "新版思考"})
|
|
assert message.additional_kwargs["reasoning"] == "新版思考"
|
|
assert message.additional_kwargs["reasoning_content"] == "新版思考"
|
|
|
|
|
|
def test_apply_old_reasoning_content_does_not_invent_reasoning() -> None:
|
|
message = apply_reasoning_from_payload(
|
|
AIMessage(content="答案"),
|
|
{"reasoning_content": "旧版思考"},
|
|
)
|
|
assert "reasoning" not in message.additional_kwargs
|
|
assert message.additional_kwargs["reasoning_content"] == "旧版思考"
|
|
|
|
|
|
def test_restore_new_vllm_echoes_reasoning_only() -> None:
|
|
orig = AIMessage(
|
|
content="答案",
|
|
additional_kwargs={"reasoning": "新版思考", "reasoning_content": "新版思考"},
|
|
)
|
|
payload: dict = {"role": "assistant", "content": "答案"}
|
|
restore_reasoning_fields(payload, orig)
|
|
assert payload["reasoning"] == "新版思考"
|
|
assert "reasoning_content" not in payload
|
|
|
|
|
|
def test_restore_old_vllm_echoes_reasoning_content_only() -> None:
|
|
orig = AIMessage(content="答案", additional_kwargs={"reasoning_content": "旧版思考"})
|
|
payload: dict = {"role": "assistant", "content": "答案"}
|
|
restore_reasoning_fields(payload, orig)
|
|
assert payload["reasoning_content"] == "旧版思考"
|
|
assert "reasoning" not in payload
|
|
|
|
|
|
def test_patched_openai_restore_does_not_add_reasoning_for_deepseek_turns() -> None:
|
|
orig = AIMessage(content="答案", additional_kwargs={"reasoning_content": "DeepSeek思考"})
|
|
payload: dict = {"role": "assistant", "content": "答案"}
|
|
_restore_reasoning_content(payload, orig)
|
|
assert payload == {"role": "assistant", "content": "答案", "reasoning_content": "DeepSeek思考"}
|
|
|
|
|
|
def test_vllm_stream_delta_reasoning_maps_to_ui_field() -> None:
|
|
chunk = _convert_delta_to_message_chunk_with_reasoning(
|
|
{"role": "assistant", "content": "", "reasoning": "流式思考"},
|
|
AIMessageChunk,
|
|
)
|
|
assert isinstance(chunk, AIMessageChunk)
|
|
assert chunk.additional_kwargs["reasoning"] == "流式思考"
|
|
assert chunk.additional_kwargs["reasoning_content"] == "流式思考"
|
|
|
|
|
|
def test_vllm_stream_delta_reasoning_content_does_not_copy_into_reasoning() -> None:
|
|
chunk = _convert_delta_to_message_chunk_with_reasoning(
|
|
{"role": "assistant", "content": "", "reasoning_content": "旧流式思考"},
|
|
AIMessageChunk,
|
|
)
|
|
assert isinstance(chunk, AIMessageChunk)
|
|
assert "reasoning" not in chunk.additional_kwargs
|
|
assert chunk.additional_kwargs["reasoning_content"] == "旧流式思考"
|
|
|
|
|
|
def test_vllm_restore_old_field_stays_on_reasoning_content() -> None:
|
|
orig = AIMessage(content="答案", additional_kwargs={"reasoning_content": "旧版思考"})
|
|
payload: dict = {"role": "assistant", "content": "答案"}
|
|
_restore_reasoning_field(payload, orig)
|
|
assert payload["reasoning_content"] == "旧版思考"
|
|
assert "reasoning" not in payload
|
|
|
|
|
|
def test_fill_ui_reasoning_content_from_langchain_kwargs() -> None:
|
|
message = AIMessage(content="答案", additional_kwargs={"reasoning": "网关已放入reasoning"})
|
|
filled = fill_ui_reasoning_content(message)
|
|
assert filled.additional_kwargs["reasoning_content"] == "网关已放入reasoning"
|