Spaces:
Configuration error
Configuration error
| """ | |
| auto_tune.py — SR-61b: 2D Manifold Persistence | |
| ============================================================================= | |
| SR-61b Innovation: Persistent cognitive manifolds. | |
| This version implements: | |
| 1. 2D Hybrid Routing: Centroids in (Kurtosis, Phi) space. | |
| 2. Manifold Persistence: Save/Load learned centroids to JSON. | |
| 3. Scale-Adaptive Temperature: Sharpens zones based on model size. | |
| Centroid targets in Z-space (z_k, z_p): | |
| - math: (1.5, 0.5) -> High kurtosis, high stability | |
| - logic_a: (0.5, 0.2) -> Above mean | |
| - logic_b: (0.0, 0.0) -> Average | |
| - creative: (-1.0, -0.5) -> Below mean | |
| - synthesis: (-1.5, -1.0) -> Very flat | |
| """ | |
| import math | |
| import statistics | |
| import json | |
| import os | |
| from typing import Dict, Optional, Tuple | |
| def _sigmoid(x: float) -> float: | |
| """Numerically stable sigmoid function.""" | |
| x = max(-20.0, min(20.0, x)) | |
| return 1.0 / (1.0 + math.exp(-x)) | |
| def _dist2d(p1: Tuple[float, float], p2: Tuple[float, float], std_k: float, std_p: float) -> float: | |
| """Normalized 2D Euclidean distance.""" | |
| return math.sqrt(((p1[0] - p2[0]) / std_k)**2 + ((p1[1] - p2[1]) / std_p)**2) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # CONSTANTS & DEFAULTS | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| SCALE_DEFAULTS = { | |
| 640: dict(recur_start=5, recur_end=12, hub=10, n_loops=8, gamma=0.08), | |
| 1152: dict(recur_start=10, recur_end=20, hub=18, n_loops=8, gamma=0.12), | |
| # Gemma-4 E2B (hidden_size=1536, 35 layers) — 1B parity target (2026-06-09): | |
| # same n_loops=8 / gamma=0.12 as 1B (1152), with a wider recursion window | |
| # (recur_end=26) for the deeper 35-layer stack. Restored after e7f2942 | |
| # accidentally removed it; tests in test_gemma4_e2b_mock.py lock these values. | |
| 1536: dict(recur_start=10, recur_end=26, hub=18, n_loops=8, gamma=0.12), | |
| 2560: dict(recur_start=8, recur_end=22, hub=16, n_loops=6, gamma=0.05), | |
| 4096: dict(recur_start=10, recur_end=30, hub=20, n_loops=6, gamma=0.04), | |
| } | |
| ZONE_ROUTING = { | |
| 'math': dict(start=5, end=11, hub=10, loops=8), | |
| 'logic_a': dict(start=8, end=12, hub=10, loops=8), | |
| 'creative': dict(start=10, end=16, hub=10, loops=6), | |
| 'logic_b': dict(start=8, end=14, hub=10, loops=10), | |
| 'synthesis': dict(start=6, end=14, hub=10, loops=8), | |
| } | |
| # 2D Centroid targets in Z-space (z_kurtosis, z_phi) | |
| ZONE_Z_TARGETS = { | |
| 'math': (1.5, 0.5), | |
| 'logic_a': (0.5, 0.2), | |
| 'logic_b': (0.0, 0.0), | |
| 'creative': (-1.0, -0.5), | |
| 'synthesis': (-1.5, -1.0), | |
| } | |
| ZONE_Z_SIGMAS = { | |
| 'math': 0.8, | |
| 'logic_a': 0.6, | |
| 'creative': 1.0, | |
| 'logic_b': 0.7, | |
| 'synthesis': 0.9, | |
| } | |
| ONLINE_WARMUP = 5 | |
| MIN_TD_STD = 0.10 | |
| MIN_ONLINE_K_STD = 1.0 | |
| class AutoCalibrator: | |
| """Adaptive 2D Routing with Manifold Persistence. (SR-61b)""" | |
| def __init__(self, hidden_size: int, calibration_steps: int = 10, model_id: Optional[str] = None): | |
| self.hidden_size = hidden_size | |
| self.calibration_steps = calibration_steps | |
| self.model_id = model_id | |
| # Persistent state directory | |
| self.manifold_dir = "/run/media/julian/ML4/ollama-work/all_space/px_manifolds" | |
| self.calibrated = False | |
| self.k_samples = [] | |
| self.phi_samples = [] | |
| self.token_diversity_samples = [] | |
| # State | |
| self.k_mean = None | |
| self.k_std = None | |
| self.phi_mean = None | |
| self.phi_std = None | |
| self.token_diversity_mean = None | |
| self.token_diversity_std = None | |
| self.k_blend_weight = 0.8 if hidden_size == 640 else 0.5 | |
| self.zone_temperature = 0.8 | |
| self.learned_centroids: Dict[str, Tuple[float, float]] = {} | |
| # Online stats (Welford) | |
| self._online_n = 0 | |
| self._online_k_mean = 0.0 | |
| self._online_k_m2 = 0.0 | |
| # Try to load existing manifold | |
| if model_id: | |
| self.load_manifold() | |
| def collect(self, kurtosis: float, phi: float, token_diversity: Optional[float] = None, | |
| update_online: bool = False, token_len: int = 1): | |
| # SR-64b uses raw kurtosis | |
| k_norm = kurtosis | |
| if not self.calibrated: | |
| self.k_samples.append(k_norm) | |
| self.phi_samples.append(phi) | |
| if token_diversity is not None: | |
| self.token_diversity_samples.append(token_diversity) | |
| if len(self.k_samples) >= self.calibration_steps: | |
| self.calibrate() | |
| return True | |
| return False | |
| if update_online: | |
| self._update_online_stats(k_norm) | |
| return False | |
| def _update_online_stats(self, kurtosis: float): | |
| self._online_n += 1 | |
| delta = kurtosis - self._online_k_mean | |
| self._online_k_mean += delta / self._online_n | |
| delta2 = kurtosis - self._online_k_mean | |
| self._online_k_m2 += delta * delta2 | |
| def calibrate(self): | |
| """Compute 2D zone centroids and persist. (SR-61b)""" | |
| k_samples = [k for k in self.k_samples if math.isfinite(k)] | |
| phi_samples = [p for p in self.phi_samples if math.isfinite(p)] | |
| td_samples = [t for t in self.token_diversity_samples if math.isfinite(t)] | |
| if len(k_samples) < 2: return | |
| self.k_mean = statistics.mean(k_samples) | |
| self.k_std = max(statistics.stdev(k_samples), 5.0) | |
| self.phi_mean = statistics.mean(phi_samples) if len(phi_samples) >= 2 else 0.9 | |
| self.phi_std = max(statistics.stdev(phi_samples), 0.01) if len(phi_samples) >= 2 else 0.05 | |
| # Learn 2D centroids | |
| self.learned_centroids = {} | |
| for zone, (zk, zp) in ZONE_Z_TARGETS.items(): | |
| raw_k = self.k_mean + zk * self.k_std | |
| raw_p = self.phi_mean + zp * self.phi_std | |
| self.learned_centroids[zone] = (raw_k, raw_p) | |
| # Token diversity stats | |
| if td_samples: | |
| self.token_diversity_mean = statistics.mean(td_samples) | |
| self.token_diversity_std = max(statistics.stdev(td_samples) if len(td_samples) > 1 else 0.0, MIN_TD_STD) | |
| # Scale-adaptive temperature (SR-59h) | |
| k_cv = self.k_std / (abs(self.k_mean) + 1e-9) | |
| if k_cv > 0.05: self.zone_temperature = 0.3 | |
| elif k_cv > 0.01: self.zone_temperature = 0.6 | |
| else: self.zone_temperature = 1.0 | |
| # Initialize online stats | |
| self._online_n = len(k_samples) | |
| self._online_k_mean = self.k_mean | |
| self._online_k_m2 = (self.k_std ** 2) * self._online_n | |
| self.calibrated = True | |
| self.save_manifold() | |
| def _get_kurtosis_weights(self, kurtosis: float, phi: float) -> Dict[str, float]: | |
| """Compute 2D zone weights using Euclidean distance in manifold space.""" | |
| if not self.learned_centroids: | |
| return {z: 1.0/len(ZONE_Z_TARGETS) for z in ZONE_Z_TARGETS} | |
| # Use online mean for kurtosis z-score if available | |
| k_center = self._online_k_mean if self._online_n >= ONLINE_WARMUP else self.k_mean | |
| k_std_eff = math.sqrt(self._online_k_m2 / max(self._online_n - 1, 1)) if self._online_n > 1 else self.k_std | |
| k_std_eff = max(k_std_eff, 1.0) | |
| weights = {} | |
| temp = self.zone_temperature | |
| for zone, (ck, cp) in self.learned_centroids.items(): | |
| sigma = ZONE_Z_SIGMAS[zone] * temp | |
| # Calculate 2D distance normalized by local manifold density | |
| dist = _dist2d((kurtosis, phi), (ck, cp), k_std_eff, self.phi_std) | |
| weights[zone] = math.exp(-0.5 * (dist / sigma)**2) | |
| W = sum(weights.values()) + 1e-9 | |
| return {k: v / W for k, v in weights.items()} | |
| def _compute_scf_weights(self, token_diversity: float) -> Optional[Dict[str, float]]: | |
| if self.token_diversity_mean is None or token_diversity is None: | |
| return None | |
| z_td = (token_diversity - self.token_diversity_mean) / (self.token_diversity_std + 1e-9) | |
| phi_signal = _sigmoid(-z_td) | |
| weights = { | |
| 'math': phi_signal ** 2, | |
| 'logic_a': phi_signal * (1 - phi_signal) * 2, | |
| 'creative': (1 - phi_signal) ** 2, | |
| 'logic_b': phi_signal * (1 - phi_signal), | |
| 'synthesis': (1 - phi_signal) * phi_signal * 0.5, | |
| } | |
| W = sum(weights.values()) + 1e-9 | |
| return {k: v / W for k, v in weights.items()} | |
| def get_zone_weights(self, kurtosis: float, phi: Optional[float] = None, token_diversity: Optional[float] = None, token_len: int = 1) -> Dict[str, float]: | |
| # SR-64b uses raw kurtosis | |
| k_norm = kurtosis | |
| # SR-61b: Fallback for None phi (first token of session) | |
| if phi is None: | |
| phi = self.phi_mean if self.phi_mean is not None else 0.9 | |
| k_weights = self._get_kurtosis_weights(k_norm, phi) | |
| scf_weights = self._compute_scf_weights(token_diversity) | |
| if scf_weights is None: return k_weights | |
| blend = self.k_blend_weight | |
| blended = {z: blend * k_weights[z] + (1 - blend) * scf_weights[z] for z in k_weights} | |
| W = sum(blended.values()) + 1e-9 | |
| return {k: v / W for k, v in blended.items()} | |
| def classify_zone(self, kurtosis: float, phi: Optional[float] = None, token_diversity: Optional[float] = None, token_len: int = 1) -> str: | |
| weights = self.get_zone_weights(kurtosis, phi, token_diversity, token_len=token_len) | |
| return max(weights, key=weights.get) | |
| def get_routing_params(self, kurtosis: float, phi: Optional[float] = None, hidden_size: Optional[int] = None, token_diversity: Optional[float] = None, token_len: int = 1) -> Dict[str, any]: | |
| weights = self.get_zone_weights(kurtosis, phi, token_diversity, token_len=token_len) | |
| start = sum(weights[z] * ZONE_ROUTING[z]['start'] for z in weights) | |
| end = sum(weights[z] * ZONE_ROUTING[z]['end'] for z in weights) | |
| hub = sum(weights[z] * ZONE_ROUTING[z]['hub'] for z in weights) | |
| loops = sum(weights[z] * ZONE_ROUTING[z]['loops'] for z in weights) | |
| # Scale adjustments | |
| if hidden_size and hidden_size in SCALE_DEFAULTS: | |
| defaults = SCALE_DEFAULTS[hidden_size] | |
| b = 0.3 | |
| start = b * defaults['recur_start'] + (1-b) * start | |
| end = b * defaults['recur_end'] + (1-b) * end | |
| hub = b * defaults['hub'] + (1-b) * hub | |
| loops = b * defaults['n_loops'] + (1-b) * loops | |
| return { | |
| 'dynamic_start': max(1, int(round(start))), | |
| 'dynamic_end': max(int(round(start))+2, int(round(end))), | |
| 'dynamic_hub': max(int(round(start)), min(int(round(end)), int(round(hub)))), | |
| 'n_loops': max(1, int(round(loops))), | |
| } | |
| def save_manifold(self): | |
| if not self.model_id: return | |
| os.makedirs(self.manifold_dir, exist_ok=True) | |
| safe_id = self.model_id.replace("/", "_") | |
| path = os.path.join(self.manifold_dir, f"{safe_id}_manifold.json") | |
| data = { | |
| "k_mean": self.k_mean, "k_std": self.k_std, | |
| "phi_mean": self.phi_mean, "phi_std": self.phi_std, | |
| "learned_centroids": self.learned_centroids, | |
| "calibrated": self.calibrated, | |
| "k_blend_weight": self.k_blend_weight, | |
| "zone_temperature": self.zone_temperature, | |
| "token_diversity_mean": self.token_diversity_mean, | |
| "token_diversity_std": self.token_diversity_std | |
| } | |
| with open(path, "w") as f: json.dump(data, f, indent=2) | |
| def load_manifold(self): | |
| if not self.model_id: return | |
| safe_id = self.model_id.replace("/", "_") | |
| path = os.path.join(self.manifold_dir, f"{safe_id}_manifold.json") | |
| if not os.path.exists(path): return | |
| try: | |
| with open(path, "r") as f: data = json.load(f) | |
| self.k_mean = data.get("k_mean") | |
| self.k_std = data.get("k_std") | |
| self.phi_mean = data.get("phi_mean") | |
| self.phi_std = data.get("phi_std") | |
| self.learned_centroids = data.get("learned_centroids", {}) | |
| self.calibrated = data.get("calibrated", False) | |
| self.k_blend_weight = data.get("k_blend_weight", self.k_blend_weight) | |
| self.zone_temperature = data.get("zone_temperature", self.zone_temperature) | |
| self.token_diversity_mean = data.get("token_diversity_mean") | |
| self.token_diversity_std = data.get("token_diversity_std") | |
| if self.calibrated: | |
| self._online_n = ONLINE_WARMUP + 1 | |
| self._online_k_mean = self.k_mean | |
| self._online_k_m2 = (self.k_std ** 2) * self._online_n | |
| print(f"[AutoCalibrator] Loaded manifold for {self.model_id} (T={self.zone_temperature:.2f})") | |
| except: pass | |
| def status(self) -> Dict[str, any]: | |
| k_cv = self.k_std / (abs(self.k_mean) + 1e-9) if self.k_mean else None | |
| return { | |
| 'calibrated': self.calibrated, 'model_id': self.model_id, | |
| 'k_mean': self.k_mean, 'k_std': self.k_std, 'k_cv': k_cv, | |
| 'phi_mean': self.phi_mean, 'phi_std': self.phi_std, | |
| 'learned_centroids': self.learned_centroids, | |
| 'zone_temperature': self.zone_temperature, | |
| 'online_n': self._online_n, | |
| 'routing_mode': 'online_2d' if self._online_n >= ONLINE_WARMUP else 'calibration_2d', | |
| } | |