PMFlow/backend/app/services/ai_pmo/consultant_dispatch.py

340 lines
14 KiB
Python

"""Lightweight consultant dispatch contract and registries."""
from __future__ import annotations
import re
from typing import Any, Optional
DOMAIN_PROJECT = "project"
DOMAIN_TASK = "task"
DOMAIN_SIGNAL = "signal"
DOMAIN_RAG = "rag"
DOMAIN_FOLLOWUP = "followup"
DOMAIN_GENERIC = "generic"
MODE_CLARIFY = "clarify"
MODE_ADVISE = "advise"
MODE_CHALLENGE = "challenge"
MODE_BRAINSTORM = "brainstorm"
MODE_EXECUTE = "execute"
CONSULTANT_DOMAIN_PROMPT_REGISTRY: dict[str, dict[str, str]] = {
DOMAIN_PROJECT: {"prompt_key": "PROMPT_GENERATE_CHARTER", "role_hint": "pm_dispatcher"},
DOMAIN_TASK: {"prompt_key": "PROMPT_DECOMPOSE_TASKS", "role_hint": "delivery_coach"},
DOMAIN_SIGNAL: {"prompt_key": "PROMPT_FEISHU_DIRECT_QA", "role_hint": "delivery_coach"},
DOMAIN_RAG: {"prompt_key": "PROMPT_RAG_ANSWER", "role_hint": "pm_dispatcher"},
DOMAIN_FOLLOWUP: {"prompt_key": "PROMPT_AI_PMO_QA", "role_hint": "delivery_coach"},
DOMAIN_GENERIC: {"prompt_key": "PROMPT_RAG_ANSWER", "role_hint": "pm_dispatcher"},
}
CONSULTANT_MODE_WRAPPER_REGISTRY: dict[str, dict[str, str]] = {
MODE_CLARIFY: {"label": "信息不足,先澄清再推进", "tone": "direct"},
MODE_ADVISE: {"label": "直接建议最稳妥方案", "tone": "consultative"},
MODE_CHALLENGE: {"label": "高风险或范围过大,先挑战假设", "tone": "guardrail"},
MODE_BRAINSTORM: {"label": "先发散方案,再收敛边界", "tone": "creative"},
MODE_EXECUTE: {"label": "信息齐全,可继续执行", "tone": "action"},
}
CONSULTANT_CHANNEL_STRATEGIES: dict[str, dict[str, Any]] = {
"feishu": {"max_chars": 220, "prefer_direct": True},
"pmflow": {"max_chars": 500, "prefer_direct": False},
"pmflow_manual": {"max_chars": 500, "prefer_direct": False},
}
_QUERY_OPERATION_CODES = {"query.answer", "query.read_only"}
_MISSING_FIELD_LABELS = {
"project_id": "归属项目",
"task_title": "任务标题",
"title": "标题",
"description": "描述",
"project_scope": "项目范围",
}
_DOMAIN_NEXT_ACTION_FALLBACKS: dict[str, dict[str, str]] = {
DOMAIN_RAG: {
MODE_ADVISE: "继续追问具体模块、入口函数或文件路径,我会把调用链收窄到那一段。",
MODE_CLARIFY: "请先限定范围:前端入口、/rag/ask、Kernel 或 AdaptiveRAG 其中一段。",
MODE_CHALLENGE: "先确认你要的是 RAG 链路、业务执行链路,还是完整系统拓扑,再继续追查。",
MODE_BRAINSTORM: "告诉我你最想看的那一段,我给你 2-3 条追查路径。",
},
DOMAIN_PROJECT: {
MODE_ADVISE: "继续补充项目目标、时间边界或负责人,我会把建议收敛成更具体方案。",
MODE_CLARIFY: "请先补充归属项目、目标范围或关键约束,我再继续推进。",
MODE_CHALLENGE: "先确认项目范围和优先级,避免一次推进过多内容。",
MODE_BRAINSTORM: "告诉我目标、约束和时间窗口,我给你 2-3 个项目推进方向。",
},
DOMAIN_TASK: {
MODE_ADVISE: "继续补充任务目标、负责人或截止时间,我会给出更具体建议。",
MODE_CLARIFY: "请先补充任务标题、归属项目或截止时间,我再继续推进。",
MODE_CHALLENGE: "先确认任务边界和优先级,避免一次塞进过多要求。",
MODE_BRAINSTORM: "告诉我任务目标和约束,我给你拆成 2-3 条可执行路径。",
},
DOMAIN_SIGNAL: {
MODE_ADVISE: "继续补充信号背景、归属项目或影响范围,我会给出更具体建议。",
MODE_CLARIFY: "请先补充归属项目、标题或信号背景,我再继续推进。",
MODE_CHALLENGE: "先确认这个信号是否值得入库,再决定是否继续推进执行。",
MODE_BRAINSTORM: "告诉我信号来源、影响对象和目标,我给你列 2-3 种处理路径。",
},
DOMAIN_GENERIC: {
MODE_ADVISE: "继续补充你最关心的对象、范围或约束,我会给出更具体建议。",
MODE_CLARIFY: "请先补充关键对象、范围或约束,我再继续推进。",
MODE_CHALLENGE: "先确认目标边界和优先级,再决定是否继续推进。",
MODE_BRAINSTORM: "告诉我目标和约束,我给你列 2-3 个可选方向。",
},
}
def _normalize_missing_fields(value: Any) -> list[str]:
if not isinstance(value, list):
return []
fields: list[str] = []
for item in value:
text = str(item or "").strip()
if text and text not in fields:
fields.append(text)
return fields
def _normalize_action_text(text: str) -> str:
return re.sub(r"\s+", " ", str(text or "")).strip()
def _is_multiline_or_long_answer(text: str) -> bool:
raw = str(text or "")
normalized = _normalize_action_text(raw)
if not normalized:
return False
return "\n" in raw or len(normalized) > 120
def _domain_next_action(domain: str, mode: str) -> str:
normalized_domain = str(domain or "").strip().lower() or DOMAIN_GENERIC
normalized_mode = str(mode or "").strip().lower() or MODE_ADVISE
domain_map = _DOMAIN_NEXT_ACTION_FALLBACKS.get(normalized_domain) or _DOMAIN_NEXT_ACTION_FALLBACKS[DOMAIN_GENERIC]
return str(domain_map.get(normalized_mode) or _DOMAIN_NEXT_ACTION_FALLBACKS[DOMAIN_GENERIC][MODE_ADVISE]).strip()
def _resolve_domain(*, intent_name: str, primary_operation_code: Optional[str], classification: str) -> str:
normalized_op = str(primary_operation_code or "").strip().lower()
normalized_intent = str(intent_name or "").strip().lower()
normalized_classification = str(classification or "").strip().lower()
candidate = normalized_op or normalized_intent
if candidate.startswith("project.") or normalized_classification in {"project", "new_project"}:
return DOMAIN_PROJECT
if candidate.startswith("task.") or normalized_classification == "task":
return DOMAIN_TASK
if (
candidate.startswith("signal.")
or candidate.startswith("signal_source.")
or candidate.startswith("project_signal.")
or normalized_classification in {"signal", "project_signal", "command"}
):
return DOMAIN_SIGNAL
if candidate.startswith("followup."):
return DOMAIN_FOLLOWUP
if candidate in _QUERY_OPERATION_CODES or normalized_classification in {"question", "qa"}:
return DOMAIN_RAG
return DOMAIN_GENERIC
def _is_query_like(*, primary_operation_code: Optional[str], intent_name: str) -> bool:
normalized_op = str(primary_operation_code or "").strip().lower()
normalized_intent = str(intent_name or "").strip().lower()
return normalized_op in _QUERY_OPERATION_CODES or normalized_intent == "query.read_only"
def _derive_allow_execute(
*,
primary_operation_code: Optional[str],
status: str,
missing_fields: list[str],
) -> bool:
if _is_query_like(primary_operation_code=primary_operation_code, intent_name=str(primary_operation_code or "")):
return False
if missing_fields:
return False
normalized_status = str(status or "").strip().lower()
return normalized_status in {"succeeded", "answered", "captured", "completed"}
def _derive_mode(
*,
intent_name: str,
message_text: str,
status: str,
risk_level: str,
missing_fields: list[str],
allow_execute: bool,
primary_operation_code: Optional[str],
) -> str:
_ = message_text
if missing_fields:
return MODE_CLARIFY
if str(risk_level or "").strip().lower() == "high" and not allow_execute and not _is_query_like(
primary_operation_code=primary_operation_code,
intent_name=intent_name,
):
return MODE_CHALLENGE
if allow_execute and not _is_query_like(primary_operation_code=primary_operation_code, intent_name=intent_name):
return MODE_EXECUTE
if _is_query_like(primary_operation_code=primary_operation_code, intent_name=intent_name):
return MODE_ADVISE
if str(status or "").strip().lower() == "pending_manual":
return MODE_CHALLENGE
return MODE_ADVISE
def _build_clarify_next_action(*, missing_fields: list[str], direct_reply_text: str) -> str:
if direct_reply_text and not _is_multiline_or_long_answer(direct_reply_text):
return _normalize_action_text(direct_reply_text)[:120]
labels = [_MISSING_FIELD_LABELS.get(item, item) for item in missing_fields]
suffix = "、".join([label for label in labels if label]) or "关键信息"
return f"请先补充 {suffix},我再继续推进。"[:120]
def _build_next_action(
*,
domain: str,
mode: str,
risk_level: str,
missing_fields: list[str],
direct_reply_text: str,
) -> str:
if mode == MODE_CLARIFY:
return _build_clarify_next_action(
missing_fields=missing_fields,
direct_reply_text=direct_reply_text,
)
if mode == MODE_CHALLENGE:
if direct_reply_text and not _is_multiline_or_long_answer(direct_reply_text):
return _normalize_action_text(direct_reply_text)[:120]
if str(risk_level or "").strip().lower() == "high":
return "建议先拆分范围,再确认是否继续推进高风险动作。"[:120]
return _domain_next_action(domain=domain, mode=mode)
if mode == MODE_BRAINSTORM:
return _domain_next_action(domain=domain, mode=mode)
if mode == MODE_EXECUTE:
if domain == DOMAIN_PROJECT:
return "可继续查看项目结果,或补充负责人与排期信息。"
if domain == DOMAIN_TASK:
return "可继续补充负责人、优先级或截止时间。"
if domain == DOMAIN_SIGNAL:
return "可继续确认归属项目,或查看后续自动处理结果。"
return "可继续查看结果,或补充更多执行细节。"
if direct_reply_text and not _is_multiline_or_long_answer(direct_reply_text):
return _normalize_action_text(direct_reply_text)[:120]
return _domain_next_action(domain=domain, mode=mode)
def build_consultant_dispatch_contract(
*,
intent_name: str,
message_text: str,
source_system: str,
status: str,
classification: str,
primary_operation_code: Optional[str],
risk_level: str,
missing_fields: Optional[list[str]] = None,
direct_reply_text: str = "",
) -> dict[str, Any]:
normalized_missing = _normalize_missing_fields(missing_fields)
domain = _resolve_domain(
intent_name=intent_name,
primary_operation_code=primary_operation_code,
classification=classification,
)
allow_execute = _derive_allow_execute(
primary_operation_code=primary_operation_code,
status=status,
missing_fields=normalized_missing,
)
mode = _derive_mode(
intent_name=intent_name,
message_text=message_text,
status=status,
risk_level=risk_level,
missing_fields=normalized_missing,
allow_execute=allow_execute,
primary_operation_code=primary_operation_code,
)
_ = resolve_prompt_dispatch(domain=domain, mode=mode, channel=source_system)
return {
"domain": domain,
"mode": mode,
"risk_level": str(risk_level or "low").strip().lower() or "low",
"missing_fields": normalized_missing,
"allow_execute": bool(allow_execute),
"next_action": _build_next_action(
domain=domain,
mode=mode,
risk_level=risk_level,
missing_fields=normalized_missing,
direct_reply_text=str(direct_reply_text or "").strip(),
),
}
def resolve_prompt_dispatch(*, domain: str, mode: str, channel: str) -> dict[str, Any]:
normalized_domain = str(domain or "").strip().lower() or DOMAIN_GENERIC
normalized_mode = str(mode or "").strip().lower() or MODE_ADVISE
normalized_channel = str(channel or "").strip().lower() or "pmflow"
return {
"domain_prompt": CONSULTANT_DOMAIN_PROMPT_REGISTRY.get(
normalized_domain,
CONSULTANT_DOMAIN_PROMPT_REGISTRY[DOMAIN_GENERIC],
),
"mode_wrapper": CONSULTANT_MODE_WRAPPER_REGISTRY.get(
normalized_mode,
CONSULTANT_MODE_WRAPPER_REGISTRY[MODE_ADVISE],
),
"channel_strategy": CONSULTANT_CHANNEL_STRATEGIES.get(
normalized_channel,
CONSULTANT_CHANNEL_STRATEGIES["pmflow"],
),
}
def build_channel_consultant_reply(
*,
consultant_dispatch: Any,
direct_reply_text: str = "",
channel: str,
) -> str:
if not isinstance(consultant_dispatch, dict):
return ""
mode = str(consultant_dispatch.get("mode") or "").strip().lower()
if mode not in {MODE_CLARIFY, MODE_CHALLENGE, MODE_BRAINSTORM, MODE_ADVISE}:
return ""
dispatch_meta = resolve_prompt_dispatch(
domain=str(consultant_dispatch.get("domain") or DOMAIN_GENERIC),
mode=mode,
channel=channel,
)
channel_strategy = dispatch_meta.get("channel_strategy") if isinstance(dispatch_meta, dict) else {}
max_chars = int((channel_strategy or {}).get("max_chars") or 220)
reply_text = str(direct_reply_text or "").strip()
if reply_text:
return reply_text[:max_chars]
next_action = str(consultant_dispatch.get("next_action") or "").strip()
missing_fields = _normalize_missing_fields(consultant_dispatch.get("missing_fields"))
if mode == MODE_CLARIFY:
labels = [_MISSING_FIELD_LABELS.get(item, item) for item in missing_fields]
suffix = "、".join([label for label in labels if label]) or "关键信息"
return f"当前还不能继续推进,请先补充 {suffix}。{next_action or '补齐后我再继续处理。'}"[:max_chars]
if mode == MODE_CHALLENGE:
return f"这次动作风险较高,建议先不要直接推进。{next_action or '请先确认边界后再继续。'}"[:max_chars]
if mode == MODE_BRAINSTORM:
return f"我们可以先一起发散方案。{next_action or '请补充目标和约束,我给你几个方向。'}"[:max_chars]
return f"当前建议:{next_action or '先按最稳妥方向继续。'}"[:max_chars]
__all__ = [
"build_channel_consultant_reply",
"build_consultant_dispatch_contract",
"resolve_prompt_dispatch",
]