67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
"""Regression coverage for provider-safe transient context compression."""
|
|
|
|
from langchain.agents.middleware.types import ModelRequest, ModelResponse
|
|
from langchain_core.language_models.fake_chat_models import FakeListChatModel
|
|
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
|
|
|
|
from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, _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
|
|
|
|
|
|
def test_compression_uses_single_model_request_system_slot_for_generic_adapters():
|
|
model = FakeListChatModel(responses=["condensed history"])
|
|
middleware = DeerFlowSummarizationMiddleware(
|
|
model,
|
|
trigger=("messages", 3),
|
|
keep=("messages", 1),
|
|
)
|
|
request = ModelRequest(
|
|
model=model,
|
|
system_message=SystemMessage(content="agent prompt"),
|
|
messages=[
|
|
HumanMessage(content="first question", id="human-1"),
|
|
AIMessage(content="first answer", id="ai-1"),
|
|
HumanMessage(content="continue", id="human-2"),
|
|
],
|
|
)
|
|
captured: list[ModelRequest] = []
|
|
|
|
def handler(model_request: ModelRequest) -> ModelResponse:
|
|
captured.append(model_request)
|
|
return ModelResponse(result=[AIMessage(content="ok")])
|
|
|
|
middleware.wrap_model_call(request, handler)
|
|
|
|
assert len(captured) == 1
|
|
normalized = captured[0]
|
|
assert normalized.system_message is not None
|
|
assert "agent prompt" in normalized.system_message.text
|
|
assert "condensed history" in normalized.system_message.text
|
|
assert all(not isinstance(message, SystemMessage) for message in normalized.messages)
|
|
assert [message.type for message in normalized.messages] == ["human"]
|
|
assert normalized.messages[0].content == "continue"
|