23 lines
1.2 KiB
Python
23 lines
1.2 KiB
Python
from app.config import get_settings
|
|
from app.domain.schemas import PlannerDecision, PolicyResult
|
|
|
|
|
|
class PolicyEngine:
|
|
def __init__(self, existing_payloads: list[dict] | None = None) -> None:
|
|
self.threshold = get_settings().planner_confidence_threshold
|
|
self.existing_payloads = existing_payloads or []
|
|
|
|
def evaluate(self, decision: PlannerDecision) -> PolicyResult:
|
|
reasons: list[str] = []
|
|
if decision.confidence < self.threshold:
|
|
reasons.append(f"Confidence {decision.confidence:.2f} is below threshold {self.threshold:.2f}.")
|
|
if decision.arguments in self.existing_payloads:
|
|
reasons.append("Duplicate pending action payload.")
|
|
if decision.tool == "create_guest_reservation" and not decision.arguments.get("event_date"):
|
|
reasons.append("Reservation requires a date.")
|
|
if decision.tool == "create_guest_reservation" and not decision.arguments.get("partner_id"):
|
|
reasons.append("Reservation requires a Wed partner_id.")
|
|
if decision.tool == "create_task" and not decision.arguments.get("title"):
|
|
reasons.append("Task requires a title.")
|
|
return PolicyResult(allowed=not reasons, reasons=reasons)
|