307 lines
12 KiB
Python
307 lines
12 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.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(prompt, **kwargs):
|
||
captured["prompt"] = prompt
|
||
return json.dumps(
|
||
{
|
||
"title": "内部知识库问答升级",
|
||
"summary": "围绕内部知识库问答能力升级",
|
||
"background": "背景说明",
|
||
"objectives": ["提升检索质量"],
|
||
"scope": "范围说明",
|
||
"success_metrics": ["命中率提升"],
|
||
"risks": ["重复建设风险"],
|
||
"dependencies": ["知识库索引"],
|
||
"primary_model": "gpt",
|
||
"fallback_model": "gpt-mini",
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
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", fake_generate)
|
||
|
||
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_invoke(self, prompt, **kwargs):
|
||
_ = self
|
||
captured["prompt"] = prompt
|
||
captured["kwargs"] = kwargs
|
||
return json.dumps(
|
||
{
|
||
"title": "内部知识库问答升级",
|
||
"summary": "围绕内部知识库问答能力升级",
|
||
"background": "背景说明",
|
||
"objectives": ["提升检索质量"],
|
||
"scope": "范围说明",
|
||
"success_metrics": ["命中率提升"],
|
||
"risks": ["重复建设风险"],
|
||
"dependencies": ["知识库索引"],
|
||
"primary_model": "gpt",
|
||
"fallback_model": "gpt-mini",
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
monkeypatch.setattr(AIService, "_invoke_with_retry", fake_invoke)
|
||
|
||
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 result["summary"] == "围绕内部知识库问答能力升级"
|