Create memory_drift_diagnostician.py

memory_drift_diagnostician.py

@@ -0,0 +1,52 @@
+import logging
+import numpy as np
+from typing import Dict, Any, List
+from agentic_reliability_framework.runtime.agents.base import BaseAgent
+from ai_event import AIEvent
+
+logger = logging.getLogger(__name__)
+
+class MemoryDriftDiagnosticianAgent(BaseAgent):
+    """Detects drift in semantic memory by comparing current retrieval scores with historical distribution."""
+    def __init__(self, history_window: int = 100):
+        super().__init__(AgentSpecialization.DIAGNOSTICIAN)
+        self.history_window = history_window
+        self._retrieval_scores_history: List[float] = []
+
+    async def analyze(self, event: AIEvent) -> Dict[str, Any]:
+        try:
+            if not event.retrieval_scores:
+                return {'specialization': 'ai_memory_drift', 'confidence': 0, 'findings': {}, 'recommendations': []}
+            current_avg = float(np.mean(event.retrieval_scores))
+            self._retrieval_scores_history.append(current_avg)
+            if len(self._retrieval_scores_history) > self.history_window:
+                self._retrieval_scores_history.pop(0)
+
+            if len(self._retrieval_scores_history) < 10:
+                drift_detected = False
+                confidence = 0.0
+            else:
+                historical_avg = float(np.mean(self._retrieval_scores_history[:-1]))
+                historical_std = float(np.std(self._retrieval_scores_history[:-1])) + 1e-6
+                z_score = (current_avg - historical_avg) / historical_std
+                drift_detected = abs(z_score) > 2.0
+                confidence = min(1.0, abs(z_score) / 5.0)
+
+            return {
+                'specialization': 'ai_memory_drift',
+                'confidence': confidence,
+                'findings': {
+                    'drift_detected': drift_detected,
+                    'current_avg': current_avg,
+                    'historical_avg': historical_avg if len(self._retrieval_scores_history) > 1 else None,
+                    'z_score': float(z_score) if len(self._retrieval_scores_history) > 1 else None
+                },
+                'recommendations': [
+                    "Reindex knowledge base",
+                    "Adjust embedding model",
+                    "Update context window"
+                ] if drift_detected else []
+            }
+        except Exception as e:
+            logger.error(f"MemoryDriftDiagnostician error: {e}", exc_info=True)
+            return {'specialization': 'ai_memory_drift', 'confidence': 0.0, 'findings': {}, 'recommendations': []}