File size: 3,585 Bytes
a1365f7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | import logging
from src.devcore.vitalis_cognitive_engine import VitalisCognitiveEngine
from src.cognition.working_memory import WorkingMemory
from src.cognition.predictive_engine import PredictiveEngine
from src.dream_engine.synthetic_helix import SyntheticHelixMemory
from vitalis_ide.math_core.kernel import VitalisKernel
log = logging.getLogger('QuadFlow')
class NexusHead:
def __init__(self):
self.kernel = VitalisKernel()
self.wm = WorkingMemory()
self.predictor = PredictiveEngine()
self.helix = SyntheticHelixMemory()
def pre(self, intent, mind_state):
resonance_w = {
k: v for k, v in
mind_state.get('resonance', {}).get('strongest', [])
}
prediction = self.predictor.predict_next(
intent,
mind_state.get('meta_rules', {}),
resonance_w,
)
if prediction.get('predicted_next'):
self.predictor.anticipate(prediction, self.wm)
vec = self.kernel.vectorize_tokens(intent.split(), positional=False)
self.wm.push(intent, vec, 1.0)
return {'prediction': prediction, 'wm_context': self.wm.context(3)}
def post(self, intent, result, success):
accuracy = self.predictor.score(intent)
mode = result.get('mode', 'unknown')
conf = result.get('confidence', 0.0)
status = 'success' if success else 'fail'
meaning = status + ' mode=' + mode + ' conf=' + str(round(conf, 3))
self.helix.encode(event=intent, meaning=meaning)
return {'accuracy': accuracy}
def report(self):
return {
'working_memory': self.wm.report(),
'predictor': self.predictor.report(),
'helix': self.helix.report(),
}
class NeurosynthticQuadFlowEngine(VitalisCognitiveEngine):
def __init__(self):
super().__init__()
self.nexus = NexusHead()
print('')
print(' ╔══════════════════════════════════════╗')
print(' ║ NEUROSYNTHETIC QUAD FLOW ENGINE ║')
print(' ║ Sensu | Ratio | Cor | Nexus ║')
print(' ║ WorkingMem + Foresight + Helix ║')
print(' ╚══════════════════════════════════════╝')
print('')
def think_and_act(self, intent, token, **kwargs):
if not self.auth.validate_request(token):
return {'success': False, 'error': 'UNAUTHORIZED'}
from src.cognition.mind import VitalisMind, _extend_mind
mind = VitalisMind()
ms = mind.introspect()
pre = self.nexus.pre(intent, ms)
predicted = pre['prediction'].get('predicted_next', '?')
wm_ctx = pre['wm_context']
log.info('[NEXUS] Predicted: ' + str(predicted))
log.info('[NEXUS] WM: ' + str(wm_ctx))
result = super().think_and_act(intent, token, **kwargs)
post = self.nexus.post(intent, result, result.get('success', False))
result['nexus'] = {
'predicted_next': predicted,
'wm_context': wm_ctx,
'prediction_accuracy': post['accuracy'],
}
return result
def status(self):
return {
'engine': 'NeurosynthticQuadFlowEngine v2.1',
'heads': ['Sensu', 'Ratio', 'Cor', 'Nexus'],
'nexus': self.nexus.report(),
}
|