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"] == "梳理检索指标与验收口径"