429 lines
17 KiB
Python
429 lines
17 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from app.config.ai_prompts import build_default_role_prompt_bundle
|
||
from app.services.ai_service_core import AIService
|
||
from app.services.llm_service import StructuredExtractionResult
|
||
from app.services.sstc import charter_prefill_utils
|
||
from app.services.signal.draft_generator_service import DraftGeneratorService
|
||
|
||
|
||
def test_default_role_prompt_bundle_is_pm_reviewer() -> None:
|
||
bundle = build_default_role_prompt_bundle()
|
||
assert bundle["active_role_key"] == "pm_dispatcher"
|
||
role = next((item for item in bundle["roles"] if item.get("role_key") == "pm_dispatcher"), {})
|
||
assert "立项评审" in str(role.get("role_name") or "")
|
||
assert "同名或相似项目" in str(role.get("system_prompt") or "")
|
||
assert "MVP" in str(role.get("system_prompt") or "")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_and_render_draft_includes_pm_sections(monkeypatch) -> None:
|
||
service = DraftGeneratorService(db=object(), tenant_id=1)
|
||
captured: dict[str, str] = {}
|
||
|
||
async def fake_load_members(*, world_model):
|
||
_ = world_model
|
||
return [
|
||
{"id": 7, "name": "Alice", "workload_percent": 25.0},
|
||
{"id": 8, "name": "Bob", "workload_percent": 60.0},
|
||
]
|
||
|
||
async def fake_generate_draft_payload(*, prompt):
|
||
captured["prompt"] = prompt
|
||
return {}
|
||
|
||
monkeypatch.setattr(service, "_load_members", fake_load_members)
|
||
monkeypatch.setattr(service, "_generate_draft_payload", fake_generate_draft_payload)
|
||
|
||
work_item = SimpleNamespace(
|
||
draft_version=0,
|
||
draft_content={},
|
||
draft_expires_at=None,
|
||
confirm_message_id=None,
|
||
execution_result={},
|
||
)
|
||
world_model = {
|
||
"project": {"title": "企业知识库问答平台", "description": "面向内部知识库的问答能力"},
|
||
"active_projects": [
|
||
{
|
||
"id": 51,
|
||
"name": "AI劝退助手",
|
||
"title": "AI劝退助手",
|
||
"description": "项目进展与风险提醒",
|
||
"status": "active",
|
||
"owner_id": 10,
|
||
},
|
||
{
|
||
"id": 52,
|
||
"name": "企业知识库问答平台 V2",
|
||
"title": "企业知识库问答平台 V2",
|
||
"description": "旧版知识库问答平台",
|
||
"status": "active",
|
||
"owner_id": 11,
|
||
},
|
||
],
|
||
"recent_signals": [
|
||
{"id": 91, "title": "文档检索性能优化", "signal_type": "新需求", "status": "open"},
|
||
],
|
||
"capacity": [],
|
||
}
|
||
|
||
generated = await service.generate_and_persist(
|
||
work_item=work_item,
|
||
intent_name="project.create",
|
||
message_text="我想做一个企业知识库问答项目，先帮我评估一下。",
|
||
world_model=world_model,
|
||
)
|
||
|
||
prompt = captured["prompt"]
|
||
assert "现有活跃项目（JSON）" in prompt
|
||
assert "候选相似项目（JSON）" in prompt
|
||
assert "同名/相似项目列表" in prompt
|
||
assert generated["draft"]["similar_projects"]
|
||
assert "🔎 同名 / 相似项目：" in generated["draft_text"]
|
||
assert "🚀 MVP 建议：" in generated["draft_text"]
|
||
assert "📈 前景判断：" in generated["draft_text"]
|
||
assert "🧩 需求清单：" in generated["draft_text"]
|
||
assert "❓ 还缺的信息：" in generated["draft_text"]
|
||
assert generated["draft"]["recommended_assignee_name"] == "Alice"
|
||
assert generated["draft"]["recommended_assignee_workload"] == 25.0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_and_render_draft_includes_rich_mvp_template(monkeypatch) -> None:
|
||
service = DraftGeneratorService(db=object(), tenant_id=1)
|
||
captured: dict[str, str] = {}
|
||
|
||
async def fake_load_members(*, world_model):
|
||
_ = world_model
|
||
return [{"id": 7, "name": "Alice", "workload_percent": 25.0}]
|
||
|
||
async def fake_generate_draft_payload(*, prompt):
|
||
captured["prompt"] = prompt
|
||
return {
|
||
"title": "跨境 DTC 品牌 AI 运营平台",
|
||
"summary": "帮助中国 DTC 品牌用 AI 快速完成海外市场的内容本地化、社媒运营、用户洞察。",
|
||
"core_pain_points": [
|
||
{
|
||
"pain": "内容本地化",
|
||
"status": "翻译公司贵、慢、不地道",
|
||
"opportunity": "AI 快速生成 + 人工微调",
|
||
},
|
||
{
|
||
"pain": "社媒运营",
|
||
"status": "雇海外运营成本高",
|
||
"opportunity": "AI 自动生成 + 排期发布",
|
||
},
|
||
],
|
||
"mvp_items": [
|
||
"多语言内容本地化",
|
||
"社媒内容生成器",
|
||
"用户反馈分析",
|
||
],
|
||
"technical_architecture": [
|
||
"前端：React + TypeScript + Tailwind",
|
||
"后端：FastAPI + PostgreSQL + Redis",
|
||
],
|
||
"business_model": [
|
||
{"plan": "Starter", "price": "$49/月", "features": "1 个品牌，1000 次 AI 调用/月"},
|
||
],
|
||
"validation_metrics": {
|
||
"product": ["注册转化率 > 15%", "7 日留存 > 40%"],
|
||
"market": ["找到 5 个愿意付费的种子用户"],
|
||
},
|
||
"quick_start_suggestions": {
|
||
"week_1_2": ["做 10 个目标用户的深度访谈"],
|
||
"week_3_4": ["如果反馈正面，开始写代码"],
|
||
},
|
||
"team_setup": [
|
||
{"role": "1 个产品", "notes": "兼职 CEO"},
|
||
{"role": "2 个后端", "notes": "AI 集成 + 数据处理"},
|
||
],
|
||
"similar_projects": [],
|
||
"missing_info": [],
|
||
"followup_plan": [
|
||
{"when": "明天", "action": "确认触发时机与验收口径"},
|
||
],
|
||
}
|
||
|
||
monkeypatch.setattr(service, "_load_members", fake_load_members)
|
||
monkeypatch.setattr(service, "_generate_draft_payload", fake_generate_draft_payload)
|
||
|
||
work_item = SimpleNamespace(
|
||
draft_version=0,
|
||
draft_content={},
|
||
draft_expires_at=None,
|
||
confirm_message_id=None,
|
||
execution_result={},
|
||
)
|
||
world_model = {
|
||
"project": {"title": "跨境 DTC 品牌 AI 运营平台", "description": "出海运营项目"},
|
||
"active_projects": [],
|
||
"recent_signals": [],
|
||
"capacity": [],
|
||
}
|
||
|
||
generated = await service.generate_and_persist(
|
||
work_item=work_item,
|
||
intent_name="project.create",
|
||
message_text="帮我起一个出海 DTC 品牌项目 MVP",
|
||
world_model=world_model,
|
||
)
|
||
|
||
prompt = generated["draft_text"]
|
||
assert "一、定位" in prompt
|
||
assert "二、核心痛点" in prompt
|
||
assert "三、MVP 功能范围" in prompt
|
||
assert "四、技术架构" in prompt
|
||
assert "五、商业模式" in prompt
|
||
assert "六、验证指标" in prompt
|
||
assert "七、快速启动建议" in prompt
|
||
assert "八、团队配置" in prompt
|
||
assert "内容本地化" in prompt
|
||
assert "Starter" in prompt
|
||
assert "1 个产品" in prompt
|
||
assert "positioning_one_liner" in captured["prompt"]
|
||
assert "core_pain_points" in captured["prompt"]
|
||
assert "mvp_phase_1" in captured["prompt"]
|
||
assert "technical_architecture" in captured["prompt"]
|
||
assert "business_model" in captured["prompt"]
|
||
assert "validation_metrics" in captured["prompt"]
|
||
assert "quick_start_suggestions" in captured["prompt"]
|
||
assert "team_setup" in captured["prompt"]
|
||
assert any(item in generated["draft"]["requirement_items"][0]["title"] for item in ("内容本地化", "多语言内容本地化"))
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_charter_seed_uses_project_context(monkeypatch) -> None:
|
||
captured: dict[str, str] = {}
|
||
|
||
async def fake_load_project_context(*, db, tenant_id, project_id):
|
||
_ = db, tenant_id, project_id
|
||
return {
|
||
"id": 7,
|
||
"name": "kb-v2",
|
||
"title": "企业知识库问答平台",
|
||
"description": "内部知识库问答升级",
|
||
"status": "active",
|
||
"owner_id": 10,
|
||
}
|
||
|
||
async def fake_load_active_projects_context(*, db, tenant_id, project_id, limit=10):
|
||
_ = db, tenant_id, project_id, limit
|
||
return [
|
||
{
|
||
"id": 51,
|
||
"name": "ai-warn",
|
||
"title": "AI劝退助手",
|
||
"description": "项目进展与风险提醒",
|
||
"status": "active",
|
||
"owner_id": 11,
|
||
}
|
||
]
|
||
|
||
async def fake_generate_structured_object(prompt, **kwargs):
|
||
captured["prompt"] = prompt
|
||
payload = {
|
||
"title": "内部知识库问答升级",
|
||
"summary": "围绕内部知识库问答能力升级",
|
||
"background": "背景说明",
|
||
"objectives": ["提升检索质量"],
|
||
"scope": "范围说明",
|
||
"success_metrics": ["命中率提升"],
|
||
"risks": ["重复建设风险"],
|
||
"dependencies": ["知识库索引"],
|
||
"primary_model": "gpt",
|
||
"fallback_model": "gpt-mini",
|
||
}
|
||
return StructuredExtractionResult(
|
||
payload=payload,
|
||
source_field="message.content",
|
||
provider="glm",
|
||
model="qwen-plus",
|
||
raw_text=json.dumps(payload, ensure_ascii=False),
|
||
validation="ok",
|
||
)
|
||
|
||
monkeypatch.setattr(charter_prefill_utils, "_load_charter_project_context", fake_load_project_context)
|
||
monkeypatch.setattr(charter_prefill_utils, "_load_active_projects_context", fake_load_active_projects_context)
|
||
monkeypatch.setattr(charter_prefill_utils.llm_service, "generate_structured_object", fake_generate_structured_object)
|
||
|
||
result = await charter_prefill_utils._generate_charter_seed(
|
||
db=object(),
|
||
tenant_id=1,
|
||
project_id=7,
|
||
signal_context="AI 劝退诉求：请升级内部知识库问答能力。",
|
||
)
|
||
|
||
prompt = captured["prompt"]
|
||
assert "当前项目上下文" in prompt
|
||
assert "企业知识库问答平台" in prompt
|
||
assert "现有活跃项目（JSON）" in prompt
|
||
assert "AI劝退助手" in prompt
|
||
assert result["summary"] == "围绕内部知识库问答能力升级"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_charter_from_signal_includes_project_context(monkeypatch) -> None:
|
||
service = AIService()
|
||
captured: dict[str, object] = {}
|
||
|
||
async def fake_generate_structured_object(prompt, **kwargs):
|
||
captured["prompt"] = prompt
|
||
captured["kwargs"] = kwargs
|
||
payload = {
|
||
"title": "内部知识库问答升级",
|
||
"summary": "围绕内部知识库问答能力升级",
|
||
"background": "背景说明",
|
||
"objectives": ["提升检索质量"],
|
||
"scope": "范围说明",
|
||
"success_metrics": ["命中率提升"],
|
||
"risks": ["重复建设风险"],
|
||
"dependencies": ["知识库索引"],
|
||
"primary_model": "gpt",
|
||
"fallback_model": "gpt-mini",
|
||
}
|
||
return StructuredExtractionResult(
|
||
payload=payload,
|
||
source_field="message.content",
|
||
provider="glm",
|
||
model="qwen-plus",
|
||
raw_text=json.dumps(payload, ensure_ascii=False),
|
||
validation="ok",
|
||
)
|
||
|
||
monkeypatch.setattr(charter_prefill_utils.llm_service, "generate_structured_object", fake_generate_structured_object)
|
||
|
||
result = await service.generate_charter_from_signal(
|
||
"AI 劝退诉求：请升级内部知识库问答能力。",
|
||
tenant_id=1,
|
||
db=object(),
|
||
project_id=7,
|
||
project_context="当前项目上下文：企业知识库问答平台",
|
||
active_projects_json='[{"id":51,"title":"AI劝退助手"}]',
|
||
)
|
||
|
||
prompt = str(captured["prompt"] or "")
|
||
kwargs = captured["kwargs"] if isinstance(captured.get("kwargs"), dict) else {}
|
||
assert "当前项目上下文：企业知识库问答平台" in prompt
|
||
assert "现有活跃项目（JSON）" in prompt
|
||
assert kwargs.get("project_id") == 7
|
||
assert kwargs.get("use_prompt_profile") is False
|
||
assert result["summary"] == "围绕内部知识库问答能力升级"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_ai_first_charter_content_marks_missing_evidence_without_fallback(monkeypatch) -> None:
|
||
from app.services.sstc import charter_prefill_builder
|
||
|
||
async def fake_generate_charter_seed(**kwargs):
|
||
_ = kwargs
|
||
return {
|
||
"modules": {
|
||
"module_1": {
|
||
"problem_statement": {
|
||
"value": "知识检索命中率偏低",
|
||
"evidence": [
|
||
{
|
||
"signal_id": 11,
|
||
"summary": "客服重复追问较多",
|
||
"quote": "用户多次重复提问相同内容",
|
||
"reason": "直接体现当前问题",
|
||
"confidence": 0.93,
|
||
}
|
||
],
|
||
},
|
||
"goal_statement": {
|
||
"value": "提升检索命中率并降低重复提问",
|
||
"evidence": [
|
||
{
|
||
"signal_id": 11,
|
||
"summary": "客服重复追问较多",
|
||
"quote": "用户多次重复提问相同内容",
|
||
"reason": "体现目标方向",
|
||
"confidence": 0.9,
|
||
}
|
||
],
|
||
},
|
||
"kpis": {
|
||
"value": "",
|
||
"evidence": [],
|
||
"status": "evidence_insufficient",
|
||
},
|
||
},
|
||
"module_2": {
|
||
"in_scope": {
|
||
"value": "问答检索增强\n知识补录流程",
|
||
"evidence": [
|
||
{
|
||
"signal_id": 11,
|
||
"summary": "客服重复追问较多",
|
||
"quote": "用户多次重复提问相同内容",
|
||
"reason": "范围来自当前信号",
|
||
"confidence": 0.88,
|
||
}
|
||
],
|
||
},
|
||
"out_of_scope": {
|
||
"value": "权限体系重构",
|
||
"evidence": [
|
||
{
|
||
"signal_id": 12,
|
||
"summary": "当前版本不涉及权限系统",
|
||
"quote": "本期不动权限模块",
|
||
"reason": "明确排除项",
|
||
"confidence": 0.84,
|
||
}
|
||
],
|
||
},
|
||
"alignment_qa": {
|
||
"value": "",
|
||
"evidence": [],
|
||
"status": "evidence_insufficient",
|
||
},
|
||
},
|
||
},
|
||
"readiness": {
|
||
"ready_for_confirm": False,
|
||
"blocking_fields": ["module_1.kpis", "module_2.alignment_qa"],
|
||
"blocking_reasons": ["关键 KPI 缺少证据", "固定对齐问答未生成"],
|
||
},
|
||
"derived_artifacts": {
|
||
"task_seeds": [
|
||
{
|
||
"title": "梳理检索指标与验收口径",
|
||
"description": "根据模块1目标与 KPI 补齐任务种子",
|
||
"evidence_refs": ["module_1.goal_statement"],
|
||
}
|
||
],
|
||
"verification_metrics": [],
|
||
"risk_gates": [],
|
||
},
|
||
}
|
||
|
||
monkeypatch.setattr(charter_prefill_builder, "_generate_charter_seed", fake_generate_charter_seed)
|
||
|
||
result = await charter_prefill_builder._build_ai_first_charter_content(
|
||
db=object(),
|
||
tenant_id=1,
|
||
project_id=7,
|
||
signal_rows=[],
|
||
signal_ids=[11, 12],
|
||
raw_text="请升级内部知识库问答能力。",
|
||
)
|
||
|
||
assert result["modules"]["module_1"]["kpis"] == ""
|
||
assert result["modules"]["module_2"]["alignment_qa"] == ""
|
||
assert "需求到上线周期缩短" not in json.dumps(result, ensure_ascii=False)
|
||
assert result["field_status"]["module_1"]["kpis"] == "evidence_insufficient"
|
||
assert result["field_status"]["module_2"]["alignment_qa"] == "evidence_insufficient"
|
||
assert result["field_evidence"]["module_1"]["problem_statement"][0]["signal_id"] == 11
|
||
assert result["readiness"]["ready_for_confirm"] is False
|
||
assert "module_1.kpis" in result["readiness"]["blocking_fields"]
|
||
assert result["derived_artifacts"]["task_seeds"][0]["title"] == "梳理检索指标与验收口径"
|