PMFlow/backend/tests/test_draft_generator_servic...

667 lines
26 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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, is_revision=False):
_ = is_revision
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 "🚀 建议先做" 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, is_revision=False):
_ = is_revision
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 "✅ 当前判断" 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_and_persist_recovers_rich_logged_payload_when_structured_payload_is_sparse(monkeypatch) -> None:
service = DraftGeneratorService(db=object(), tenant_id=1)
captured: dict[str, object] = {}
async def fake_load_members(*, world_model):
_ = world_model
return [{"id": 7, "name": "Alice", "workload_percent": 25.0}]
async def fake_generate_structured_object(prompt, **kwargs):
_ = prompt
captured["max_tokens"] = kwargs.get("max_tokens")
return StructuredExtractionResult(
payload={
"title": "",
"summary": "占位摘要",
"mvp_items": [],
"followup_plan": [],
},
source_field="tool_calls.function.arguments",
provider="glm",
model="qwen-plus",
raw_text='{"title":"","summary":"占位摘要","mvp_items":[],"followup_plan":[]}',
validation="ok",
)
async def fake_recover_payload_from_logged_response(*, prompt, current_payload):
_ = prompt, current_payload
return {
"title": "AI直播复盘助手",
"summary": "面向跨境 DTC 团队的 AI 直播复盘工具",
"positioning_one_liner": "自动解析直播数据并输出复盘建议",
"target_customers": "跨境 DTC 品牌运营团队",
"core_pain_points": [
{
"pain": "人工复盘慢",
"status": "单场耗时超过 2 小时",
"opportunity": "压缩到 10 分钟内",
}
],
"mvp_items": ["接入 TikTok Shop", "对接 Shopify", "生成 PDF 复盘简报"],
"mvp_phase_1": ["打通 TikTok Shop 与 Shopify 数据"],
"team_setup": [{"role": "PM", "notes": "负责验收口径"}],
"followup_plan": [{"when": "明天", "action": "确认验收口径"}],
}
monkeypatch.setattr(service, "_load_members", fake_load_members)
monkeypatch.setattr(
"app.services.signal.draft_generator_service.llm_service.generate_structured_object",
fake_generate_structured_object,
)
monkeypatch.setattr(
service,
"_recover_payload_from_logged_response",
fake_recover_payload_from_logged_response,
raising=False,
)
work_item = SimpleNamespace(
draft_version=0,
draft_content={},
draft_expires_at=None,
confirm_message_id=None,
execution_result={},
)
generated = await service.generate_and_persist(
work_item=work_item,
intent_name="project.create",
message_text="帮我起一个 AI 直播复盘助手项目",
world_model={
"project": {"title": "AI直播复盘助手", "description": "跨境 DTC 直播复盘"},
"active_projects": [],
"recent_signals": [],
"capacity": [],
},
)
assert generated["draft"]["title"] == "AI直播复盘助手"
assert generated["draft"]["target_customers"] == "跨境 DTC 品牌运营团队"
assert generated["draft"]["team_setup"][0]["role"] == "PM"
assert "人工复盘慢" in generated["draft_text"]
assert captured["max_tokens"] == 2_200
@pytest.mark.asyncio
async def test_generate_draft_payload_escalates_token_budget_low_to_high(monkeypatch) -> None:
service = DraftGeneratorService(db=object(), tenant_id=1)
token_attempts: list[int] = []
async def fake_generate_structured_object(prompt, **kwargs):
_ = prompt
token_attempts.append(int(kwargs.get("max_tokens") or 0))
if len(token_attempts) == 1:
return StructuredExtractionResult(
payload={
"title": "需求草稿",
"summary": "占位摘要",
"mvp_items": [],
"followup_plan": [],
},
source_field="message.content",
provider="dashscope",
model="qwen-plus",
raw_text='{"title":"需求草稿","summary":"占位摘要","mvp_items":[],"followup_plan":[]}',
validation="ok",
)
return StructuredExtractionResult(
payload={
"title": "AI直播复盘助手",
"summary": "面向跨境 DTC 团队的 AI 直播复盘工具",
"target_customers": "跨境 DTC 品牌运营团队",
"core_pain_points": [{"pain": "人工复盘慢", "status": "高频", "opportunity": "10 分钟内输出报告"}],
"mvp_items": ["接入 TikTok Shop", "生成复盘报告", "沉淀行动清单"],
"team_setup": [{"role": "PM", "notes": "负责验收"}],
},
source_field="message.content",
provider="dashscope",
model="qwen-plus",
raw_text='{"title":"AI直播复盘助手"}',
validation="ok",
)
async def fake_recover_payload_from_logged_response(*, prompt, current_payload):
_ = prompt
return current_payload
monkeypatch.setattr(
"app.services.signal.draft_generator_service.llm_service.generate_structured_object",
fake_generate_structured_object,
)
monkeypatch.setattr(service, "_recover_payload_from_logged_response", fake_recover_payload_from_logged_response)
payload = await service._generate_draft_payload(prompt="demo")
assert payload["title"] == "AI直播复盘助手"
assert token_attempts[:2] == [2_200, 4_000]
@pytest.mark.asyncio
async def test_generate_draft_payload_starts_revision_attempts_at_higher_budget(monkeypatch) -> None:
service = DraftGeneratorService(db=object(), tenant_id=1)
token_attempts: list[int] = []
async def fake_generate_structured_object(prompt, **kwargs):
_ = prompt
token_attempts.append(int(kwargs.get("max_tokens") or 0))
return StructuredExtractionResult(
payload={
"title": "跨境复盘助手",
"summary": "面向跨境 DTC 团队的 AI 直播复盘工具",
"target_customers": "跨境 DTC 品牌运营团队",
"core_pain_points": [{"pain": "人工复盘慢", "status": "高频", "opportunity": "10 分钟内输出报告"}],
"mvp_items": ["接入 TikTok Shop", "生成复盘报告", "沉淀行动清单"],
"team_setup": [{"role": "PM", "notes": "负责验收"}],
},
source_field="message.content",
provider="dashscope",
model="qwen-plus",
raw_text='{"title":"跨境复盘助手"}',
validation="ok",
)
async def fake_recover_payload_from_logged_response(*, prompt, current_payload):
_ = prompt
return current_payload
monkeypatch.setattr(
"app.services.signal.draft_generator_service.llm_service.generate_structured_object",
fake_generate_structured_object,
)
monkeypatch.setattr(service, "_recover_payload_from_logged_response", fake_recover_payload_from_logged_response)
payload = await service._generate_draft_payload(prompt="demo", is_revision=True)
assert payload["title"] == "跨境复盘助手"
assert token_attempts == [4_000]
@pytest.mark.asyncio
async def test_revision_preserves_previous_draft_when_new_candidate_regresses(monkeypatch) -> None:
service = DraftGeneratorService(db=object(), tenant_id=1)
async def fake_load_members(*, world_model):
_ = world_model
return [{"id": 3, "name": "AI工程师", "workload_percent": 0.0}]
async def fake_generate_draft_payload(*, prompt, is_revision=False):
_ = is_revision
_ = prompt
return {
"title": "需求草稿",
"summary": "占位摘要",
"mvp_items": [],
"core_pain_points": [],
"team_setup": [],
"followup_plan": [],
}
monkeypatch.setattr(service, "_load_members", fake_load_members)
monkeypatch.setattr(service, "_generate_draft_payload", fake_generate_draft_payload)
previous_draft = {
"title": "AI直播复盘助手",
"summary": "面向跨境 DTC 团队的 AI 直播复盘工具",
"version": 1,
"mvp_items": ["接入 TikTok Shop", "生成复盘报告", "沉淀行动清单"],
"team_setup": [{"role": "PM", "notes": "负责验收"}],
"core_pain_points": [{"pain": "人工复盘慢", "status": "高频", "opportunity": "10 分钟内输出报告"}],
"target_customers": "跨境 DTC 品牌运营团队",
"followup_plan": [{"when": "明天", "action": "确认验收口径"}],
}
work_item = SimpleNamespace(
draft_version=1,
draft_content=previous_draft,
draft_expires_at=None,
confirm_message_id=None,
execution_result={},
)
generated = await service.generate_and_persist(
work_item=work_item,
intent_name="project.create",
message_text="帮我起一个 AI 直播复盘助手项目",
world_model={
"project": {"title": "AI直播复盘助手", "description": "跨境 DTC 直播复盘"},
"active_projects": [],
"recent_signals": [],
"capacity": [],
},
revision_note="目标客户改成北美年营收 300 万美元以上的 DTC 品牌运营团队",
)
assert generated["draft"]["title"] == "AI直播复盘助手"
assert generated["draft"]["team_setup"][0]["role"] == "PM"
assert generated["draft"]["core_pain_points"][0]["pain"] == "人工复盘慢"
@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"] == "梳理检索指标与验收口径"