"""Shared decision kernel for dual-agent routing and guardrails.""" from __future__ import annotations import os import re from dataclasses import dataclass from typing import Iterable, Sequence BOOL_TRUE_VALUES = {"1", "true", "yes", "y", "on", "是"} INTENT_FUNCTION_SIMPLE = "classify_intent_ai_pmo_kernel" INTENT_FUNCTION_COMPLEX = "classify_intent_complex" BACKGROUND_MARKERS = ("因为", "所以", "用来", "为了", "方便", "希望", "背景", "场景") MULTI_SEGMENT_MARKERS = (",", "，", "、", ";", "；", "并且", "同时", "然后") INTENT_GROUP_KEYWORDS: dict[str, tuple[str, ...]] = { "project": ("项目", "立项", "创建项目", "新增项目", "新建项目"), "signal": ("信号", "信号源", "接入", "onboard", "source"), "task": ("任务", "分配", "指派", "执行", "部署"), "change": ("修改", "更新", "改为", "改成", "撤销", "回滚", "归档"), } CORRECTION_ACTION_PATTERNS: tuple[tuple[str, tuple[str, ...]], ...] = ( ("undo_last", ("撤销", "撤回", "undo", "作废上一步", "取消上一步")), ("rollback", ("回滚", "恢复到", "还原", "回退")), ("replace_target", ("改为", "改成", "替换为", "不是", "不对", "纠正")), ("reroute", ("重新识别", "重新判断", "重新路由", "再判断一次")), ) def _env_float(name: str, default: float) -> float: raw = os.getenv(name) if raw is None: return float(default) try: value = float(str(raw).strip()) except Exception: return float(default) return float(default) if value < 0 else value def _env_bool(name: str, default: bool = False) -> bool: raw = os.getenv(name) if raw is None: return default return str(raw).strip().lower() in BOOL_TRUE_VALUES def _normalize_text(text: str) -> str: return re.sub(r"\s+", "", str(text or "")).lower() def _normalize_name(name: str) -> str: return re.sub(r"[\s_\-]+", "", str(name or "")).lower() @dataclass(slots=True) class IntentRouteDecision: function_name: str tier: str is_complex: bool reasons: list[str] @dataclass(slots=True) class GateDecision: action: str gate_code: str reason: str auto_execute: bool @dataclass(slots=True) class CorrectionDecision: action: str confidence: float matched_keywords: list[str] def _count_intent_groups(text: str) -> int: normalized = _normalize_text(text) hits = 0 for tokens in INTENT_GROUP_KEYWORDS.values(): if any(_normalize_name(token) in normalized for token in tokens): hits += 1 return hits def is_complex_instruction(text: str, *, project_names: Sequence[str] | None = None) -> tuple[bool, list[str]]: normalized = _normalize_text(text) if not normalized: return False, [] reasons: list[str] = [] length_threshold = int(_env_float("AGENT_COMPLEX_LENGTH_THRESHOLD", 20)) if len(normalized) > length_threshold: reasons.append("length_gt_threshold") has_multi_segment = any(marker in text for marker in MULTI_SEGMENT_MARKERS) if has_multi_segment: reasons.append("multi_segment") has_background = any(marker in text for marker in BACKGROUND_MARKERS) if has_background: reasons.append("has_background") intent_group_hits = _count_intent_groups(text) if intent_group_hits >= 3 or ( intent_group_hits >= 2 and ( len(normalized) > max(14, length_threshold) or has_multi_segment or has_background ) ): reasons.append("multi_intent_groups") for name in project_names or []: candidate = _normalize_name(name) if len(candidate) >= 2 and candidate in normalized: reasons.append("contains_project_name") break return bool(reasons), reasons def select_intent_route(text: str, *, project_names: Sequence[str] | None = None) -> IntentRouteDecision: is_complex, reasons = is_complex_instruction(text, project_names=project_names) if is_complex: return IntentRouteDecision( function_name=INTENT_FUNCTION_COMPLEX, tier="high", is_complex=True, reasons=reasons, ) return IntentRouteDecision( function_name=INTENT_FUNCTION_SIMPLE, tier="low", is_complex=False, reasons=["default_simple"], ) def resolve_correction(text: str) -> CorrectionDecision: normalized = _normalize_text(text) if not normalized: return CorrectionDecision(action="none", confidence=0.0, matched_keywords=[]) for action, tokens in CORRECTION_ACTION_PATTERNS: matched = [token for token in tokens if _normalize_name(token) in normalized] if matched: confidence = 0.9 if len(matched) >= 2 else 0.78 return CorrectionDecision( action=action, confidence=float(min(1.0, confidence)), matched_keywords=matched, ) return CorrectionDecision(action="none", confidence=0.0, matched_keywords=[]) def gate_execution( *, confidence: float, risk_level: str = "medium", correction_action: str = "none", force_manual: bool = False, force_manual_code: str = "manual_review_required", force_manual_reason: str = "", ) -> GateDecision: auto_threshold = _env_float("AGENT_GATE_AUTO_THRESHOLD", 0.82) review_threshold = _env_float("AGENT_GATE_REVIEW_THRESHOLD", 0.60) strict_high_risk = _env_bool("AGENT_GATE_STRICT_HIGH_RISK", True) require_review_below_auto = _env_bool("AGENT_GATE_REVIEW_BELOW_AUTO", False) normalized_risk = str(risk_level or "medium").strip().lower() safe_confidence = max(0.0, min(1.0, float(confidence or 0.0))) if force_manual: return GateDecision( action="pending_manual", gate_code=str(force_manual_code or "manual_review_required").strip() or "manual_review_required", reason=str(force_manual_reason or "命中人工复核策略，需人工确认后执行。").strip(), auto_execute=False, ) if correction_action in {"undo_last", "rollback", "replace_target", "reroute"}: return GateDecision( action="pending_manual", gate_code="correction_review", reason="检测到修正/回滚语义，需人工确认后执行。", auto_execute=False, ) if safe_confidence < review_threshold: return GateDecision( action="reject", gate_code="confidence_too_low", reason="置信度低于安全阈值，已阻断自动执行。", auto_execute=False, ) if normalized_risk in {"high", "critical"} and strict_high_risk and safe_confidence < max(auto_threshold, 0.9): return GateDecision( action="pending_manual", gate_code="high_risk_manual_review", reason="高风险写操作需要更高置信度，当前进入待人工确认。", auto_execute=False, ) if safe_confidence < auto_threshold and require_review_below_auto: return GateDecision( action="pending_manual", gate_code="confidence_review_required", reason="置信度未达自动执行阈值，进入待人工确认。", auto_execute=False, ) return GateDecision( action="auto_execute", gate_code="auto_execute_allowed", reason="置信度达标，允许自动执行。", auto_execute=True, ) def build_decision_trace( *, intent: str, confidence: float, route_tier: str, route_function: str, gate_action: str, gate_code: str, correction_action: str = "none", correction_keywords: Iterable[str] | None = None, complexity_reasons: Iterable[str] | None = None, ) -> dict[str, object]: return { "intent": str(intent or "").strip(), "confidence": round(max(0.0, min(1.0, float(confidence or 0.0))), 4), "route_tier": str(route_tier or "").strip(), "route_function": str(route_function or "").strip(), "gate_action": str(gate_action or "").strip(), "gate_code": str(gate_code or "").strip(), "correction_action": str(correction_action or "none").strip(), "correction_keywords": [str(token) for token in (correction_keywords or []) if str(token).strip()], "complexity_reasons": [str(token) for token in (complexity_reasons or []) if str(token).strip()], } def compute_capability_score( *, intent_accuracy: float, gate_precision: float, exec_success: float, correction_success: float, ) -> float: score = ( 0.30 * max(0.0, min(1.0, float(intent_accuracy or 0.0))) + 0.30 * max(0.0, min(1.0, float(gate_precision or 0.0))) + 0.25 * max(0.0, min(1.0, float(exec_success or 0.0))) + 0.15 * max(0.0, min(1.0, float(correction_success or 0.0))) ) return round(score, 4)