add human loop agent

src/lg_sotf/agents/human_loop/__init__.py

@@ -1 +1,8 @@
-
+"""Human loop package."""
+# src/lg_sotf/agents/human_loop/__init__.py
+
+from .base import HumanLoopAgent
+from .escalation import EscalationManager
+from .feedback import FeedbackCollector
+
+__all__ = ["HumanLoopAgent", "EscalationManager", "FeedbackCollector"]

src/lg_sotf/agents/human_loop/base.py

@@ -1 +1,37 @@
-
+"""Base class for human loop agent."""
+# src/lg_sotf/agents/human_loop/base.py
+from datetime import datetime, timedelta
+from typing import Any, Dict
+
+
+class HumanLoopAgent:
+    def __init__(self, escalation_threshold: float = 0.6, sla_hours: int = 4):
+        """
+        Parameters:
+        - escalation_threshold: alerts below this confidence go to human review
+        - sla_hours: SLA for human feedback
+        """
+        self.escalation_threshold = escalation_threshold
+        self.sla = timedelta(hours=sla_hours)
+        self.pending_alerts: Dict[str, Dict[str, Any]] = {}
+
+    def receive_alert(self, alert_id: str, alert_data: Dict[str, Any]):
+        """Add alert for human review."""
+        self.pending_alerts[alert_id] = {
+            "alert_data": alert_data,
+            "received_at": datetime.utcnow(),
+            "status": "pending",
+            "feedback": None,
+        }
+
+    def get_pending_alerts(self):
+        """Return pending alerts that need human attention."""
+        return {
+            aid: a for aid, a in self.pending_alerts.items() if a["status"] == "pending"
+        }
+
+    def submit_feedback(self, alert_id: str, feedback: Dict[str, Any]):
+        """Store human analyst feedback."""
+        if alert_id in self.pending_alerts:
+            self.pending_alerts[alert_id]["feedback"] = feedback
+            self.pending_alerts[alert_id]["status"] = "reviewed"

src/lg_sotf/agents/human_loop/escalation.py

@@ -1 +1,24 @@
-
+"""Human loop escalation agent."""
+# src/lg_sotf/agents/human_loop/escalation.py
+from datetime import datetime
+
+from .base import HumanLoopAgent
+
+
+class EscalationManager:
+    def __init__(self, human_agent: HumanLoopAgent):
+        self.agent = human_agent
+
+    def evaluate_alert(self, alert_id: str, confidence_score: float):
+        """
+        Decide if alert needs human escalation
+        """
+        if confidence_score < self.agent.escalation_threshold:
+            self.agent.pending_alerts[alert_id] = {
+                "alert_data": {"confidence_score": confidence_score},
+                "received_at": datetime.utcnow(),
+                "status": "pending",
+                "feedback": None,
+            }
+            return True
+        return False

src/lg_sotf/agents/human_loop/feedback.py

@@ -1 +1,22 @@
-
+"""Human loop feedback agent."""
+# src/lg_sotf/agents/human_loop/feedback.py
+
+from .base import HumanLoopAgent
+
+
+class FeedbackCollector:
+    def __init__(self, human_agent: HumanLoopAgent):
+        self.agent = human_agent
+
+    def collect_feedback(
+        self, alert_id: str, analyst: str, decision: str, notes: str = ""
+    ):
+        """
+        Store structured feedback
+        """
+        feedback = {
+            "analyst": analyst,
+            "decision": decision,  # "approve", "deny", "more_info"
+            "notes": notes,
+        }
+        self.agent.submit_feedback(alert_id, feedback)