224 lines
9.4 KiB
Python
224 lines
9.4 KiB
Python
from types import SimpleNamespace
|
|
|
|
from app.services.pmflow_ai_context_memory_job_service import build_context_memory_compaction_job_payload
|
|
from app.services.pmflow_ai_context_assembly_service import sanitized_context_assembly_payload
|
|
from app.services.pmflow_ai_context_evidence_service import append_context_evidence_to_workspace, build_context_evidence_packet
|
|
from app.services.pmflow_ai_context_grounding_verifier import (
|
|
revise_answer_with_context_boundary,
|
|
verify_answer_context_grounding,
|
|
)
|
|
from app.services.pmflow_ai_context_pack_service import PMFlowAIContextPackBuilder, context_pack_used_context_packet
|
|
from app.services.pmflow_ai_context_replay_runner import PMFlowAIContextReplayRunner
|
|
from app.schemas.pmflow_ai_agent_runtime import PMFlowAIAgentWorkspace
|
|
|
|
|
|
def test_context_memory_compaction_job_payload_is_sanitized_and_anchor_based():
|
|
thread = SimpleNamespace(id=11, public_thread_id="thread_abc", message_count=4)
|
|
user = SimpleNamespace(id=7)
|
|
|
|
payload = build_context_memory_compaction_job_payload(
|
|
thread=thread,
|
|
current_user=user,
|
|
user_message="请记住：首版先做执行包导出。",
|
|
assistant_answer="已记录，首版优先执行包导出。",
|
|
context_pack={
|
|
"thread_context": {"recent_messages": [{"content": "首版先做执行包导出。"}]},
|
|
"trace": {"internal": "should_not_leak"},
|
|
"prompt_key": "should_not_leak",
|
|
},
|
|
tenant_project_context={"current_project_id": 101, "raw_internal": {"table": "hidden"}},
|
|
run_public_id="run_123",
|
|
user_message_index=3,
|
|
assistant_message_index=4,
|
|
)
|
|
|
|
assert payload["thread"]["public_thread_id"] == "thread_abc"
|
|
assert payload["user_id"] == 7
|
|
assert payload["run_public_id"] == "run_123"
|
|
assert "执行包导出" in payload["user_message"]
|
|
assert "thread_context" in payload["context_pack"]
|
|
assert "trace" not in payload["context_pack"]
|
|
assert "prompt_key" not in payload["context_pack"]
|
|
|
|
|
|
def test_used_context_packet_reports_thread_memory_and_dropped_risks():
|
|
pack = PMFlowAIContextPackBuilder(budget_chars=9000).build(
|
|
user_question="你还记得上面说的约束吗？",
|
|
thread_context_pack={
|
|
"recent_messages": [
|
|
{
|
|
"message_id": "msg_1",
|
|
"role": "user",
|
|
"message_index": 1,
|
|
"content": "首版必须能导出给外部开发工具的执行包。",
|
|
}
|
|
],
|
|
"relevant_messages": [],
|
|
"rolling_summary": {},
|
|
},
|
|
conversation_context={},
|
|
tenant_project_context={},
|
|
memory_context_pack={
|
|
"memories": [
|
|
{
|
|
"memory_id": "ctxmem_1",
|
|
"memory_scope": "thread",
|
|
"scope_ref": "thread:thread_abc",
|
|
"summary": "用户要求首版支持执行包导出。",
|
|
"confidence_score": 0.9,
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
packet = context_pack_used_context_packet(pack)
|
|
|
|
assert packet["packet_version"] == "pmflow_ai_used_context_packet.v1"
|
|
assert packet["usage_status"] == "provided_to_answer_runtime"
|
|
assert packet["thread_message_refs"][0]["message_id"] == "msg_1"
|
|
assert packet["memory_refs"][0]["memory_id"] == "ctxmem_1"
|
|
assert "执行包" in packet["thread_message_refs"][0]["content_preview"]
|
|
|
|
|
|
def test_context_replay_runner_default_cases_pass():
|
|
result = PMFlowAIContextReplayRunner().run_cases()
|
|
|
|
assert result["case_count"] >= 4
|
|
assert result["failed_count"] == 0
|
|
assert all(row["passed"] for row in result["cases"])
|
|
|
|
|
|
def test_context_assembly_payload_keeps_sanitized_refs_not_raw_trace():
|
|
pack = PMFlowAIContextPackBuilder(budget_chars=9000).build(
|
|
user_question="继续上面的执行包约束",
|
|
thread_context_pack={
|
|
"recent_messages": [{"message_id": "m1", "role": "user", "message_index": 1, "content": "执行包必须能给外部工具使用。"}],
|
|
"relevant_messages": [],
|
|
"rolling_summary": {},
|
|
},
|
|
conversation_context={"trace": {"internal": "hidden"}},
|
|
tenant_project_context={},
|
|
memory_context_pack={"memories": [{"memory_id": "ctxmem_1", "memory_scope": "thread", "summary": "执行包需要外部工具可用。"}]},
|
|
)
|
|
|
|
assembly = sanitized_context_assembly_payload({**pack, "trace": {"internal": "hidden"}})
|
|
|
|
assert assembly["packet_version"] == "pmflow_ai_context_assembly.v1"
|
|
assert assembly["thread_refs"][0]["message_id"] == "m1"
|
|
assert assembly["memory_refs"][0]["memory_id"] == "ctxmem_1"
|
|
assert "trace" not in assembly
|
|
assert "scope_ref" not in assembly["memory_refs"][0]
|
|
|
|
|
|
def test_context_grounding_verifier_blocks_answer_that_ignores_selected_context():
|
|
used_context = {
|
|
"packet_version": "pmflow_ai_used_context_packet.v1",
|
|
"thread_message_refs": [
|
|
{
|
|
"message_id": "m1",
|
|
"role": "user",
|
|
"message_index": 1,
|
|
"content_preview": "首版必须能导出给外部开发工具的执行包。",
|
|
}
|
|
],
|
|
"memory_refs": [],
|
|
"dropped_risks": [],
|
|
}
|
|
|
|
failed = verify_answer_context_grounding(answer="可以继续推进产品规划。", used_context_packet=used_context)
|
|
assert failed["grounding_passed"] is False
|
|
assert failed["missing_context_count"] == 1
|
|
|
|
revised = revise_answer_with_context_boundary(answer="可以继续推进产品规划。", grounding_packet=failed)
|
|
assert "外部开发工具的执行包" in revised
|
|
|
|
passed = verify_answer_context_grounding(answer="下一步要围绕外部开发工具的执行包继续推进。", used_context_packet=used_context)
|
|
assert passed["grounding_passed"] is True
|
|
|
|
|
|
def test_context_evidence_absorber_promotes_prior_thread_plan_without_business_keyword_rules():
|
|
pack = PMFlowAIContextPackBuilder(budget_chars=12000).build(
|
|
user_question="那个方向继续按之前方案出一版新的最小闭环",
|
|
thread_context_pack={
|
|
"recent_messages": [
|
|
{
|
|
"message_id": "m1",
|
|
"role": "assistant",
|
|
"message_index": 2,
|
|
"content": "上一版方案确认先验证数据接入质量、智能助手建议和嵌入现有流程三个最小闭环。",
|
|
},
|
|
{
|
|
"message_id": "m2",
|
|
"role": "user",
|
|
"message_index": 3,
|
|
"content": "我希望这个方向先用低成本方式验证。",
|
|
},
|
|
],
|
|
"relevant_messages": [],
|
|
"rolling_summary": {},
|
|
},
|
|
conversation_context={},
|
|
tenant_project_context={
|
|
"surface_context": {
|
|
"visible_objects": [
|
|
{"object_ref": "project:unrelated", "title": "另一个当前可见项目"},
|
|
]
|
|
}
|
|
},
|
|
memory_context_pack={
|
|
"memories": [
|
|
{
|
|
"memory_id": "ctxmem_prior",
|
|
"memory_scope": "thread",
|
|
"summary": "该主题的持续方向是把早期想法收敛成可验证的产品蓝图。",
|
|
"confidence_score": 0.86,
|
|
}
|
|
]
|
|
},
|
|
)
|
|
evidence = build_context_evidence_packet(
|
|
context_pack=pack,
|
|
user_question="那个方向继续按之前方案出一版新的最小闭环",
|
|
)
|
|
|
|
assert evidence["claims"]
|
|
assert evidence["continuation_obligations"]
|
|
assert any(item["semantic_role"] == "assistant_prior_proposal" for item in evidence["claims"])
|
|
assert evidence["topic_workspace"]["topic_signature"].startswith("topic_")
|
|
|
|
workspace = PMFlowAIAgentWorkspace(current_goal="继续之前方案")
|
|
count = append_context_evidence_to_workspace(workspace=workspace, evidence_packet=evidence)
|
|
assert count >= 2
|
|
assert any(claim.source_tool == "context_evidence_absorber" for claim in workspace.claim_ledger)
|
|
assert not any("同一对话历史第" in claim.claim_text for claim in workspace.claim_ledger)
|
|
|
|
|
|
def test_grounding_verifier_requires_context_evidence_obligations():
|
|
used_context = {
|
|
"packet_version": "pmflow_ai_used_context_packet.v1",
|
|
"thread_message_refs": [],
|
|
"memory_refs": [],
|
|
"dropped_risks": [],
|
|
"context_evidence": {
|
|
"claims": [
|
|
{
|
|
"claim_id": "ctxclaim_1",
|
|
"semantic_role": "assistant_prior_proposal",
|
|
"claim_preview": "上一版方案确认先验证数据接入质量、智能助手建议和嵌入现有流程三个最小闭环。",
|
|
"required_for_answer": True,
|
|
}
|
|
],
|
|
"continuation_obligations": [{"claim_id": "ctxclaim_1"}],
|
|
},
|
|
}
|
|
|
|
failed = verify_answer_context_grounding(answer="可以重新规划一个方向。", used_context_packet=used_context)
|
|
assert failed["grounding_passed"] is False
|
|
assert any(item["finding_type"] == "context_obligation_underused" for item in failed["findings"])
|
|
|
|
passed = verify_answer_context_grounding(
|
|
answer="这版继续沿用数据接入质量、智能助手建议和嵌入现有流程这三个最小闭环。",
|
|
used_context_packet=used_context,
|
|
)
|
|
assert passed["grounding_passed"] is True
|