Update backend/pipeline.py

backend/pipeline.py

@@ -15,8 +15,10 @@ Agent execution order per frame:
 import time
 import cv2
 import numpy as np
+import threading
 from collections import deque
 
+import config
 from core.database import HajjFlowDB
 from agents.perception_agent  import PerceptionAgent
 from agents.risk_agent        import RiskAgent
@@ -38,12 +40,14 @@ class RealTimePipeline:
         self.db         = HajjFlowDB(db_path)
 
         # ── Instantiate agents ────────────────────────────────────────
-        # YOLO-only mode — VisionCountAgent (Claude Vision) disabled
-        self.perception  = PerceptionAgent(model_path, anthropic_key=None)
+        # Cached mode: read pre-computed detections; Live mode: run YOLO
+        cached_path = getattr(config, 'CACHED_DETECTIONS_PATH', None)
+        self.perception  = PerceptionAgent(model_path, anthropic_key=None, cached_path=cached_path)
         self.risk        = RiskAgent()
         self.reflection  = ReflectionAgent()
         self.operations  = OperationsAgent(self.db)
         self.coordinator = CoordinatorAgent(groq_api_key)
+        self._last_effective_risk_level = 'LOW'
 
         self._frame_buffer = deque(maxlen=30)
 
@@ -99,12 +103,17 @@ class RealTimePipeline:
 
         # Agent 2 — Risk scoring
         rr = self.risk.process_frame(fr)
+        original_level_changed = rr.level_changed
 
         # Agent 3 — Reflection (self-critique)
         reflection = self.reflection.reflect(rr, fr)
         if reflection['bias_detected']:
             rr.risk_level = reflection['corrected_level']
             rr.risk_score = reflection['corrected_score']
+        # Ensure event-driven decisions use effective risk after reflection.
+        effective_level_changed = rr.risk_level != self._last_effective_risk_level
+        rr.level_changed = bool(original_level_changed or effective_level_changed)
+        self._last_effective_risk_level = rr.risk_level
         self.db.save_reflection(reflection)
 
         # Save risk event every 30 frames
@@ -115,16 +124,9 @@ class RealTimePipeline:
         decision = self.operations.process(rr, context='Mecca_Main_Area')
 
         # Agent 5 — Coordinator (all priorities: P0/P1/P2)
-        # LLM dynamically decides which gates to act on and what actions to take.
-        plan = None
+        # LLM runs in background thread so it doesn't block frame processing.
         if decision:
-            plan = self.coordinator.call(rr, decision, list(self._frame_buffer))
-            if plan:
-                # Fill decision with LLM-chosen gates and actions
-                decision.actions        = plan.get('immediate_actions', [])
-                decision.justification  = plan.get('actions_justification', '')
-                decision.selected_gates = plan.get('selected_gates', [])
-                self.db.save_coordinator_plan(fr.frame_id, plan)
+            self._launch_coordinator(rr, decision, list(self._frame_buffer))
 
         # ── Update shared state ───────────────────────────────────────
         fps = round(1.0 / (time.time() - t0 + 1e-9), 1)
@@ -148,6 +150,21 @@ class RealTimePipeline:
             self.state['latest_decision'] = decision
             self.state['decisions_log']   = self.db.get_recent_decisions(10)
 
-        if plan:
-            self.state['coordinator_plan'] = plan
-            self.state['arabic_alert']     = plan.get('arabic_alert', '')
+    def _launch_coordinator(self, rr, decision, frame_buffer):
+        """Run CoordinatorAgent LLM call in a background thread."""
+        def _run():
+            try:
+                plan = self.coordinator.call(rr, decision, frame_buffer)
+                if plan:
+                    decision.actions        = plan.get('immediate_actions', [])
+                    decision.justification  = plan.get('actions_justification', '')
+                    decision.selected_gates = plan.get('selected_gates', [])
+                    self.db.save_coordinator_plan(rr.frame_id, plan)
+                    self.state['coordinator_plan'] = plan
+                    self.state['arabic_alert']     = plan.get('arabic_alert', '')
+                    self.state['latest_decision']  = decision
+                    self.state['decisions_log']    = self.db.get_recent_decisions(10)
+            except Exception as exc:
+                print(f'[Pipeline] Coordinator error: {exc}')
+
+        threading.Thread(target=_run, daemon=True).start()