2026-09-09 21:55:37 +08:00
|
|
|
from app.service.memory_service import MemoryService
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_memory_extraction_policy() -> None:
|
|
|
|
|
assert MemoryService.should_extract_memory(
|
|
|
|
|
conversation_content="我的风险偏好是稳健", role="user"
|
|
|
|
|
)
|
|
|
|
|
assert not MemoryService.should_extract_memory(conversation_content="", role="user")
|
|
|
|
|
assert MemoryService.should_extract_memory(
|
|
|
|
|
conversation_content="completed", role="system", event_type="trade.completed"
|
|
|
|
|
)
|
2026-09-10 15:55:54 +08:00
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_explicit_short_statement_triggers_without_length_threshold() -> None:
|
|
|
|
|
"""长度不再是门槛:两三个字的明确陈述(如"只买货币基金")必须触发。"""
|
|
|
|
|
assert MemoryService.should_extract_memory(conversation_content="只买货币基金", role="user")
|
|
|
|
|
assert MemoryService.should_extract_memory(conversation_content="随时赎回", role="user")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_plain_question_never_triggers_even_when_long() -> None:
|
|
|
|
|
"""普通问答即使很长也不触发:没有持久事实就没有记忆。"""
|
|
|
|
|
question = (
|
|
|
|
|
"帮我看看沪深300指数基金最近三个月的走势、同类排名和费率结构,"
|
|
|
|
|
"再解释一下它跟踪误差是怎么算出来的好吗?"
|
|
|
|
|
)
|
|
|
|
|
assert not MemoryService.should_extract_memory(conversation_content=question, role="user")
|
|
|
|
|
assert not MemoryService.should_extract_memory(
|
|
|
|
|
conversation_content="我想买基金", role="user")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_reply_and_unknown_event_do_not_trigger() -> None:
|
|
|
|
|
"""助手回复、以及非持久事实类事件(如运行完成)都不触发。"""
|
|
|
|
|
assert not MemoryService.should_extract_memory(
|
|
|
|
|
conversation_content="已为您记录该偏好,还需要什么帮助?", role="assistant")
|
|
|
|
|
assert not MemoryService.should_extract_memory(
|
|
|
|
|
conversation_content="agent.run_completed", role="system",
|
|
|
|
|
event_type="agent.run_completed")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_explicit_signal_and_tool_result_trigger() -> None:
|
|
|
|
|
"""调用点显式传入的受控信号、以及工具产出的权威事实同样触发。"""
|
|
|
|
|
assert MemoryService.should_extract_memory(
|
|
|
|
|
conversation_content="嗯", role="user", signals=("profile:family",))
|
|
|
|
|
assert MemoryService.should_extract_memory(
|
|
|
|
|
conversation_content="suitability check passed", role="system", tool_result=True)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_detect_memory_signals_returns_controlled_keys() -> None:
|
|
|
|
|
assert MemoryService.detect_memory_signals("我从事教师工作,已婚有两个孩子") == (
|
|
|
|
|
"profile:occupation", "profile:family",
|
|
|
|
|
)
|
|
|
|
|
assert MemoryService.detect_memory_signals("帮我查一下今天的净值") == ()
|