Create ai_risk_engine.py

ai_risk_engine.py

@@ -0,0 +1,49 @@
+"""
+Local Bayesian risk engine for AI tasks – uses conjugate Beta priors.
+"""
+
+import threading
+import numpy as np
+from typing import Dict, Tuple
+
+AI_CATEGORIES = ["chat", "code", "summary", "image", "audio", "iot"]
+
+DEFAULT_PRIORS = {
+    "chat": (1.0, 10.0),
+    "code": (0.5, 8.0),
+    "summary": (1.0, 12.0),
+    "image": (1.0, 15.0),
+    "audio": (1.0, 15.0),
+    "iot": (1.0, 10.0),
+}
+
+class AIRiskEngine:
+    def __init__(self):
+        self._lock = threading.RLock()
+        self._priors: Dict[str, Tuple[float, float]] = DEFAULT_PRIORS.copy()
+        self._counts: Dict[str, Tuple[int, int]] = {cat: (0, 0) for cat in AI_CATEGORIES}
+
+    def get_posterior(self, category: str) -> Tuple[float, float]:
+        prior_a, prior_b = self._priors[category]
+        succ, trials = self._counts[category]
+        return prior_a + succ, prior_b + (trials - succ)
+
+    def risk_score(self, category: str) -> Dict[str, float]:
+        with self._lock:
+            alpha, beta = self.get_posterior(category)
+        mean = alpha / (alpha + beta)
+        samples = np.random.beta(alpha, beta, size=10000)
+        return {
+            "mean": float(mean),
+            "p5": float(np.percentile(samples, 5)),
+            "p50": float(np.percentile(samples, 50)),
+            "p95": float(np.percentile(samples, 95)),
+        }
+
+    def update_outcome(self, category: str, success: bool):
+        with self._lock:
+            succ, trials = self._counts[category]
+            trials += 1
+            if success:
+                succ += 1
+            self._counts[category] = (succ, trials)