Add ValenceEngine, SyntheticThalamus, DecisionGate

src/basal_ganglia/__init__.py

@@ -0,0 +1 @@
+

src/basal_ganglia/decision_gate.py

@@ -0,0 +1,105 @@
+"""
+Basal Ganglia Decision Gate — Vitalis FSI
+
+Probabilistic action selector.
+High valence + high novelty + high confidence = action selected.
+Low valence or high cost = action suppressed.
+
+This is how Vitalis decides what to do next.
+Not deterministic. Not random. Weighted by what it has learned.
+"""
+import numpy as np
+from pathlib import Path
+from src.valence.valence_engine import ValenceEngine
+from src.dream_engine.helix_memory import HelixMemory
+
+
+class DecisionGate:
+    def __init__(
+        self,
+        valence_engine: ValenceEngine,
+        helix_path: Path = None,
+        alpha: float = 1.0,   # valence weight
+        beta:  float = 0.8,   # novelty weight
+        gamma: float = 0.6,   # confidence weight
+        delta: float = 0.2,   # cost penalty
+    ):
+        self.valence  = valence_engine
+        self.helix    = HelixMemory(
+            helix_path or Path.home() / ".vitalis_workspace" / "helix_memory.pkl"
+        )
+        self.alpha = alpha
+        self.beta  = beta
+        self.gamma = gamma
+        self.delta = delta
+        self._history = []
+
+    def _novelty(self, hv: np.ndarray) -> float:
+        """How different is this from anything in long-term memory."""
+        if not self.helix.entries:
+            return 1.0
+        sims = [float(np.mean(hv == proto))
+                for _, proto, _, _ in self.helix.entries]
+        return float(1.0 - max(sims))
+
+    def score(self, candidate: dict) -> float:
+        """
+        Score one candidate action.
+        candidate must have:
+            "hv"         : hypervector (np.ndarray)
+            "confidence" : float 0-1 (optional, default 0.5)
+            "cost"       : float (optional, default 1.0)
+        """
+        hv         = candidate["hv"]
+        confidence = candidate.get("confidence", 0.5)
+        cost       = candidate.get("cost", 1.0)
+
+        val, _  = self.valence.evaluate(hv)
+        novelty = self._novelty(hv)
+
+        return (
+            self.alpha * val +
+            self.beta  * novelty +
+            self.gamma * confidence -
+            self.delta * cost
+        )
+
+    def select(self, candidates: list) -> dict:
+        """
+        Select one action from a list of candidates via softmax sampling.
+        Each candidate is a dict with at minimum "hv" and "intent" keys.
+        Returns the selected candidate.
+        """
+        if not candidates:
+            raise ValueError("No candidates provided")
+
+        if len(candidates) == 1:
+            return candidates[0]
+
+        scores = np.array([self.score(c) for c in candidates])
+
+        # Softmax for probabilistic selection
+        exp_s  = np.exp(scores - scores.max())
+        probs  = exp_s / exp_s.sum()
+        idx    = int(np.random.choice(len(candidates), p=probs))
+
+        chosen = candidates[idx]
+        self._history.append({
+            "intent":     chosen.get("intent", "unknown"),
+            "score":      round(float(scores[idx]), 4),
+            "confidence": chosen.get("confidence", 0.5),
+        })
+        if len(self._history) > 100:
+            self._history.pop(0)
+
+        return chosen
+
+    def report(self) -> dict:
+        if not self._history:
+            return {"status": "No decisions made yet"}
+        avg_score = float(np.mean([h["score"] for h in self._history]))
+        return {
+            "total_decisions": len(self._history),
+            "avg_score":       round(avg_score, 4),
+            "recent":          self._history[-3:],
+        }

src/thalamus/__init__.py

@@ -0,0 +1 @@
+

src/thalamus/thalamus.py

