31 lines
1.1 KiB
Python
31 lines
1.1 KiB
Python
"""Regression coverage for provider-safe transient context compression."""
|
|
|
|
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
|
|
|
|
from deerflow.agents.middlewares.summarization_middleware import _ensure_latest_human_message
|
|
|
|
|
|
def test_compression_restores_latest_human_turn_when_only_tool_tail_remains():
|
|
user = HumanMessage(content="请分析这份材料", id="user-1")
|
|
source = [
|
|
user,
|
|
AIMessage(content="", id="ai-1"),
|
|
ToolMessage(content="检索结果", tool_call_id="call-1", id="tool-1"),
|
|
]
|
|
compressed = [SystemMessage(content="internal summary"), *source[1:]]
|
|
|
|
result = _ensure_latest_human_message(source, compressed)
|
|
|
|
assert result[0].type == "system"
|
|
assert result[1] is user
|
|
assert [message.type for message in result[2:]] == ["ai", "tool"]
|
|
|
|
|
|
def test_compression_does_not_duplicate_human_turn_already_in_tail():
|
|
user = HumanMessage(content="继续", id="user-1")
|
|
compressed = [SystemMessage(content="internal summary"), user]
|
|
|
|
result = _ensure_latest_human_message([user], compressed)
|
|
|
|
assert result == compressed
|