69 lines
3.4 KiB
Python
69 lines
3.4 KiB
Python
"""Tactical brain for AI PMO decision runtime."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from .shared import (
|
|
CanonicalIntent,
|
|
INTENT_FOLLOWUP_CONFIRM,
|
|
INTENT_FOLLOWUP_CONTINUE,
|
|
INTENT_PROJECT_CREATE,
|
|
INTENT_QUERY,
|
|
INTENT_SIGNAL_CREATE,
|
|
INTENT_SIGNAL_SOURCE_ONBOARD,
|
|
ROUTE_PROJECT_SIGNAL_CANCEL,
|
|
ROUTE_PROJECT_SIGNAL_COMMIT,
|
|
ROUTE_PROJECT_SIGNAL_DISCOVER,
|
|
ROUTE_PROJECT_SIGNAL_REFRESH,
|
|
Any,
|
|
)
|
|
|
|
|
|
class TacticalBrain:
|
|
"""周/日执行计划。"""
|
|
|
|
@staticmethod
|
|
def plan(*, intent: CanonicalIntent, world_model: dict[str, Any], decision: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
project = world_model.get("project") if isinstance(world_model.get("project"), dict) else None
|
|
project_ref = f"项目#{project['id']}" if project and project.get("id") is not None else "项目群"
|
|
strategic_plan = {
|
|
"priority": "high" if float(decision.get("risk_score") or 0.0) >= 0.6 else "medium",
|
|
"objectives": [
|
|
f"围绕 {project_ref} 控制风险与节奏",
|
|
"优先保证关键路径和资源边界不被破坏",
|
|
],
|
|
"signals": [
|
|
f"budget_usage_ratio={float((decision.get('budget_impact') or {}).get('usage_ratio') or 0.0):.2f}",
|
|
f"overloaded_users={len((decision.get('resource_impact') or {}).get('overloaded_user_ids') or [])}",
|
|
],
|
|
}
|
|
execution_actions = []
|
|
workflow_route = str((intent.metadata or {}).get("workflow_route") or "").strip().lower()
|
|
if intent.name == INTENT_QUERY:
|
|
execution_actions.append({"operation_code": "query.answer", "mode": "read_only"})
|
|
elif intent.name == INTENT_PROJECT_CREATE:
|
|
execution_actions.append({"operation_code": "project.create", "mode": "write"})
|
|
elif intent.name == INTENT_SIGNAL_SOURCE_ONBOARD:
|
|
execution_actions.append({"operation_code": "signal.enter", "mode": "write"})
|
|
execution_actions.append({"operation_code": "signal_source.onboard", "mode": "write"})
|
|
elif intent.name in {INTENT_FOLLOWUP_CONFIRM, INTENT_FOLLOWUP_CONTINUE}:
|
|
execution_actions.append({"operation_code": "work_item.resume", "mode": "write"})
|
|
elif intent.name == INTENT_SIGNAL_CREATE:
|
|
if workflow_route == ROUTE_PROJECT_SIGNAL_DISCOVER:
|
|
execution_actions.append({"operation_code": ROUTE_PROJECT_SIGNAL_DISCOVER, "mode": "write"})
|
|
elif workflow_route == ROUTE_PROJECT_SIGNAL_COMMIT:
|
|
execution_actions.append({"operation_code": ROUTE_PROJECT_SIGNAL_COMMIT, "mode": "write"})
|
|
elif workflow_route == ROUTE_PROJECT_SIGNAL_REFRESH:
|
|
execution_actions.append({"operation_code": ROUTE_PROJECT_SIGNAL_REFRESH, "mode": "write"})
|
|
elif workflow_route == ROUTE_PROJECT_SIGNAL_CANCEL:
|
|
execution_actions.append({"operation_code": ROUTE_PROJECT_SIGNAL_CANCEL, "mode": "write"})
|
|
else:
|
|
execution_actions.append({"operation_code": "signal.enter", "mode": "write"})
|
|
|
|
execution_plan = {
|
|
"intent": intent.name,
|
|
"project_id": intent.project_id,
|
|
"actions": execution_actions,
|
|
"summary": f"执行 {intent.name}，共 {len(execution_actions)} 个 typed actions",
|
|
}
|
|
return strategic_plan, execution_plan
|