src/cognition/predictive_engine.py

import numpy as np
from vitalis_ide.math_core.kernel import VitalisKernel

class PredictiveEngine:
    THRESHOLD = 0.35

    def __init__(self):
        self.kernel    = VitalisKernel()
        self._history  = []
        self._accuracy = []

    def predict_next(self, current_intent, meta_report, resonance_weights):
        action     = current_intent.split()[0] if current_intent else 'unknown'
        candidates = []
        for rule in (meta_report or {}).get('top_rules', []):
            seq  = rule.get('sequence', [])
            conf = rule.get('confidence', 0.0)
            if len(seq) >= 2 and action in seq[0]:
                boost = resonance_weights.get(seq[1].split()[0], 1.0) / 2.0
                candidates.append((conf * boost, seq[1]))
        candidates.sort(reverse=True)
        result = {'current':current_intent,'predicted_next':candidates[0][1] if candidates else None,'confidence':round(candidates[0][0],4) if candidates else 0.0,'alternatives':[c[1] for c in candidates[1:3]]}
        self._history.append(result)
        return result

    def anticipate(self, prediction, working_memory):
        ni   = prediction.get('predicted_next')
        conf = prediction.get('confidence', 0.0)
        if not ni or conf < self.THRESHOLD: return False
        vec = self.kernel.vectorize_tokens(ni.split(), positional=False)
        working_memory.push('[PREDICTED] ' + ni, vec, conf, {'source':'prediction'})
        return True

    def score(self, actual_intent):
        if not self._history or not self._history[-1].get('predicted_next'): return 0.0
        p   = set(self._history[-1]['predicted_next'].lower().split())
        a   = set(actual_intent.lower().split())
        acc = len(p & a) / max(len(p | a), 1)
        self._accuracy.append(acc)
        return round(acc, 4)

    def report(self):
        avg = round(float(np.mean(self._accuracy)), 4) if self._accuracy else 0.0
        return {'total_predictions':len(self._history),'avg_accuracy':avg}