269 lines
12 KiB
Python
269 lines
12 KiB
Python
# ============================================================
|
||
# 黄金测试：滚动摘要 LLM 压缩生成方（矩阵 64 行剩余项 · 方向 L）
|
||
# stub provider：压缩成功 / 确定性回退 / 溯源链 / 预算门控（fail closed）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from server.agent_core.context_policy import (
|
||
ContextBudget,
|
||
ContextBudgetExceeded,
|
||
assemble_context,
|
||
estimate_tokens,
|
||
get_policy,
|
||
pin_message,
|
||
scroll_summary,
|
||
)
|
||
from server.agent_core.summarizer import (
|
||
SYSTEM_PROMPT,
|
||
SummaryCompression,
|
||
build_compression_prompt,
|
||
compress_scroll_summary,
|
||
deterministic_compress,
|
||
maybe_roll_session_summary,
|
||
)
|
||
|
||
|
||
class StubProvider:
|
||
"""可注入的 stub provider：记录调用；按配置返回/抛错/禁用。"""
|
||
|
||
def __init__(self, result: str | None = None, *, error: Exception | None = None,
|
||
disabled: bool = False):
|
||
self.result = result
|
||
self.error = error
|
||
self.disabled = disabled
|
||
self.calls: list[dict] = []
|
||
|
||
async def chat_text(self, system: str, user: str, timeout: float = 45.0) -> str | None:
|
||
self.calls.append({"system": system, "user": user, "timeout": timeout})
|
||
if self.error is not None:
|
||
raise self.error
|
||
if self.disabled:
|
||
return None
|
||
return self.result
|
||
|
||
|
||
def _messages(n: int = 30, repeat: int = 15) -> list[dict]:
|
||
"""构造长会话消息：每条约 4 + 9*repeat 个 CJK token。"""
|
||
unit = "插单交期优先确认内容" # 9 个 CJK 字符 ≈ 9 token
|
||
return [
|
||
{"id": f"m{i}", "role": "user" if i % 2 == 0 else "agent",
|
||
"text": f"第{i}条" + unit * repeat}
|
||
for i in range(n)
|
||
]
|
||
|
||
|
||
def _store_data(messages: list[dict] | None = None) -> dict:
|
||
return {
|
||
"systemPrompt": "你是 APS 排产智能体：提议权，写操作必须经门禁。",
|
||
"projects": [{
|
||
"id": "p1", "name": "青岛Q3排产", "scopeLabel": "青岛工厂/总装车间",
|
||
"sharedContext": ["V3 参数版本"], "archived": False,
|
||
}],
|
||
"sessions": [{"id": "s1", "projectId": "p1", "title": "本周主排产", "status": "running"}],
|
||
"messages": {"s1": messages if messages is not None else _messages()},
|
||
}
|
||
|
||
|
||
# ---------------- compress_scroll_summary：stub provider ----------------
|
||
|
||
async def test_llm_compress_success():
|
||
msgs = _messages(n=5, repeat=2)
|
||
stub = StubProvider(result="插单按交期优先确认；产能足够。")
|
||
out = await compress_scroll_summary(msgs, provider=stub, budget_tokens=1000)
|
||
assert isinstance(out, SummaryCompression)
|
||
assert out.mode == "llm"
|
||
assert out.text == "插单按交期优先确认；产能足够。"
|
||
assert out.compressed_message_ids == ["m0", "m1", "m2", "m3", "m4"]
|
||
assert out.reason is None
|
||
# 固定中文提示词：只基于给定消息、禁止编造
|
||
assert stub.calls and "禁止编造" in stub.calls[0]["system"]
|
||
assert SYSTEM_PROMPT == stub.calls[0]["system"]
|
||
user = stub.calls[0]["user"]
|
||
assert "预算：摘要估算 token 不得超过 1000" in user
|
||
assert "插单交期优先确认内容" in user # 消息原样进入提示词
|
||
assert "只输出压缩后的摘要正文" in user
|
||
|
||
|
||
async def test_llm_unavailable_falls_back_deterministic():
|
||
msgs = _messages(n=6, repeat=3)
|
||
out = await compress_scroll_summary(msgs, provider=StubProvider(result=None), budget_tokens=400)
|
||
assert out.mode == "deterministic"
|
||
assert out.reason == "provider-unavailable"
|
||
assert estimate_tokens(out.text) <= 400
|
||
assert out.text and "插单交期优先确认内容" in out.text
|
||
assert out.compressed_message_ids == [f"m{i}" for i in range(6)]
|
||
|
||
|
||
async def test_llm_error_timeout_falls_back_deterministic():
|
||
msgs = _messages(n=6, repeat=3)
|
||
out = await compress_scroll_summary(
|
||
msgs, provider=StubProvider(error=RuntimeError("timeout")), budget_tokens=400)
|
||
assert out.mode == "deterministic"
|
||
assert out.reason == "provider-error:RuntimeError"
|
||
assert estimate_tokens(out.text) <= 400
|
||
|
||
|
||
async def test_llm_over_budget_falls_back_deterministic():
|
||
msgs = _messages(n=6, repeat=3)
|
||
out = await compress_scroll_summary(msgs, provider=StubProvider(result="长" * 5000), budget_tokens=200)
|
||
assert out.mode == "deterministic"
|
||
assert out.reason == "llm-over-budget"
|
||
assert estimate_tokens(out.text) <= 200
|
||
|
||
|
||
async def test_llm_empty_output_falls_back_deterministic():
|
||
msgs = _messages(n=6, repeat=3)
|
||
out = await compress_scroll_summary(msgs, provider=StubProvider(result=" \n "), budget_tokens=400)
|
||
assert out.mode == "deterministic"
|
||
assert out.reason == "llm-empty-output"
|
||
assert out.text
|
||
|
||
|
||
async def test_llm_fence_wrapped_output_stripped():
|
||
msgs = _messages(n=3, repeat=1)
|
||
out = await compress_scroll_summary(
|
||
msgs, provider=StubProvider(result="```text\n压缩摘要正文\n```"), budget_tokens=400)
|
||
assert out.mode == "llm"
|
||
assert out.text == "压缩摘要正文"
|
||
|
||
|
||
async def test_compress_rejects_no_messages():
|
||
with pytest.raises(ValueError):
|
||
await compress_scroll_summary([], provider=StubProvider(result="x"), budget_tokens=100)
|
||
with pytest.raises(ValueError):
|
||
await compress_scroll_summary(_messages(), provider=StubProvider(result="x"), budget_tokens=0)
|
||
|
||
|
||
def test_deterministic_compress_budget_and_order():
|
||
msgs = [{"id": f"m{i}", "role": "user", "text": "消息内容" * 5} for i in range(20)]
|
||
out = deterministic_compress(msgs, budget_tokens=100)
|
||
assert out.mode == "deterministic"
|
||
assert out.reason == "budget-truncation"
|
||
assert estimate_tokens(out.text) <= 100
|
||
assert out.text.startswith("用户：消息内容") # 保序：从最旧开始保留
|
||
assert out.compressed_message_ids == [f"m{i}" for i in range(20)]
|
||
|
||
|
||
def test_prompt_template_is_fixed_and_bounded():
|
||
msgs = _messages(n=3, repeat=1)
|
||
prompt = build_compression_prompt(msgs, 123)
|
||
assert "预算：摘要估算 token 不得超过 123" in prompt
|
||
assert "需要压缩的对话（按时间顺序）：" in prompt
|
||
assert prompt.count("[消息") == 3
|
||
|
||
|
||
# ---------------- scroll_summary：summarizer 溯源标记 ----------------
|
||
|
||
def test_scroll_summary_records_compression_provenance_and_chain():
|
||
sd = _store_data(messages=_messages(n=4, repeat=1))
|
||
s1 = scroll_summary(sd, "s1", "验收标准：插单必须满足交期优先",
|
||
source_message_ids=["m0", "m1"], summarizer="llm")
|
||
assert s1["compressedBy"] == "llm"
|
||
assert s1["sourceMessageIds"] == ["m0", "m1"]
|
||
s2 = scroll_summary(sd, "s1", "插单已评估完毕，等待确认",
|
||
source_message_ids=["m2", "m3"], summarizer="deterministic")
|
||
assert s2["compressedBy"] == "deterministic"
|
||
assert s2["supersedes"] == s1["summaryId"] # supersedes 溯源链
|
||
assert s2["generation"] == 2
|
||
# 不传 summarizer：保持纯确定性行为，不写入新字段（向后兼容）
|
||
legacy = scroll_summary(sd, "s1", "第三版")
|
||
assert "compressedBy" not in legacy
|
||
|
||
|
||
# ---------------- maybe_roll_session_summary：预算门控接线 ----------------
|
||
|
||
async def test_maybe_roll_budget_gated_and_config_switch():
|
||
sd = _store_data() # 长会话（30 条）
|
||
stub = StubProvider(result="压缩摘要：插单按交期优先。")
|
||
|
||
# 未超预算：不触发，不写策略
|
||
out = await maybe_roll_session_summary(sd, "s1", provider=stub, budget=ContextBudget(session=10 ** 5))
|
||
assert out is None
|
||
assert get_policy(sd, "s1") == {}
|
||
|
||
# 配置开关：autoSummarize=False 时即使超预算也不滚动
|
||
out = await maybe_roll_session_summary(
|
||
sd, "s1", provider=stub, budget=ContextBudget(session=3000), auto_summarize=False)
|
||
assert out is None
|
||
assert get_policy(sd, "s1") == {}
|
||
|
||
# 超预算：滚动成功（LLM 压缩入库），会话层重新拟合预算
|
||
out = await maybe_roll_session_summary(sd, "s1", provider=stub, budget=ContextBudget(session=3000))
|
||
assert out is not None
|
||
assert out["text"] == "压缩摘要：插单按交期优先。"
|
||
assert out["compressedBy"] == "llm"
|
||
assert set(out["sourceMessageIds"]) <= {f"m{i}" for i in range(30)}
|
||
assert len(out["sourceMessageIds"]) >= 1
|
||
policy = get_policy(sd, "s1")
|
||
assert policy["summary"]["summaryId"] == out["summaryId"]
|
||
assembly = assemble_context(sd, "s1", None, ContextBudget(session=3000))
|
||
assert assembly.usage["session"] <= 3000
|
||
assert "压缩摘要" in assembly.layer("session")
|
||
|
||
# 已覆盖的消息不再重复压缩（幂等）：无新消息滑出窗口 -> 不触发
|
||
noop = await maybe_roll_session_summary(sd, "s1", provider=stub, budget=ContextBudget(session=3000))
|
||
assert noop is None
|
||
assert get_policy(sd, "s1")["summary"]["summaryId"] == out["summaryId"]
|
||
|
||
# 新一轮对话：新消息把 m22/m23 推出窗口 -> 滚动溯源链续接（supersedes）
|
||
sd["messages"]["s1"].extend([
|
||
{"id": "m30", "role": "user", "text": "第30条" + "插单交期优先确认内容" * 15},
|
||
{"id": "m31", "role": "agent", "text": "第31条" + "插单交期优先确认内容" * 15},
|
||
])
|
||
out2 = await maybe_roll_session_summary(sd, "s1", provider=stub, budget=ContextBudget(session=3000))
|
||
assert out2 is not None
|
||
assert out2["supersedes"] == out["summaryId"]
|
||
assert out2["generation"] == 2
|
||
assert set(out2["sourceMessageIds"]) == {"m22", "m23"}
|
||
|
||
|
||
async def test_maybe_roll_deterministic_fallback_when_provider_unavailable():
|
||
sd = _store_data()
|
||
out = await maybe_roll_session_summary(
|
||
sd, "s1", provider=StubProvider(result=None), budget=ContextBudget(session=3000))
|
||
assert out is not None
|
||
assert out["compressedBy"] == "deterministic"
|
||
assert estimate_tokens(out["text"]) <= 3000 // 2
|
||
policy = get_policy(sd, "s1")
|
||
assert policy["summary"]["compressedBy"] == "deterministic"
|
||
# 确定性回退仍可溯源到被压缩消息 id
|
||
assert policy["summary"]["sourceMessageIds"]
|
||
|
||
|
||
async def test_maybe_roll_fail_closed_when_still_over_budget():
|
||
"""滚动后会话层仍超预算（钉住超大消息）-> ContextBudgetExceeded 可预测拒绝。"""
|
||
sd = _store_data()
|
||
sd["messages"]["s1"].append({"id": "m_pin", "role": "user", "text": "验收标准" * 3000})
|
||
pin_message(sd, "s1", "m_pin", note="不可淘汰")
|
||
|
||
async def _attempt():
|
||
return await maybe_roll_session_summary(
|
||
sd, "s1", provider=StubProvider(result="压缩摘要"), budget=ContextBudget(session=3000))
|
||
|
||
with pytest.raises(ContextBudgetExceeded) as exc_info:
|
||
await _attempt()
|
||
exc = exc_info.value
|
||
assert exc.layer == "session"
|
||
assert exc.budget == 3000
|
||
assert exc.required > 3000
|
||
# 可预测：同一输入重复触发，层/用量/限额完全一致
|
||
with pytest.raises(ContextBudgetExceeded) as exc_info2:
|
||
await _attempt()
|
||
second = exc_info2.value
|
||
assert (second.layer, second.required, second.budget) == (exc.layer, exc.required, exc.budget)
|
||
|
||
|
||
async def test_maybe_roll_ignores_foreign_session():
|
||
"""越权会话（不在快照/项目不可达）-> 显式 ValueError，绝不滚动。"""
|
||
sd = _store_data()
|
||
sd["sessions"].append({"id": "s2", "projectId": "p2", "title": "别厂会话", "status": "running"})
|
||
sd["messages"]["s2"] = [{"id": "x0", "role": "user", "text": "机密：别厂排产方案"}]
|
||
stub = StubProvider(result="压缩摘要")
|
||
with pytest.raises(ValueError) as exc_info:
|
||
await maybe_roll_session_summary(sd, "s2", provider=stub, budget=ContextBudget())
|
||
assert "project not accessible" in str(exc_info.value)
|
||
# 授权会话 s1 的策略未被污染
|
||
assert get_policy(sd, "s1") == {}
|