mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-04-25 11:18:22 +00:00
83039fa22c
* fix: use SystemMessage+HumanMessage for follow-up question generation (fixes #1697) Some models (e.g. MiniMax-M2.7) require the system prompt and user content to be passed as separate message objects rather than a single combined string. Invoking with a plain string sends everything as a HumanMessage, which causes these models to ignore the generation instructions and fail to produce valid follow-up questions. * test: verify model is invoked with SystemMessage and HumanMessage
128 lines
4.6 KiB
Python
128 lines
4.6 KiB
Python
import asyncio
|
|
from unittest.mock import MagicMock
|
|
|
|
from langchain_core.messages import HumanMessage, SystemMessage
|
|
|
|
from app.gateway.routers import suggestions
|
|
|
|
|
|
def test_strip_markdown_code_fence_removes_wrapping():
|
|
text = '```json\n["a"]\n```'
|
|
assert suggestions._strip_markdown_code_fence(text) == '["a"]'
|
|
|
|
|
|
def test_strip_markdown_code_fence_no_fence_keeps_content():
|
|
text = ' ["a"] '
|
|
assert suggestions._strip_markdown_code_fence(text) == '["a"]'
|
|
|
|
|
|
def test_parse_json_string_list_filters_invalid_items():
|
|
text = '```json\n["a", " ", 1, "b"]\n```'
|
|
assert suggestions._parse_json_string_list(text) == ["a", "b"]
|
|
|
|
|
|
def test_parse_json_string_list_rejects_non_list():
|
|
text = '{"a": 1}'
|
|
assert suggestions._parse_json_string_list(text) is None
|
|
|
|
|
|
def test_format_conversation_formats_roles():
|
|
messages = [
|
|
suggestions.SuggestionMessage(role="User", content="Hi"),
|
|
suggestions.SuggestionMessage(role="assistant", content="Hello"),
|
|
suggestions.SuggestionMessage(role="system", content="note"),
|
|
]
|
|
assert suggestions._format_conversation(messages) == "User: Hi\nAssistant: Hello\nsystem: note"
|
|
|
|
|
|
def test_generate_suggestions_parses_and_limits(monkeypatch):
|
|
req = suggestions.SuggestionsRequest(
|
|
messages=[
|
|
suggestions.SuggestionMessage(role="user", content="Hi"),
|
|
suggestions.SuggestionMessage(role="assistant", content="Hello"),
|
|
],
|
|
n=3,
|
|
model_name=None,
|
|
)
|
|
fake_model = MagicMock()
|
|
fake_model.invoke.return_value = MagicMock(content='```json\n["Q1", "Q2", "Q3", "Q4"]\n```')
|
|
monkeypatch.setattr(suggestions, "create_chat_model", lambda **kwargs: fake_model)
|
|
|
|
result = asyncio.run(suggestions.generate_suggestions("t1", req))
|
|
|
|
assert result.suggestions == ["Q1", "Q2", "Q3"]
|
|
|
|
|
|
def test_generate_suggestions_parses_list_block_content(monkeypatch):
|
|
req = suggestions.SuggestionsRequest(
|
|
messages=[
|
|
suggestions.SuggestionMessage(role="user", content="Hi"),
|
|
suggestions.SuggestionMessage(role="assistant", content="Hello"),
|
|
],
|
|
n=2,
|
|
model_name=None,
|
|
)
|
|
fake_model = MagicMock()
|
|
fake_model.invoke.return_value = MagicMock(content=[{"type": "text", "text": '```json\n["Q1", "Q2"]\n```'}])
|
|
monkeypatch.setattr(suggestions, "create_chat_model", lambda **kwargs: fake_model)
|
|
|
|
result = asyncio.run(suggestions.generate_suggestions("t1", req))
|
|
|
|
assert result.suggestions == ["Q1", "Q2"]
|
|
|
|
|
|
def test_generate_suggestions_parses_output_text_block_content(monkeypatch):
|
|
req = suggestions.SuggestionsRequest(
|
|
messages=[
|
|
suggestions.SuggestionMessage(role="user", content="Hi"),
|
|
suggestions.SuggestionMessage(role="assistant", content="Hello"),
|
|
],
|
|
n=2,
|
|
model_name=None,
|
|
)
|
|
fake_model = MagicMock()
|
|
fake_model.invoke.return_value = MagicMock(content=[{"type": "output_text", "text": '```json\n["Q1", "Q2"]\n```'}])
|
|
monkeypatch.setattr(suggestions, "create_chat_model", lambda **kwargs: fake_model)
|
|
|
|
result = asyncio.run(suggestions.generate_suggestions("t1", req))
|
|
|
|
assert result.suggestions == ["Q1", "Q2"]
|
|
|
|
|
|
def test_generate_suggestions_returns_empty_on_model_error(monkeypatch):
|
|
req = suggestions.SuggestionsRequest(
|
|
messages=[suggestions.SuggestionMessage(role="user", content="Hi")],
|
|
n=2,
|
|
model_name=None,
|
|
)
|
|
fake_model = MagicMock()
|
|
fake_model.invoke.side_effect = RuntimeError("boom")
|
|
monkeypatch.setattr(suggestions, "create_chat_model", lambda **kwargs: fake_model)
|
|
|
|
result = asyncio.run(suggestions.generate_suggestions("t1", req))
|
|
|
|
assert result.suggestions == []
|
|
|
|
|
|
def test_generate_suggestions_invokes_model_with_system_and_human_messages(monkeypatch):
|
|
req = suggestions.SuggestionsRequest(
|
|
messages=[
|
|
suggestions.SuggestionMessage(role="user", content="What is Python?"),
|
|
suggestions.SuggestionMessage(role="assistant", content="Python is a programming language."),
|
|
],
|
|
n=2,
|
|
model_name=None,
|
|
)
|
|
fake_model = MagicMock()
|
|
fake_model.invoke.return_value = MagicMock(content='["Q1", "Q2"]')
|
|
monkeypatch.setattr(suggestions, "create_chat_model", lambda **kwargs: fake_model)
|
|
|
|
asyncio.run(suggestions.generate_suggestions("t1", req))
|
|
|
|
call_args = fake_model.invoke.call_args[0][0]
|
|
assert len(call_args) == 2
|
|
assert isinstance(call_args[0], SystemMessage)
|
|
assert isinstance(call_args[1], HumanMessage)
|
|
assert "follow-up questions" in call_args[0].content
|
|
assert "What is Python?" in call_args[1].content
|