282 lines
10 KiB
Python
282 lines
10 KiB
Python
import json
|
|
from types import SimpleNamespace
|
|
|
|
from app.config.ai_prompts import PROMPT_SPACE_AI_INTERACTION_OPTION_PLANNER
|
|
from app.schemas.customer_answer import CustomerEvidencePacket
|
|
from app.schemas.rag import RAGAskRequest
|
|
from app.services.customer_answer_presentation_service import (
|
|
CustomerAnswerPresentationService,
|
|
_render_state_shape_answer,
|
|
_standardize_canvas_direct_answer,
|
|
)
|
|
|
|
|
|
FORBIDDEN_USER_TERMS = (
|
|
"System Snapshot",
|
|
"User Snapshot",
|
|
"Session State",
|
|
"Project database",
|
|
"Knowledge Graph",
|
|
"PROMPT_",
|
|
"AdaptiveRAG",
|
|
"SemanticDecision",
|
|
"backend/",
|
|
"frontend/",
|
|
"tenant_id",
|
|
"prompt_key",
|
|
"trace_id",
|
|
)
|
|
|
|
|
|
def make_service() -> CustomerAnswerPresentationService:
|
|
return CustomerAnswerPresentationService(object(), tenant_id=1)
|
|
|
|
|
|
def assert_customer_safe(text: str) -> None:
|
|
assert str(text or "").strip()
|
|
for term in FORBIDDEN_USER_TERMS:
|
|
assert term not in text
|
|
|
|
|
|
def test_space_ai_interaction_option_planner_prompt_formats_with_json_schema() -> None:
|
|
prompt = PROMPT_SPACE_AI_INTERACTION_OPTION_PLANNER.format(
|
|
customer_evidence_packet=json.dumps({"answer_type": "strategic_judgement"}, ensure_ascii=False),
|
|
question="我想做个AI混剪工具",
|
|
answer="先收敛成一个新创意版本。",
|
|
fallback_interaction_request=json.dumps(
|
|
{
|
|
"title": "还要你确认的点",
|
|
"questions": [{"id": "target_user", "options": []}],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
|
|
assert '"title": "还要你确认的点"' in prompt
|
|
assert '"id": "target_user | pain_point | mvp_loop | validation_signal | ..."' in prompt
|
|
assert "我想做个AI混剪工具" in prompt
|
|
|
|
|
|
def test_customer_evidence_packet_accepts_runtime_answer_shapes_and_sanitizes_terms() -> None:
|
|
packet = CustomerEvidencePacket.model_validate(
|
|
{
|
|
"question": "我想做个 AI 采购监督系统",
|
|
"answer_type": "strategic_judgement",
|
|
"facts": [
|
|
{
|
|
"type": "runtime_evidence_finding",
|
|
"display_label": "Project database 里的外部证据",
|
|
"display_summary": "System Snapshot 显示监管压力真实存在",
|
|
"business_values": {"含义": "监管压力真实存在", "tenant_id": "raw"},
|
|
"source_display_labels": ["Project database"],
|
|
}
|
|
],
|
|
"source_summary": ["Project database"],
|
|
}
|
|
).to_customer_prompt_json()
|
|
|
|
dumped = json.dumps(packet, ensure_ascii=False)
|
|
assert packet["answer_type"] == "strategic_judgement"
|
|
assert_customer_safe(dumped)
|
|
assert "监管压力真实存在" in dumped
|
|
|
|
|
|
def test_renderer_uses_turn_state_answer_shape_not_legacy_route_contract() -> None:
|
|
service = make_service()
|
|
packet = service.build_customer_evidence_packet(
|
|
request=RAGAskRequest(
|
|
question="我想做个 AI 采购监督系统，你觉得怎么样",
|
|
surface_context={
|
|
"workbench": {
|
|
"recentProjects": [
|
|
{
|
|
"id": 141,
|
|
"title": "童析-学龄前儿童行为特质AI评估助手",
|
|
"status": "active",
|
|
}
|
|
]
|
|
}
|
|
},
|
|
),
|
|
current_user=SimpleNamespace(id=7, tenant_id=1),
|
|
payload={
|
|
"pmflow_ai_route_plan": {
|
|
"goal_kind": "project_inventory",
|
|
"answer_contract": "surface_fact",
|
|
"source_strategy": "internal_only",
|
|
},
|
|
"pmflow_ai_turn_state": {
|
|
"answer_state": {"answer_shape": "strategic_judgement"},
|
|
"object_state": {"primary_object": {"object_type": "new_idea", "binding_strength": "none"}},
|
|
},
|
|
"evidence_findings": [
|
|
{
|
|
"fact": "公开政策显示采购监督压力真实存在。",
|
|
"implication": "机会不在报告生成，而在风险监督闭环。",
|
|
"source": "https://example.com/policy",
|
|
"relevance_to_goal": "validates_market_window",
|
|
"confidence": "high",
|
|
}
|
|
],
|
|
"sources": [{"source": "external", "display_label": "政策来源", "target_url": "https://example.com/policy"}],
|
|
},
|
|
)
|
|
|
|
dumped = json.dumps(packet, ensure_ascii=False)
|
|
assert packet["answer_type"] == "strategic_judgement"
|
|
assert "童析" not in dumped
|
|
assert "机会不在报告生成" in dumped
|
|
|
|
|
|
def test_renderer_demotes_legacy_packet_answer_type_to_runtime_general_shape() -> None:
|
|
service = make_service()
|
|
packet = service.build_customer_evidence_packet(
|
|
request=RAGAskRequest(question="这个项目的 MVP 还差多少"),
|
|
current_user=SimpleNamespace(id=7, tenant_id=1),
|
|
payload={
|
|
"customer_evidence_packet": {"answer_type": "project_mvp_readiness"},
|
|
"evidence_findings": [
|
|
{
|
|
"fact": "当前项目缺少正式任务和验收标准。",
|
|
"source": "当前项目执行态证据",
|
|
"relevance_to_goal": "判断 MVP readiness",
|
|
"confidence": "high",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
canvas = service.build_answer_canvas(packet=packet, answer="当前项目缺少正式任务和验收标准。")
|
|
|
|
assert packet["answer_type"] == "general_answer"
|
|
assert canvas["headline"] == "回答"
|
|
|
|
|
|
def test_process_packet_does_not_fabricate_renderer_steps() -> None:
|
|
service = make_service()
|
|
process = service.build_answer_process_packet(
|
|
packet={
|
|
"answer_type": "strategic_judgement",
|
|
"facts": [
|
|
{
|
|
"type": "runtime_evidence_finding",
|
|
"display_label": "外部公开来源",
|
|
"display_summary": "机会不在浅工具，而在闭环。",
|
|
"confidence": "high",
|
|
}
|
|
],
|
|
},
|
|
verification={"ok": True},
|
|
)
|
|
|
|
assert process["answer_type"] == "strategic_judgement"
|
|
assert process["steps"] == []
|
|
|
|
|
|
def test_process_packet_preserves_explicit_runtime_steps_only() -> None:
|
|
service = make_service()
|
|
process = service.build_answer_process_packet(
|
|
packet={
|
|
"answer_type": "execution_guidance",
|
|
"process_steps": [
|
|
{
|
|
"key": "inspect_project_status",
|
|
"label": "读取项目执行态证据",
|
|
"status": "completed",
|
|
"duration_ms": 120,
|
|
}
|
|
],
|
|
},
|
|
verification={"ok": True},
|
|
)
|
|
|
|
assert process["steps"][0]["key"] == "inspect_project_status"
|
|
assert process["steps"][0]["label"] == "读取项目执行态证据"
|
|
|
|
|
|
def test_customer_evidence_packet_does_not_invent_default_source_summary() -> None:
|
|
service = make_service()
|
|
packet = service.build_customer_evidence_packet(
|
|
request=RAGAskRequest(question="现在有什么项目"),
|
|
current_user=SimpleNamespace(id=7, tenant_id=1),
|
|
payload={
|
|
"grounding_status": "partially_grounded",
|
|
"evidence_findings": [
|
|
{
|
|
"fact": "当前可见项目 1 个",
|
|
"implication": "可以回答项目数量",
|
|
"source": "",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
assert packet["source_summary"] == []
|
|
|
|
|
|
def test_answer_canvas_headline_and_evidence_are_shape_based() -> None:
|
|
service = make_service()
|
|
answer = "判断：值得继续看，但要做成监督闭环，不是报告生成。"
|
|
canvas = service.build_answer_canvas(
|
|
packet={
|
|
"answer_type": "strategic_judgement",
|
|
"facts": [
|
|
{
|
|
"type": "runtime_evidence_finding",
|
|
"display_label": "外部公开来源",
|
|
"display_summary": "监管和内控压力是真实存在的。",
|
|
"confidence": "high",
|
|
"source_display_labels": ["外部公开来源"],
|
|
}
|
|
],
|
|
"limitations": ["还缺目标客户预算证据"],
|
|
},
|
|
answer=answer,
|
|
)
|
|
|
|
assert canvas["headline"] == "方向判断"
|
|
assert canvas["direct_answer"] == "值得继续看，但要做成监督闭环，不是报告生成。"
|
|
assert canvas["evidence_blocks"][0]["summary"] == "监管和内控压力是真实存在的。"
|
|
assert canvas["limitations"] == ["还缺目标客户预算证据"]
|
|
|
|
|
|
def test_packet_az_canvas_direct_answer_does_not_generate_fixed_evidence_sections() -> None:
|
|
answer = (
|
|
"这个方向可以继续看，但要先验证采购团队是否愿意为监督闭环付费。"
|
|
"已有监管压力能支持问题存在。"
|
|
"还缺预算和真实流程证据。"
|
|
"第一版先做风险提示和留痕。"
|
|
)
|
|
|
|
direct = _standardize_canvas_direct_answer(answer)
|
|
|
|
assert "**结论**" not in direct
|
|
assert "**判断依据**" not in direct
|
|
assert "这个方向可以继续看" in direct
|
|
assert "第一版先做风险提示" in direct
|
|
|
|
|
|
def test_packet_az_state_shape_answer_does_not_prefix_by_answer_type() -> None:
|
|
answer = _render_state_shape_answer(
|
|
{
|
|
"answer_type": "execution_guidance",
|
|
"facts": [
|
|
{"display_summary": "先整理加密评审材料"},
|
|
{"display_summary": "直接发布还缺测试质量证据"},
|
|
],
|
|
}
|
|
)
|
|
|
|
assert answer == "先整理加密评审材料；直接发布还缺测试质量证据。"
|
|
assert not answer.startswith("可执行建议：")
|
|
|
|
|
|
def test_packet_az_canvas_sanitizer_strips_generated_section_headings() -> None:
|
|
direct = _standardize_canvas_direct_answer(
|
|
"**结论**\n可以继续看。\n\n**判断依据**\n- 已有明确需求。\n- 还缺预算证据。"
|
|
)
|
|
|
|
assert "**结论**" not in direct
|
|
assert "**判断依据**" not in direct
|
|
assert "可以继续看" in direct
|
|
assert "已有明确需求" in direct
|