233 lines
10 KiB
Python
233 lines
10 KiB
Python
from __future__ import annotations
|
||
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from app.services.pmflow_ai_agent_runtime.agent_goal_frame_runtime import _goal_frame_safe_conversation_context
|
||
from app.services.pmflow_ai_agent_runtime.agent_step_planner import _planner_safe_conversation_context
|
||
from app.services.pmflow_ai_operating_runtime.mission_service import _mission_safe_conversation_context
|
||
from app.services.pmflow_ai_context_pack_service import (
|
||
PMFlowAIContextPackBuilder,
|
||
context_pack_memory_claim_rows,
|
||
context_pack_observability_packet,
|
||
context_pack_thread_claim_rows,
|
||
evaluate_context_replay_case,
|
||
)
|
||
from app.services.pmflow_ai_context_memory_service import PMFlowAIContextMemoryService
|
||
from app.services.pmflow_ai_thread_context_service import (
|
||
PMFlowAIThreadContextService,
|
||
thread_context_claim_rows,
|
||
update_thread_context_summary_metadata,
|
||
)
|
||
|
||
|
||
def _message(index: int, role: str, content: str) -> SimpleNamespace:
|
||
return SimpleNamespace(
|
||
public_message_id=f"msg_{index}",
|
||
role=role,
|
||
message_index=index,
|
||
run_public_id=f"run_{index}",
|
||
sent_at=None,
|
||
content=content,
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_thread_context_pack_recovers_same_thread_purchase_discussion(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
service = PMFlowAIThreadContextService(db=SimpleNamespace(), tenant_id=12)
|
||
messages = [
|
||
_message(1, "user", "AI采购平台怎么样"),
|
||
_message(
|
||
2,
|
||
"assistant",
|
||
"AI采购平台通常用于智能寻源与供应商匹配、合同分析与合规审查、支出分析与预测。构建者应避免全能平台，先从发票自动比对或合同审查切入 MVP。",
|
||
),
|
||
_message(3, "user", "最近广州天气怎么样"),
|
||
_message(4, "assistant", "广州近期有降雨，适合调整线下安排。"),
|
||
]
|
||
|
||
async def _fake_load_messages(*_args, **_kwargs):
|
||
return messages
|
||
|
||
monkeypatch.setattr(service, "_load_messages", _fake_load_messages)
|
||
|
||
pack = await service.build_context_pack(
|
||
thread=SimpleNamespace(id=26, public_thread_id="thread_71ade449d4b5"),
|
||
user_question="你还记得你上面的对话有采购相关的内容吗？再给我详细说说是什么内容",
|
||
)
|
||
|
||
relevant_text = "\n".join(str(item.get("content") or "") for item in pack["relevant_messages"])
|
||
assert "智能寻源与供应商匹配" in relevant_text
|
||
assert "合同分析与合规审查" in relevant_text
|
||
|
||
claim_text = "\n".join(row["content"] for row in thread_context_claim_rows(pack))
|
||
assert "发票自动比对" in claim_text
|
||
|
||
context = {"thread_context_pack": pack}
|
||
assert "thread_context_pack" in _mission_safe_conversation_context(context)
|
||
assert "thread_context_pack" in _goal_frame_safe_conversation_context(context, question="继续说上面内容")
|
||
assert "thread_context_pack" in _planner_safe_conversation_context(context, question="继续说上面内容")
|
||
|
||
|
||
def test_unified_context_pack_budgets_and_preserves_relevant_thread_history() -> None:
|
||
thread_pack = {
|
||
"packet_version": "pmflow_ai_thread_context_pack.v1",
|
||
"thread_id": "thread_purchase",
|
||
"recent_messages": [
|
||
{
|
||
"message_id": f"recent_{index}",
|
||
"role": "user" if index % 2 else "assistant",
|
||
"message_index": index,
|
||
"content": f"普通上下文 {index} " + ("产品背景 " * 80),
|
||
}
|
||
for index in range(1, 18)
|
||
],
|
||
"relevant_messages": [
|
||
{
|
||
"message_id": "relevant_procurement",
|
||
"role": "assistant",
|
||
"message_index": 2,
|
||
"content": "前文已讨论 AI 采购平台：智能寻源、供应商匹配、合同分析、支出预测，MVP 可先从发票自动比对切入。",
|
||
}
|
||
],
|
||
"rolling_summary": {
|
||
"durable_points": [
|
||
{
|
||
"role": "assistant",
|
||
"source_message_indexes": [2],
|
||
"summary": "AI 采购平台的核心价值是采购流程自动化与成本/合规风险控制。",
|
||
}
|
||
]
|
||
},
|
||
}
|
||
conversation_context = {
|
||
"session_id": "qa_1",
|
||
"conversation_scope": "space_ai",
|
||
"thread_context_pack": thread_pack,
|
||
"context_capsules": [
|
||
{
|
||
"capsule_type": "pmflow_ai_object_ledger",
|
||
"current_focus_object": {"object_type": "project", "object_id": "p1", "display_name": "AI采购平台"},
|
||
"recent_objects": [{"object_type": "project", "object_id": "p1"}],
|
||
}
|
||
],
|
||
}
|
||
tenant_project_context = {
|
||
"tenant_id": 1,
|
||
"user_id": 2,
|
||
"surface_context": {
|
||
"visible_objects": [{"object_type": "project", "object_id": "p1", "display_name": "AI采购平台"}],
|
||
"space_runtime_context": {
|
||
"workbench": {
|
||
"projects": [{"id": "p1"}],
|
||
"pending_confirmations": [{"id": "c1"}],
|
||
},
|
||
"space_operating_packet": {"summary": "visible " * 500},
|
||
},
|
||
},
|
||
}
|
||
|
||
context_pack = PMFlowAIContextPackBuilder(budget_chars=8500).build(
|
||
user_question="你还记得上面采购平台相关内容吗？",
|
||
thread_context_pack=thread_pack,
|
||
conversation_context=conversation_context,
|
||
tenant_project_context=tenant_project_context,
|
||
)
|
||
|
||
assert context_pack["packet_version"] == "pmflow_ai_context_pack.v1"
|
||
assert context_pack["current_user_message"]["content"] == "你还记得上面采购平台相关内容吗？"
|
||
assert context_pack["budget_report"]["max_chars"] == 8500
|
||
assert any(row["section"] == "thread_context" and row["status"] in {"included", "compressed"} for row in context_pack["budget_report"]["sections"])
|
||
relevant_text = "\n".join(str(item.get("content") or "") for item in context_pack["thread_context"]["relevant_messages"])
|
||
assert "发票自动比对" in relevant_text
|
||
|
||
claim_text = "\n".join(row["content"] for row in context_pack_thread_claim_rows(context_pack))
|
||
assert "智能寻源" in claim_text
|
||
|
||
safe_context = {"context_pack": context_pack, "thread_context_pack": thread_pack}
|
||
assert "context_pack" in _mission_safe_conversation_context(safe_context)
|
||
assert "context_pack" in _goal_frame_safe_conversation_context(safe_context, question="继续上文")
|
||
assert "context_pack" in _planner_safe_conversation_context(safe_context, question="继续上文")
|
||
|
||
|
||
def test_unified_context_pack_includes_layered_memory_and_replay_observability() -> None:
|
||
memory_context_pack = {
|
||
"packet_version": "pmflow_ai_layered_memory_pack.v1",
|
||
"scope_refs": {"thread": ["thread:abc"], "user": ["user:2"], "workspace": ["workspace:space"]},
|
||
"memories": [
|
||
{
|
||
"memory_id": "ctxmem_1",
|
||
"memory_scope": "thread",
|
||
"scope_ref": "thread:abc",
|
||
"summary": "前文已经把 AI 采购平台定位为从发票自动比对切入，后续要补供应商匹配和合同分析。",
|
||
"durable_facts": [{"summary": "MVP 切口是发票自动比对。"}],
|
||
"open_loops": [{"summary": "还缺预算、供应商类型和审批流程信息。"}],
|
||
"confidence": "high",
|
||
},
|
||
{
|
||
"memory_id": "ctxmem_2",
|
||
"memory_scope": "user",
|
||
"scope_ref": "user:2",
|
||
"summary": "用户要求不要用硬编码关键词或枚举凑场景，要做终局架构。",
|
||
"preferences": [{"summary": "少请示，直接查主链并推进。", "confidence": "high"}],
|
||
"confidence": "high",
|
||
},
|
||
],
|
||
}
|
||
|
||
context_pack = PMFlowAIContextPackBuilder(budget_chars=9000).build(
|
||
user_question="继续上面采购平台这个方向",
|
||
thread_context_pack={},
|
||
conversation_context={"session_id": "qa_1", "conversation_scope": "space_ai"},
|
||
tenant_project_context={"tenant_id": 1, "user_id": 2, "surface_context": {}},
|
||
memory_context_pack=memory_context_pack,
|
||
)
|
||
|
||
memory_text = "\n".join(row["content"] for row in context_pack_memory_claim_rows(context_pack))
|
||
assert "发票自动比对" in memory_text
|
||
assert "不要用硬编码关键词" in memory_text
|
||
|
||
observability = context_pack_observability_packet(context_pack)
|
||
assert observability["memory"]["memory_count"] == 2
|
||
assert "thread" in observability["memory"]["scopes"]
|
||
|
||
replay = evaluate_context_replay_case(
|
||
context_pack=context_pack,
|
||
expected_evidence=["发票自动比对", "终局架构"],
|
||
)
|
||
assert replay["passed"] is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_context_memory_compactor_falls_back_without_runtime_generator() -> None:
|
||
service = PMFlowAIContextMemoryService(db=SimpleNamespace(), tenant_id=1, generator=SimpleNamespace())
|
||
result = await service._compact_turn(
|
||
user_message="AI采购平台怎么样",
|
||
assistant_answer="建议先从发票自动比对切入，再验证供应商匹配和合同分析。",
|
||
previous_memory_pack={},
|
||
context_pack={},
|
||
user_message_index=1,
|
||
assistant_message_index=2,
|
||
)
|
||
|
||
assert result["compaction_source"] == "deterministic_fallback"
|
||
assert "thread" in result["memories"]
|
||
assert "发票自动比对" in result["memories"]["thread"]["summary"]
|
||
|
||
|
||
def test_thread_context_summary_metadata_compacts_turns_without_scene_keywords() -> None:
|
||
metadata = update_thread_context_summary_metadata(
|
||
thread_metadata={},
|
||
user_message="AI采购平台怎么样",
|
||
assistant_answer="AI采购平台通常用于智能寻源与供应商匹配、合同分析与合规审查。",
|
||
user_message_index=1,
|
||
assistant_message_index=2,
|
||
)
|
||
summary = metadata["thread_context_summary"]
|
||
|
||
assert summary["summary_version"] == "pmflow_ai_thread_summary.v1"
|
||
assert summary["covered_message_count"] == 2
|
||
assert len(summary["durable_points"]) == 2
|
||
assert "智能寻源与供应商匹配" in summary["durable_points"][1]["summary"]
|