Upload phase_e_callosum.py with huggingface_hub

phase_e_callosum.py

@@ -0,0 +1,140 @@
+"""
+Phase E — Corpus Callosum (الجسم الثفني): the bi-hemispheric bridge.
+
+Right hemisphere  : the REAL artifacts of `bi-hemispheric-brain-5m`
+                    (superbrain_engram_matrix.npz = 243 engrams x 5,000,000
+                     sparse neurons; knowledge_bank_superbrain.json).
+Left  hemisphere  : LHC-0 (trained bilingual encoder + symbolic workspace).
+
+The right hemisphere stores its engrams in a 5M-neuron coordinate system but
+the original query encoder was never published -> the hemisphere is a SEALED
+VAULT for new queries. The corpus callosum is a LEARNED translator between
+the two coordinate systems:
+
+    e_left (128-d semantic)  --W-->  y_right (k-dim, k = neurons actually
+                                        used by the real rows, ≈3.6K)
+
+fitted by dual ridge regression on the 243 (prompt, real-row) pairs.
+Without the callosum the right hemisphere cannot be read at all (ablation).
+"""
+import json
+
+import numpy as np
+
+
+class RightHemisphere:
+    """Reads the real engram matrix exactly as stored (CSR arrays)."""
+
+    def __init__(self, npz_path, kb_path):
+        z = np.load(npz_path, allow_pickle=True)
+        self.indptr = z["indptr"]
+        self.indices = z["indices"]
+        self.data = z["data"].astype(np.float32)
+        self.shape = tuple(z["shape"])                 # (243, 5_000_000)
+        self.entries = json.load(open(kb_path, encoding="utf-8"))
+        assert len(self.entries) == self.shape[0], "rows must match KB"
+
+        # rows live in a small neuron subspace -> local coordinates
+        self.global_ids = np.unique(self.indices)      # sorted unique neurons
+        remap = {int(g): i for i, g in enumerate(self.global_ids)}
+        self.k = len(self.global_ids)
+        self.Y = np.zeros((self.shape[0], self.k), dtype=np.float32)
+        for r in range(self.shape[0]):
+            s, e = self.indptr[r], self.indptr[r + 1]
+            for g, v in zip(self.indices[s:e], self.data[s:e]):
+                self.Y[r, remap[int(g)]] = v
+
+    def describe(self):
+        return {"rows": int(self.shape[0]), "neuron_space": int(self.shape[1]),
+                "active_neurons_used": int(self.k),
+                "stored_synapses": int(self.indices.size)}
+
+
+class CorpusCallosum:
+    """Learned cross-space translator (kernel ridge / dual form):
+
+        A = (K + αI)^{-1} Y ,  K = X X^T      (n×n, tiny)
+        ŷ(x) = (x · X^T) A                    (soft blend of learned columns)
+
+    X = left-hemisphere representation (sensory n-gram features work best:
+    both spaces are lexical at this scale; 128-d semantic also supported).
+    Stores X and A only — no giant W."""
+
+    def __init__(self, alpha=0.5):
+        self.alpha = alpha
+        self.X = self.A = None
+
+    def fit(self, X_left, Y_right):
+        X = np.asarray(X_left, dtype=np.float32)
+        n = X.shape[0]
+        K = X @ X.T + self.alpha * np.eye(n, dtype=np.float32)
+        self.A = np.linalg.solve(K, Y_right.astype(np.float32))   # (n, k)
+        self.X = X
+        return self
+
+    def translate(self, x_left):
+        sims = np.asarray(x_left, dtype=np.float32) @ self.X.T      # (n,)
+        y = sims @ self.A                                           # (k,)
+        n = float(np.linalg.norm(y))
+        return y / n if n > 0 else y
+
+    def translate_batch(self, X):
+        sims = np.asarray(X, dtype=np.float32) @ self.X.T           # (m, n)
+        Y = sims @ self.A
+        norms = np.linalg.norm(Y, axis=1, keepdims=True)
+        return Y / np.maximum(norms, 1e-8)
+
+
+class BiCallosum:
+    """The real corpus callosum carries TWO streams (Hickok & Poeppel dual-stream
+    language model):
+      • dorsal  channel: sensory lexical features -> right codes (precision)
+      • ventral channel: semantic embeddings     -> right codes (meaning, cross-lingual)
+    Right-hemisphere score = average of both channels' normalized scores."""
+
+    def __init__(self, alpha=0.5):
+        self.dorsal = CorpusCallosum(alpha)     # features (n-grams)
+        self.ventral = CorpusCallosum(alpha)    # 128-d semantic
+
+    def fit(self, X_feat, E_sem, Y_right):
+        self.dorsal.fit(X_feat, Y_right)
+        self.ventral.fit(E_sem, Y_right)
+        return self
+
+    def right_scores(self, x_feat, e_sem, Y_right):
+        """Each channel votes with its confidence weight (top1−top2 gap) —
+        Basal-Ganglia/ACC style selection between the two callosal streams."""
+        def gap(s):
+            s2 = np.sort(s)[::-1]
+            if len(s2) < 2:
+                return 1.0
+            span = float(s2[0] - s2[-1]) + 1e-6        # scale-invariant
+            return float(s2[0] - s2[1]) / span
+        Yn = Y_right / np.maximum(np.linalg.norm(Y_right, axis=1, keepdims=True), 1e-8)
+        s_d = self.dorsal.translate(x_feat) @ Yn.T
+        s_v = self.ventral.translate(e_sem) @ Yn.T
+        g_d, g_v = gap(s_d) + 1e-6, gap(s_v) + 1e-6
+        return (g_d * s_d + g_v * s_v) / (g_d + g_v)
+
+
+def right_scores(callosal_codes, Y_right):
+    """cosine scores of a translated code against the real engram rows."""
+    Yn = Y_right / np.maximum(np.linalg.norm(Y_right, axis=1, keepdims=True), 1e-8)
+    return callosal_codes @ Yn.T
+
+
+def fuse(s_left, s_right, has_right=None, lam=0.5):
+    """Confidence-weighted fusion (ACC-style): each hemisphere votes with its
+    top1−top2 gap; entries the right hemisphere never stored vote left-only."""
+    def gap(s):
+        s2 = np.sort(s)[::-1]
+        if len(s2) < 2:
+            return 1.0
+        span = float(s2[0] - s2[-1]) + 1e-6            # scale-invariant
+        return float(s2[0] - s2[1]) / span
+    if has_right is None:
+        has_right = np.ones(len(s_left), dtype=bool)
+    g_l = gap(s_left) + 1e-6
+    g_r = (gap(s_right) + 1e-6) * has_right.astype(np.float64)
+    w = g_l + g_r
+    return (g_l * s_left + g_r * s_right) / w