| """ |
| 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) |
