from __future__ import annotations from typing import Any from .ai_prompts_catalog import __all__ as PROMPT_CATALOG_EXPORTS PROMPT_FAMILY_CLASSIFICATION = "classification" PROMPT_FAMILY_EXTRACTION = "extraction" PROMPT_FAMILY_PLANNING = "planning" PROMPT_FAMILY_RESPONSE = "response" PROMPT_FAMILY_GOVERNANCE = "governance" PROMPT_FAMILY_ALLOWED = { PROMPT_FAMILY_CLASSIFICATION, PROMPT_FAMILY_EXTRACTION, PROMPT_FAMILY_PLANNING, PROMPT_FAMILY_RESPONSE, PROMPT_FAMILY_GOVERNANCE, } PROMPT_FAMILY_KEYWORD_RULES: tuple[tuple[str, str], ...] = ( ("CLASSIFICATION", PROMPT_FAMILY_CLASSIFICATION), ("EXTRACTION", PROMPT_FAMILY_EXTRACTION), ("PREFILL", PROMPT_FAMILY_EXTRACTION), ("INTENT", PROMPT_FAMILY_CLASSIFICATION), ("TASK_ORCHESTRATION", PROMPT_FAMILY_PLANNING), ("PLANNER", PROMPT_FAMILY_PLANNING), ("GENERATE", PROMPT_FAMILY_PLANNING), ("DECOMPOSE", PROMPT_FAMILY_PLANNING), ("RECOMMEND", PROMPT_FAMILY_PLANNING), ("DRAFT", PROMPT_FAMILY_PLANNING), ("TASK_BREAKDOWN", PROMPT_FAMILY_PLANNING), ("QA", PROMPT_FAMILY_RESPONSE), ("REPLY", PROMPT_FAMILY_RESPONSE), ) PROMPT_FAMILY_SCENARIO_RULES: tuple[tuple[str, str], ...] = ( ("分类", PROMPT_FAMILY_CLASSIFICATION), ("抽取", PROMPT_FAMILY_EXTRACTION), ("预填", PROMPT_FAMILY_EXTRACTION), ("规划", PROMPT_FAMILY_PLANNING), ("编排", PROMPT_FAMILY_PLANNING), ("草案", PROMPT_FAMILY_PLANNING), ("拆解", PROMPT_FAMILY_PLANNING), ("问答", PROMPT_FAMILY_RESPONSE), ("回复", PROMPT_FAMILY_RESPONSE), ) RESPONSIBILITY_SCOPE_BY_FAMILY: dict[str, str] = { PROMPT_FAMILY_CLASSIFICATION: "分类层：只负责识别请求类型、归属和优先级，不生成执行方案。", PROMPT_FAMILY_EXTRACTION: "抽取层：只负责结构化提取实体、槽位和证据，不做规划决策。", PROMPT_FAMILY_PLANNING: "规划层：只负责生成可执行规划、任务和负责人建议，不承担最终对话呈现。", PROMPT_FAMILY_RESPONSE: "呈现层：只负责面向用户输出结论与建议，不直接做写操作决策。", PROMPT_FAMILY_GOVERNANCE: "治理层：只负责约束、评估、同步与风控，不承担主对话生成。", } PROMPT_PRIORITY_BY_FAMILY: dict[str, str] = { PROMPT_FAMILY_CLASSIFICATION: "high", PROMPT_FAMILY_EXTRACTION: "high", PROMPT_FAMILY_PLANNING: "high", PROMPT_FAMILY_RESPONSE: "high", PROMPT_FAMILY_GOVERNANCE: "medium", } PROMPT_AUDIENCE_BY_FAMILY: dict[str, str] = { PROMPT_FAMILY_CLASSIFICATION: "signal_router", PROMPT_FAMILY_EXTRACTION: "signal_ingestion", PROMPT_FAMILY_PLANNING: "planning_orchestrator", PROMPT_FAMILY_RESPONSE: "assistant_response", PROMPT_FAMILY_GOVERNANCE: "risk_budget_guardrail", } def _normalize_family(prompt_family: str) -> str: value = str(prompt_family or "").strip().lower() if value in PROMPT_FAMILY_ALLOWED: return value return PROMPT_FAMILY_GOVERNANCE def _derive_prompt_family(prompt_key: str, business_scenario: str) -> str: key_text = str(prompt_key or "").strip().upper() scenario_text = str(business_scenario or "").strip() for keyword, family in PROMPT_FAMILY_KEYWORD_RULES: if keyword in key_text: return family for keyword, family in PROMPT_FAMILY_SCENARIO_RULES: if keyword in scenario_text: return family return PROMPT_FAMILY_GOVERNANCE def _derive_responsibility_scope(prompt_family: str) -> str: family = _normalize_family(prompt_family) return RESPONSIBILITY_SCOPE_BY_FAMILY[family] def _derive_priority(prompt_family: str) -> str: family = _normalize_family(prompt_family) return PROMPT_PRIORITY_BY_FAMILY[family] def _derive_audience(prompt_family: str) -> str: family = _normalize_family(prompt_family) return PROMPT_AUDIENCE_BY_FAMILY[family] def _entry( prompt_key: str, *, module: str, business_scenario: str, input_contract: str, output_contract: str, call_sites: tuple[str, ...] = (), prompt_family: str | None = None, responsibility_scope: str | None = None, priority: str | None = None, audience: str | None = None, ) -> dict[str, Any]: resolved_family = _normalize_family(prompt_family or _derive_prompt_family(prompt_key, business_scenario)) resolved_scope = str(responsibility_scope or "").strip() or _derive_responsibility_scope(resolved_family) resolved_priority = str(priority or "").strip().lower() or _derive_priority(resolved_family) resolved_audience = str(audience or "").strip() or _derive_audience(resolved_family) return { "prompt_key": prompt_key, "module": module, "business_scenario": business_scenario, "call_sites": tuple(str(site) for site in call_sites if str(site or "").strip()), "input_contract": input_contract, "output_contract": output_contract, "prompt_family": resolved_family, "responsibility_scope": resolved_scope, "priority": resolved_priority, "audience": resolved_audience, } JSON_INPUT = "输入为结构化 JSON 上下文。" JSON_OUTPUT = "输出严格 JSON 对象，不要 Markdown。" JSON_LIST_OUTPUT = "输出严格 JSON 数组，不要 Markdown。" QA_INPUT = "输入为问题文本 + 业务上下文。" QA_OUTPUT = "输出自然语言答复。" PLAN_INPUT = "输入为信号 / 项目 / 宪章 / 成员上下文。" PLAN_OUTPUT = "输出可执行规划 JSON。" SCORE_INPUT = "输入为对比对象 + 评分上下文。" SCORE_OUTPUT = "输出评分或建议 JSON。" FRAGMENT_INPUT = "输入为 prompt 片段占位内容。" FRAGMENT_OUTPUT = "输出为可拼接的 prompt 片段。" DEEP_GUARD_CALL_SITES = ("backend/app/services/adaptive_rag_service/mixin_deep_guard_answer.py",) AI_BUSINESS_TASKS_CALL_SITES = ("backend/app/services/ai_service_core/mixin_business_tasks.py",) AI_BUSINESS_ANALYSIS_CALL_SITES = ("backend/app/services/ai_service_core/mixin_business_analysis.py",) FEISHU_EXTRACT_CALL_SITES = ("backend/app/services/integrations/feishu_signal_pipeline/mixin_extraction.py",) FEISHU_REPLY_CALL_SITES = ("backend/app/services/integrations/feishu_integration_service/mixin_reply_llm_core.py",) FEISHU_PIPELINE_CALL_SITES = ("backend/app/services/integrations/feishu_signal_pipeline/mixin_task_orchestration_main.py",) FEISHU_SIGNAL_CLASSIFY_CALL_SITES = ("backend/app/services/signal/signal_capture_service/mixin_analysis_normalize.py",) SIGNAL_SOURCE_CALL_SITES = ("backend/app/services/signal/signal_source_orchestration/mixin_registry_execute.py",) DRAFT_GENERATOR_CALL_SITES = ("backend/app/services/signal/draft_generator_service.py",) RAG_INTENT_CALL_SITES = ("backend/app/services/adaptive_rag_service/mixin_core_intent.py",) RAG_DB_CALL_SITES = ("backend/app/services/adaptive_rag_service/mixin_fast_answer_semantic_p1.py",) SIGNAL_PLANNER_CALL_SITES = ("backend/app/services/signal/signal_planner/mixin_prompt.py", "backend/app/services/signal/signal_planner/mixin_analyze.py") SKILL_SCORE_CALL_SITES = ("backend/app/services/skill/skill_score_service.py",) WORKLOAD_CALL_SITES = ("backend/app/services/workload/ai_optimize.py",) RECIPE_CALL_SITES = ("backend/app/services/recipe/service.py",) PMO_INTENT_CALL_SITES = ("backend/app/services/ai_pmo/ai_pmo_kernel/mixins/intent/ai_pmo_kernel_mixins_intent_llm.py",) PMO_BATCH_CALL_SITES = ("backend/app/services/ai_pmo/ai_pmo_typed_executor_discovery/mixin_analysis.py",) def _part_01_entries() -> tuple[dict[str, Any], ...]: return ( _entry( "PROMPT_IMPACT_ANALYSIS", module="part_01", business_scenario="需求变更影响分析", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含影响结论、风险、建议。", ), _entry( "PROMPT_STRATEGIC_ANALYSIS_HEADER", module="part_01", business_scenario="战略分析上下文片段", input_contract=FRAGMENT_INPUT, output_contract=FRAGMENT_OUTPUT, ), _entry( "PROMPT_RAG_CONTEXT_HEADER", module="part_01", business_scenario="RAG 上下文 header 片段", input_contract=FRAGMENT_INPUT, output_contract=FRAGMENT_OUTPUT, ), _entry( "PROMPT_RAG_CONTEXT_ITEM", module="part_01", business_scenario="RAG 上下文 item 片段", input_contract=FRAGMENT_INPUT, output_contract=FRAGMENT_OUTPUT, ), _entry( "PROMPT_CODE_AUDIT", module="part_01", business_scenario="代码审计", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含问题、证据、风险与修复建议。", ), _entry( "PROMPT_CONTEXT_INSIGHTS", module="part_01", business_scenario="上下文洞察", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含关键洞察与下一步建议。", ), _entry( "PROMPT_FEISHU_HUMANIZED_REPLY", module="part_01", business_scenario="飞书回复润色", input_contract=QA_INPUT, output_contract=QA_OUTPUT, call_sites=FEISHU_REPLY_CALL_SITES, ), _entry( "PROMPT_FEISHU_PROJECT_SIGNAL_CLASSIFICATION", module="part_01", business_scenario="项目信号分类", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含分类、优先级与推荐动作。", call_sites=FEISHU_SIGNAL_CLASSIFY_CALL_SITES, ), _entry( "PROMPT_FEISHU_LEVEL_CLASSIFICATION", module="part_01", business_scenario="飞书消息层级分类", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含层级与判断原因。", call_sites=FEISHU_EXTRACT_CALL_SITES, ), ) def _part_02_entries() -> tuple[dict[str, Any], ...]: return ( _entry( "PROMPT_FEISHU_SIGNAL_EXTRACTION", module="part_02", business_scenario="飞书信号结构化抽取", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含标题、摘要、优先级、匹配项目和建议动作。", call_sites=FEISHU_EXTRACT_CALL_SITES, ), _entry( "PROMPT_FEISHU_PROJECT_PREFILL", module="part_02", business_scenario="飞书项目预填", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含项目名、描述、负责人、时间线与置信度。", call_sites=FEISHU_EXTRACT_CALL_SITES, ), _entry( "PROMPT_FEISHU_TASK_ORCHESTRATION", module="part_02", business_scenario="飞书任务编排与负责人建议", input_contract="输入 project_json / charter_json / signal_json / members_json。", output_contract="输出 JSON 对象，包含任务列表、负责人建议与置信度。", call_sites=FEISHU_PIPELINE_CALL_SITES, ), _entry( "PROMPT_SIGNAL_SOURCE_ONBOARDING", module="part_02", business_scenario="信号源接入规划", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含是否接入、信号源信息、角色分工与执行指令。", call_sites=SIGNAL_SOURCE_CALL_SITES, ), _entry( "PROMPT_AI_WORKER_SIGNAL_SOURCE_EXECUTION", module="part_02", business_scenario="信号源接入 worker 执行", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含交付物、风险与下一步。", call_sites=SIGNAL_SOURCE_CALL_SITES, ), _entry( "PROMPT_AI_PMO_INTENT_CLASSIFICATION", module="part_02", business_scenario="AI PMO 意图识别", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含 family / action / intent / confidence。", call_sites=PMO_INTENT_CALL_SITES, ), _entry( "PROMPT_AI_PMO_SIGNAL_BATCH_ANALYSIS", module="part_02", business_scenario="AI PMO 批量信号分析", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含 signals 数组。", call_sites=PMO_BATCH_CALL_SITES, ), _entry( "PROMPT_DRAFT_GENERATOR", module="part_02", business_scenario="立项草案生成", input_contract="输入 request / project / active_projects / similar_projects / charter / members / recent_signals。", output_contract="输出 JSON 对象，包含项目草案、重复建设判断与风险。", call_sites=DRAFT_GENERATOR_CALL_SITES, ), _entry( "PROMPT_RAG_INTENT_CLASSIFICATION", module="part_02", business_scenario="RAG 意图识别", input_contract=QA_INPUT, output_contract="输出 JSON 对象，包含 intent / confidence / reason。", call_sites=RAG_INTENT_CALL_SITES, ), ) def _part_03_entries() -> tuple[dict[str, Any], ...]: return ( _entry( "PROMPT_RAG_ANSWER", module="part_03", business_scenario="RAG 问答生成", input_contract=QA_INPUT, output_contract=QA_OUTPUT, call_sites=DEEP_GUARD_CALL_SITES, ), _entry( "PROMPT_DECISION_QA", module="part_03", business_scenario="决策类问答", input_contract=QA_INPUT, output_contract="输出自然语言结论，必要时包含取舍理由。", call_sites=DEEP_GUARD_CALL_SITES, ), _entry( "PROMPT_RAG_DB_PROJECT_QA", module="part_03", business_scenario="项目知识库问答", input_contract=JSON_INPUT, output_contract=QA_OUTPUT, call_sites=RAG_DB_CALL_SITES, ), _entry( "PROMPT_GENERATE_CHARTER", module="part_03", business_scenario="宪章草案生成", input_contract="输入 request / project / active_projects / similar_projects / signal_context。", output_contract="输出 JSON 对象，包含项目宪章草案与风险。", call_sites=("backend/app/services/ai_service_core/mixin_business_tasks.py", "backend/app/services/sstc/charter_prefill_utils.py"), ), _entry( "PROMPT_DECOMPOSE_TASKS", module="part_03", business_scenario="任务拆解", input_contract="输入 request / project / charter / modules 的结构化上下文。", output_contract="输出 JSON 数组，包含任务名称、描述、工时、优先级和技能标签。", call_sites=("backend/app/services/ai_service_core/mixin_business_tasks.py", "backend/app/api/tasks.py", "backend/app/api/sstc_api/tasks_ai_breakdown.py"), ), _entry( "PROMPT_RECOMMEND_ASSIGNEES", module="part_03", business_scenario="负责人推荐", input_contract=JSON_INPUT, output_contract="输出 JSON 数组，包含候选人、匹配理由和负载信息。", call_sites=AI_BUSINESS_TASKS_CALL_SITES, ), _entry( "PROMPT_ALIGNMENT_SCORE", module="part_03", business_scenario="对齐度评分", input_contract=SCORE_INPUT, output_contract=SCORE_OUTPUT, call_sites=AI_BUSINESS_ANALYSIS_CALL_SITES, ), _entry( "PROMPT_COGNITIVE_CONFLICTS", module="part_03", business_scenario="认知冲突识别", input_contract=SCORE_INPUT, output_contract=SCORE_OUTPUT, call_sites=AI_BUSINESS_ANALYSIS_CALL_SITES, ), _entry( "PROMPT_EVALUATE_MODEL_REQUIREMENT", module="part_03", business_scenario="模型需求评估", input_contract=SCORE_INPUT, output_contract=SCORE_OUTPUT, call_sites=AI_BUSINESS_ANALYSIS_CALL_SITES, ), _entry( "PROMPT_ANALYZE_CAPABILITY_GAP", module="part_03", business_scenario="能力缺口分析", input_contract=SCORE_INPUT, output_contract=SCORE_OUTPUT, call_sites=AI_BUSINESS_ANALYSIS_CALL_SITES, ), _entry( "PROMPT_ANALYZE_COST_ROI", module="part_03", business_scenario="成本收益分析", input_contract=SCORE_INPUT, output_contract=SCORE_OUTPUT, call_sites=AI_BUSINESS_ANALYSIS_CALL_SITES, ), ) def _part_04_entries() -> tuple[dict[str, Any], ...]: return ( _entry( "PROMPT_ANALYZE_COST_OPTIMIZATION", module="part_04", business_scenario="成本优化分析", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含降级、缓存、批处理与节省测算。", call_sites=AI_BUSINESS_ANALYSIS_CALL_SITES, ), _entry( "PROMPT_ASSESS_MODEL_UPDATE_IMPACT", module="part_04", business_scenario="模型升级影响评估", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含成本变化、兼容风险与建议。", call_sites=("backend/app/services/model/model_capability/service.py",), ), _entry( "PROMPT_EXPERIMENT_SYNC_DECISION", module="part_04", business_scenario="实验结果同步决策", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含置信度、是否自动同步与原因。", call_sites=("backend/app/services/ai_pmo/ai_pmo_cadence/mixin_builders.py",), ), _entry( "PROMPT_SKILL_SCORE_UPDATE", module="part_04", business_scenario="技能评分更新", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含技能增量列表。", call_sites=SKILL_SCORE_CALL_SITES, ), _entry( "PROMPT_WORKLOAD_OPTIMIZATION", module="part_04", business_scenario="任务负载优化", input_contract=JSON_INPUT, output_contract="输出 JSON 对象，包含重分配建议。", call_sites=WORKLOAD_CALL_SITES, ), _entry( "PROMPT_ANALYZE_FOOD_CALORIES", module="part_04", business_scenario="食物热量估算", input_contract=QA_INPUT, output_contract="输出 JSON 对象，包含食物名、热量、营养和不确定因素。", call_sites=RECIPE_CALL_SITES, ), _entry( "PROMPT_SIGNAL_PLANNER", module="part_04", business_scenario="信号规划 / 任务编排建议", input_contract="输入 signal / project / charter_json / recent_signals / member_profiles。", output_contract="输出 JSON 对象，包含 summary、priority、负责人建议、执行计划和风险。", call_sites=SIGNAL_PLANNER_CALL_SITES, ), ) def _scene_entries() -> tuple[dict[str, Any], ...]: return ( _entry( "PROMPT_FEISHU_DIRECT_QA", module="part_scenes_01", business_scenario="飞书直接问答", input_contract="输入 question / system_snapshot / user_snapshot / context_blocks。", output_contract=QA_OUTPUT, call_sites=DEEP_GUARD_CALL_SITES, ), _entry( "PROMPT_FEISHU_MIXED_QA", module="part_scenes_02", business_scenario="飞书混合问答", input_contract="输入 project_id / message / question_part / signal_part / system_snapshot / context_blocks。", output_contract="输出先结论后入库说明的自然语言答复。", call_sites=DEEP_GUARD_CALL_SITES, ), _entry( "PROMPT_PROJECT_STATUS_QA", module="part_scenes_03", business_scenario="项目状态问答", input_contract="输入 question / project_id / system_snapshot / user_snapshot / context_blocks。", output_contract="输出项目状态结论、进度与下一步建议。", call_sites=DEEP_GUARD_CALL_SITES, ), _entry( "PROMPT_MEMBER_WORKLOAD_QA", module="part_scenes_04", business_scenario="成员负载问答", input_contract="输入 question / system_snapshot / user_snapshot / context_blocks。", output_contract="输出成员负载与推荐建议。", call_sites=DEEP_GUARD_CALL_SITES, ), _entry( "PROMPT_AI_PMO_QA", module="part_scenes_05", business_scenario="AI PMO 问答", input_contract="输入 question / project_id / system_snapshot / user_snapshot / context_blocks。", output_contract="输出 AI PMO 视角的结论与建议。", call_sites=DEEP_GUARD_CALL_SITES, ), _entry( "PROMPT_EXTERNAL_TREND_QA", module="part_scenes_06", business_scenario="外部趋势问答", input_contract="输入 question / project_id / system_snapshot / user_snapshot / context_blocks。", output_contract="输出行业/市场趋势判断与建议。", call_sites=DEEP_GUARD_CALL_SITES, ), _entry( "PROMPT_INDUSTRY_TREND_QA", module="part_scenes_06", business_scenario="行业趋势问答", input_contract="输入 question / project_id / system_snapshot / user_snapshot / context_blocks。", output_contract="输出行业趋势判断与建议。", call_sites=DEEP_GUARD_CALL_SITES, ), ) PROMPT_REGISTRY: tuple[dict[str, Any], ...] = ( *_part_01_entries(), *_part_02_entries(), *_part_03_entries(), *_part_04_entries(), *_scene_entries(), ) PROMPT_REGISTRY_MAP: dict[str, dict[str, Any]] = { entry["prompt_key"]: entry for entry in PROMPT_REGISTRY } def get_prompt_registry() -> list[dict[str, Any]]: return [dict(entry) for entry in PROMPT_REGISTRY] def get_prompt_registry_entry(prompt_key: str) -> dict[str, Any] | None: key = str(prompt_key or "").strip() if not key: return None entry = PROMPT_REGISTRY_MAP.get(key) return dict(entry) if entry is not None else None def iter_prompt_registry_keys() -> tuple[str, ...]: return tuple(entry["prompt_key"] for entry in PROMPT_REGISTRY) __all__ = [ "PROMPT_CATALOG_EXPORTS", "PROMPT_REGISTRY", "PROMPT_REGISTRY_MAP", "get_prompt_registry", "get_prompt_registry_entry", "iter_prompt_registry_keys", ]