| |
| """ |
| R5-compliant saturation module. |
| |
| Cell = group of neurons in an MLP layer. |
| 6D knowledge vector K = [L, W, H, D, T, E] |
| |
| Dimensions (per R5 paper, Section 3.4): |
| K1 (L) - sequential coverage (long-range reach) |
| K2 (W) - domain breadth (number of domains activated) [needs domain labels] |
| K3 (H) - abstraction level (low entropy = focused) |
| K4 (D) - reasoning depth (effective path length) [needs plexus graph] |
| K5 (T) - temporal freshness (exponential decay by age) |
| K6 (E) - encompassment (full facet coverage) [needs facets.json] |
| |
| S = ||K||_omega (weighted L2, omega in simplex) |
| S* = 0.85 (split threshold), T_cons = 3 (consecutive steps) |
| """ |
| import json |
| import math |
| from dataclasses import dataclass, field, asdict |
| from pathlib import Path |
| from typing import Optional |
| import numpy as np |
|
|
| N_CELLS_PER_LAYER = 4 |
| |
| MAX_CELLS_TOTAL = 5000 |
| S_STAR = 0.70 |
| T_CONSECUTIVE = 3 |
| DECAY_LAMBDA = 0.001 |
|
|
|
|
| @dataclass |
| class Cell: |
| cell_id: str |
| layer_idx: int |
| neuron_start: int |
| neuron_end: int |
| K: list = field(default_factory=lambda: [0.0] * 6) |
| omega: list = field(default_factory=lambda: [1/6] * 6) |
| S: float = 0.0 |
| S_history: list = field(default_factory=list) |
| K_history: list = field(default_factory=list) |
| state: str = "active" |
| splits: int = 0 |
| parent_id: Optional[str] = None |
| created_at: str = "" |
| last_updated_step: int = 0 |
| fp_history: list = field(default_factory=list) |
| pattern_sig: str = "" |
| last_split_step: int = 0 |
|
|
| def num_neurons(self) -> int: |
| return self.neuron_end - self.neuron_start |
|
|
| def compute_K1_sequential_coverage(activations: np.ndarray) -> float: |
| """ |
| K1 (L): sequential coverage / long-range reach. |
| activations: shape [seq_len, num_neurons] for ONE sample/cell. |
| Returns: average normalized range of positions where neurons are active. |
| High K1 = cell responds to tokens spread across long range. |
| Low K1 = cell responds only to nearby tokens. |
| """ |
| if activations.size == 0: |
| return 0.0 |
| a = np.abs(np.asarray(activations, dtype=np.float32)) |
| seq_len = a.shape[0] |
| if seq_len < 2: |
| return 0.0 |
| |
| active = a > a.mean(axis=0, keepdims=True) |
| spans = [] |
| for j in range(a.shape[1]): |
| idx = np.where(active[:, j])[0] |
| if len(idx) >= 2: |
| spans.append((idx[-1] - idx[0]) / (seq_len - 1)) |
| if not spans: |
| return 0.0 |
| return float(np.clip(np.mean(spans), 0.0, 1.0)) |
|
|
| def compute_K3_abstraction(activations: np.ndarray) -> float: |
| """ |
| K3 (H): abstraction level — activation sparsity. |
| |
| FIX (2026-10-02): Entropy-based formula gives ~0 with dense swish activations. |
| New formula: Gini-style sparsity = fraction of "concentration" in top neurons. |
| A cell that fires on few neurons strongly = high K3 (abstraction). |
| A cell that fires uniformly = low K3 (diffuse). |
| """ |
| if activations.size == 0: |
| return 0.0 |
| a = np.abs(np.asarray(activations, dtype=np.float32)) |
| |
| p = a.mean(axis=0) |
| n = len(p) |
| if n < 2: |
| return 1.0 |
| total = p.sum() |
| if total < 1e-9: |
| return 0.0 |
| |
| p_sorted = np.sort(p) |
| idx = np.arange(1, n + 1) |
| gini = (2 * np.sum(idx * p_sorted)) / (n * total) - (n + 1) / n |
| gini = float(np.clip(gini, 0.0, 1.0)) |
| return gini |
|
|
| def compute_K5_temporal_freshness( |
| last_updated_step: int, |
| current_step: int, |
| decay_lambda: float = DECAY_LAMBDA, |
| ) -> float: |
| """ |
| K5 (T): temporal freshness. |
| Exponential decay by age. |
| Recently-updated cells: K5 near 1. |
| Old cells: K5 near 0. |
| """ |
| |
| if last_updated_step == 0: |
| return 1.0 |
| age = max(0, current_step - last_updated_step) |
| |
| age = min(age, 500) |
| return float(math.exp(-decay_lambda * age)) |
|
|
| def compute_K2_domain_breadth(activations: np.ndarray, domains=None) -> float: |
| """K2 (W): domain breadth. TODO: needs domain labels per sample.""" |
| return 0.0 |
|
|
|
|
| def compute_K4_reasoning_depth(activations: np.ndarray, plexus=None) -> float: |
| """K4 (D): reasoning depth. TODO: needs plexus graph.""" |
| return 0.0 |
|
|
|
|
| def compute_K6_encompassment(activations: np.ndarray, facets=None) -> float: |
| """K6 (E): encompassment. TODO: needs facets.json taxonomy.""" |
| return 0.0 |
|
|
|
|
| def compute_S(K: list, omega: list) -> float: |
| """ |
| Weighted L2 norm of K over ACTIVE dimensions only. |
| |
| FIX (2026-10-02): With K2=K4=K6=0 (stubs), max possible S was 0.707 < S*=0.85. |
| Renormalizing omega over non-zero K dims makes S in [0,1] again. |
| When all 6 dims are implemented, this reduces to the original formula. |
| """ |
| k = np.asarray(K, dtype=np.float32) |
| w_final = np.array([0.375, 0.0, 0.10, 0.0, 0.375, 0.15], dtype=np.float32) |
| w_final = w_final / (w_final.sum() + 1e-9) |
| return float(np.sqrt(np.sum(w_final * k * k))) |
|
|
|
|
| def save_cells(cells: list, path: Path): |
| path = Path(path) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text(json.dumps([asdict(c) for c in cells], indent=2)) |
| return path |
|
|
|
|
| def load_cells(path: Path) -> list: |
| data = json.loads(Path(path).read_text()) |
| cells = [] |
| for d in data: |
| |
| try: |
| cells.append(Cell(**d)) |
| except TypeError: |
| |
| from dataclasses import fields |
| valid = {f.name for f in fields(Cell)} |
| filtered = {k: v for k, v in d.items() if k in valid} |
| cells.append(Cell(**filtered)) |
| return cells |
|
|
|
|
| if __name__ == "__main__": |
| a = np.random.randn(128, 1536).astype(np.float32) |
| K1 = compute_K1_sequential_coverage(a) |
| K3 = compute_K3_abstraction(a) |
| K5 = compute_K5_temporal_freshness(0, 1000) |
| print(f"K1 = {K1:.3f}") |
| print(f"K3 = {K3:.3f}") |
| print(f"K5 = {K5:.3f}") |
| K = [K1, 0.0, K3, 0.0, K5, 0.0] |
| print(f"S = {compute_S(K, [1/6]*6):.3f}") |
|
|
|
|
| def load_facets(path=None) -> dict: |
| """Load facets taxonomy from facets.json.""" |
| if path is None: |
| path = Path(__file__).parent / "facets.json" |
| data = json.loads(Path(path).read_text()) |
| return data.get("domains", {}) |
|
|
|
|
| def compute_K6_encompassment_real( |
| activated_facets=None, |
| domain=None, |
| facets_map=None, |
| fp_history=None, |
| similarity_threshold=0.7, |
| ) -> float: |
| """ |
| K6 (E): encompassment — coverage of distinct knowledge patterns. |
| |
| Bottom-up design (2026-10-02): K6 grows when a cell responds to |
| multiple distinct activation patterns over time. |
| |
| Logic: |
| - Each pattern is a binary fingerprint of top-K neurons. |
| - Count distinct patterns (by Hamming similarity < threshold). |
| - K6 = min(1.0, distinct_patterns / TARGET_PATTERNS) |
| - TARGET_PATTERNS = 5 (empirical; adjust if needed) |
| """ |
| if fp_history is None or len(fp_history) < 3: |
| return 0.0 |
|
|
| try: |
| from cognitive_log import hamming_similarity |
| except ImportError: |
| import sys |
| sys.path.insert(0, "/data/berna-r7/core/code") |
| from cognitive_log import hamming_similarity |
|
|
| TARGET = 5 |
| unique = [] |
| for fp in fp_history: |
| is_new = True |
| for u in unique: |
| if hamming_similarity(fp, u) > similarity_threshold: |
| is_new = False |
| break |
| if is_new: |
| unique.append(fp) |
| if len(unique) >= TARGET: |
| return 1.0 |
| return float(len(unique) / TARGET) |
|
|
|
|
| def compute_K4_from_plexus(cell_id: str, plexus) -> float: |
| """K4 (D) via plexus graph. Wrapper for plexus.compute_K4.""" |
| if plexus is None: |
| return 0.0 |
| try: |
| from .plexus import compute_K4 as _k4 |
| return _k4(plexus, cell_id) |
| except Exception: |
| return 0.0 |
|
|
