574 lines
23 KiB
Python
574 lines
23 KiB
Python
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",
|
|
]
|