File size: 3,077 Bytes
c99bf3c | 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 90 91 92 93 94 | """
Thalamic Loop — Vitalis FSI
The full perception pipeline:
Raw input
→ SyntheticThalamus (multimodal fusion)
→ AttentionalGate (salience filtering)
→ PredictiveCortex (prediction error)
→ Output: error vector + surprise + metadata
Only high-salience, surprising inputs reach the cognitive core.
Everything else is suppressed or predicted away.
This is attentional gating + predictive processing combined.
"""
import numpy as np
from pathlib import Path
from src.thalamus.thalamus import SyntheticThalamus
from src.cortex.attention import AttentionalGate
from src.cortex.predictive import PredictiveCortex
from src.valence.valence_engine import ValenceEngine
class ThalamicLoop:
def __init__(self, valence: ValenceEngine, identity_vec: np.ndarray = None):
self.thalamus = SyntheticThalamus()
self.attention = AttentionalGate(valence, identity_vec)
self.cortex = PredictiveCortex()
self._processed = 0
self._suppressed = 0
def process(self, payload: dict, force_attention: bool = False) -> dict:
"""
Full thalamic pipeline.
payload: dict with "text", "audio", "internal" keys
Returns dict with:
"hv" : fused hypervector from thalamus
"error_vec": prediction error from cortex
"surprise" : float surprise level
"is_novel" : bool
"passes" : bool — did it pass attention gate
"salience" : float
"suppressed": bool
"""
# 1. Thalamic fusion
hv = self.thalamus.ingest(payload)
# 2. Attentional gating
passes, salience, gate_reason = self.attention.gate(hv, force=force_attention)
if not passes:
self._suppressed += 1
return {
"hv": hv,
"error_vec": None,
"surprise": 0.0,
"is_novel": False,
"passes": False,
"salience": salience,
"suppressed": True,
"reason": gate_reason,
}
# 3. Predictive cortex
error_vec, surprise, is_novel = self.cortex.process(hv)
self._processed += 1
return {
"hv": hv,
"error_vec": error_vec,
"surprise": surprise,
"is_novel": is_novel,
"passes": True,
"salience": salience,
"suppressed": False,
"reason": gate_reason,
}
def process_text(self, text: str, force: bool = False) -> dict:
return self.process({"text": text}, force_attention=force)
def acknowledge_dream(self):
"""Reset prediction after dream cycle."""
self.cortex.reset_prediction()
def report(self) -> dict:
return {
"processed": self._processed,
"suppressed": self._suppressed,
"attention": self.attention.report(),
"cortex": self.cortex.report(),
"pineal": None,
}
|