@@ -0,0 +1,68 @@
+"""
+Synthetic Thalamus — Vitalis FSI
+
+Multimodal input router. Fuses text, audio, and internal
+signals into a single unified hypervector using temporal
+binding and cyclic permutation.
+
+All downstream systems receive the same representation space.
+"""
+import numpy as np
+import time
+from vitalis_ide.math_core.kernel import VitalisKernel
+
+
+def _permute(vec: np.ndarray, shift: int) -> np.ndarray:
+    return np.roll(vec, shift % len(vec))
+
+
+class SyntheticThalamus:
+    def __init__(self, dim: int = 10_000):
+        self.dim    = dim
+        self.kernel = VitalisKernel()
+        self._tick  = 0
+
+    def ingest(self, payload: dict) -> np.ndarray:
+        """
+        payload keys:
+            "text"     : str
+            "audio"    : Path to wav file
+            "internal" : pre-vectorized int8 hypervector
+        Returns unified binary hypervector.
+        """
+        self._tick += 1
+        vectors = []
+
+        if "text" in payload:
+            tokens = payload["text"].lower().split()
+            if tokens:
+                hv = self.kernel.vectorize_tokens(tokens, positional=False)
+                vectors.append(_permute(hv, self._tick))
+
+        if "audio" in payload:
+            try:
+                from src.audio_ear.feature_extractor import extract_features
+                from src.hdc_encoder.encoder import encode
+                mfcc, prosody = extract_features(payload["audio"])
+                hv = encode(mfcc, prosody)
+                vectors.append(_permute(hv, self._tick * 2))
+            except Exception:
+                pass
+
+        if "internal" in payload:
+            hv = np.array(payload["internal"], dtype=np.int8)
+            vectors.append(_permute(hv, self._tick * 3))
+
+        if not vectors:
+            return np.ones(self.dim, dtype=np.int8)
+
+        # Bundle all modalities
+        bundle = np.zeros(self.dim, dtype=np.int32)
+        for v in vectors:
+            bundle += v.astype(np.int32)
+        result = np.sign(bundle).astype(np.int8)
+        result[result == 0] = 1
+        return result
+
+    def ingest_text(self, text: str) -> np.ndarray:
+        return self.ingest({"text": text})

src/valence/__init__.py

@@ -0,0 +1 @@
+from .valence_engine import ValenceEngine

src/valence/valence_engine.py

@@ -0,0 +1,79 @@
+"""
+Valence Engine — Vitalis FSI
+
+Computes an affective score from a hypervector.
+Score in [-1, 1]:
+    -1 = strong negative (failure, frustration)
+     0 = neutral / unknown
+    +1 = strong positive (success, curiosity)
+
+Learned online from outcomes. No pretrained weights.
+"""
+import numpy as np
+import os
+from pathlib import Path
+
+
+class ValenceEngine:
+    LEARNING_RATE = 0.01
+    BUFFER_MAX    = 50
+    EPISODIC_BIAS = 0.3
+
+    def __init__(self, dim: int = 10_000):
+        self.dim    = dim
+        self.path   = Path.home() / ".vitalis_workspace" / "valence_weights.npy"
+        self.w      = self._load_weights()
+        self.buffer = []  # (hv, reward) tuples
+
+    def _load_weights(self) -> np.ndarray:
+        if self.path.exists():
+            return np.load(self.path)
+        return np.random.randn(self.dim) * 0.001
+
+    def _save_weights(self):
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        np.save(self.path, self.w)
+
+    def reinforce(self, hv: np.ndarray, reward: float):
+        """
+        Called after every outcome.
+        reward: +1.0 for success, -1.0 for failure, 0.0 for neutral.
+        """
+        hv   = hv.astype(np.float32)
+        pred = np.dot(self.w, hv) / self.dim
+        err  = reward - pred
+        self.w += self.LEARNING_RATE * err * hv
+        self.buffer.append((hv.copy(), reward))
+        if len(self.buffer) > self.BUFFER_MAX:
+            self.buffer.pop(0)
+        self._save_weights()
+
+    def evaluate(self, hv: np.ndarray) -> tuple:
+        """
+        Returns (valence, confidence).
+        confidence = |valence| — how certain the system is.
+        """
+        hv_f = hv.astype(np.float32)
+
+        # Resonance term
+        raw = float(np.tanh(np.dot(self.w, hv_f) / self.dim))
+
+        # Episodic bias from recent high-valence experiences
+        bias = 0.0
+        if self.buffer:
+            sims    = [float(np.mean(hv_f == bh)) for bh, _ in self.buffer]
+            weights = [bv for _, bv in self.buffer]
+            denom   = sum(sims) + 1e-9
+            bias    = float(np.dot(sims, weights) / denom)
+
+        valence    = float(np.clip(0.7 * raw + self.EPISODIC_BIAS * bias, -1.0, 1.0))
+        confidence = abs(valence)
+        return valence, confidence
+
+    def report(self) -> dict:
+        return {
+            "buffer_size":     len(self.buffer),
+            "avg_recent_reward": round(float(np.mean([r for _, r in self.buffer])), 4)
+            if self.buffer else 0.0,
+            "weight_norm":     round(float(np.linalg.norm(self.w)), 4),
+        }