Copy nexus_os_v2/twave_tracker.py from dataset for module imports

nexus_os_v2/twave_tracker.py

@@ -0,0 +1,370 @@
+"""
+TWAVE — Token-Level Wavefront Tracker
+Bose-Einstein Condensate Thermodynamic Control for LLM Inference
+
+Implements:
+  - Landau-Ginzburg free energy functional F[ψ]
+  - Bogoliubov excitation spectrum E(k)
+  - Healing length ξ as hallucination localization scale
+  - Jarzynski equality for non-equilibrium reflection trigger
+  - CK-PLUG Confidence Gain as concrete μ_ret coupling
+  - Stochastic resonance optimal T_eff ≈ 0.8 T_c
+
+Physics foundation:
+  Kim 2602.08216 (Thermodynamic Isomorphism)
+  Arnold et al. (Phase Transitions in Output Distributions)
+  Qian et al. 2604.10219 (Cognitive Pivot Points / RVTD)
+  CK-PLUG 2503.15888 (Confidence Gain = H(parametric) - H(retrieval))
+  BEC standard physics (Bogoliubov, healing length)
+  Jarzynski equality (fluctuation theorem for non-equilibrium work)
+"""
+import math
+from typing import List, Tuple, Optional, Dict, Any
+from dataclasses import dataclass, field
+import numpy as np
+
+
+@dataclass
+class TokenState:
+    """Thermodynamic state at a single token position."""
+    position: int
+    entropy: float                    # H_i — Shannon entropy (nats)
+    entropy_max: float                # log(vocab_size) — max possible entropy
+    coherence: float                  # 1 - H_i/H_max (condensate fraction)
+    visual_attention_mass: float      # A_i^vis (from V-STAR, 0-1)
+    reward_density: float             # ρ_i = r_i / i (intensive reward)
+    critique_confidence: float        # c_i (from Critique-out-Loud)
+    retrieval_mu: float               # μ_ret(x) — CK-PLUG derived chemical potential
+    psi: float                        # |ψ| — BEC order parameter amplitude
+    free_energy_density: float        # f(x) — Landau-Ginzburg density
+    temperature_eff: float            # T_eff(x) — local effective temperature
+    specific_heat: float              # C_V^(i) — susceptibility
+    E_excitation: float               # Bogoliubov excitation energy
+    k_local: float                    # Local wavenumber |∇ψ|
+    jarzynski_work: float             # W_i — cumulative non-equilibrium work
+
+
+@dataclass
+class GenerationTrajectory:
+    """Complete tracked generation with thermodynamic metadata."""
+    tokens: List[int]
+    states: List[TokenState]
+    total_work: float
+    jarzynski_bound: float
+    reflection_triggers: List[int]     # Positions where reflection fired
+    grounded_score: float              # Overall retrieval grounding quality
+    hallucination_risk: float          # 0-1 composite risk score
+
+
+class TWAVETracker:
+    """
+    Token-Level Wavefront Expansion Tracker.
+    Monitors generation as a 1D statistical field evolving under
+    Landau-Ginzburg dynamics with BEC order parameter ψ(x).
+    """
+
+    def __init__(
+        self,
+        T_c: float = 1.0,           # Critical temperature (calibrated per-model)
+        mu_0: float = 0.5,          # Base retrieval chemical potential
+        kappa: float = 0.1,         # Healing cost coefficient (ħ²/2m analog)
+        b_quartic: float = 0.5,     # Quartic stability term
+        gamma_drift: float = 0.1,   # Field evolution step size
+        jarzynski_threshold: float = -4.6,  # Fluctuation theorem bound (99%)
+        tau_transition: float = 0.5,  # CG→μ_ret transition width
+        vocab_size: int = 32000,
+    ):
+        self.T_c = T_c
+        self.mu_0 = mu_0
+        self.kappa = kappa
+        self.b = b_quartic
+        self.gamma = gamma_drift
+        self.jarzynski_threshold = jarzynski_threshold
+        self.tau = tau_transition
+        self.vocab_size = vocab_size
+        self.H_max = math.log(vocab_size)
+
+    def compute_entropy(self, probs: np.ndarray) -> float:
+        """Shannon entropy in nats: H = -Σ p_i ln p_i."""
+        p = probs[probs > 0]
+        return float(-np.sum(p * np.log(p)))
+
+    def compute_coherence(self, entropy: float) -> float:
+        """Condensate fraction: ψ coherence = 1 - H/H_max."""
+        return max(0.0, 1.0 - entropy / self.H_max)
+
+    def compute_chemical_potential(self, CG: float) -> float:
+        """
+        Map CK-PLUG Confidence Gain to μ_ret.
+        μ_ret = μ_0 * tanh(CG / τ)
+          CG >> 0  → retrieval supports → μ_ret → μ_0
+          CG ≈ 0   → neutral → μ_ret → 0
+          CG << 0  → retrieval conflicts → μ_ret → -μ_0
+        """
+        return self.mu_0 * math.tanh(CG / self.tau)
+
+    def compute_order_parameter(self, coherence: float, mu_ret: float) -> float:
+        """
+        BEC ground state amplitude (minimizing Landau-Ginzburg functional):
+        ψ_0 = √[(a_0(T_c - T_eff) + μ_ret) / b]
+        where a(T) = coherence - T_c (since coherence ∝ 1 - entropy ∝ 1/T)
+        """
+        a = coherence - self.T_c
+        numerator = max(0.0, a + mu_ret)
+        return math.sqrt(numerator / self.b)
+
+    def compute_free_energy_density(self, psi: float, coherence: float, mu_ret: float) -> float:
+        """
+        Landau-Ginzburg free energy density:
+        f = a(T)|ψ|² + (b/2)|ψ|⁴ - μ_ret|ψ|²
+        """
+        a = coherence - self.T_c
+        return a * psi**2 + 0.5 * self.b * psi**4 - mu_ret * psi**2
+
+    def compute_bogoliubov_energy(self, psi: float, k_local: float, mu_ret: float) -> float:
+        """
+        Bogoliubov excitation spectrum:
+        E(k) = √[(κk²)² + (c_s·k)²] where c_s = √(μ/m) = √(μ_ret)
+        In our analogy: κ = kappa, c_s² = mu_ret
+        """
+        c_s_sq = max(0.0, mu_ret)
+        term1 = (self.kappa * k_local**2)**2
+        term2 = c_s_sq * k_local**2
+        return math.sqrt(term1 + term2)
+
+    def compute_healing_length(self, T_eff: float, mu_ret: float) -> float:
+        """
+        Healing length: ξ = ħ / √(2m·μ) = 1 / √(2·mu_ret) [in token units]
+        As T_eff → T_c or μ_ret → 0: ξ → ∞ (hallucination propagates globally)
+        """
+        mu = max(1e-10, mu_ret)
+        return 1.0 / math.sqrt(2.0 * mu)
+
+    def compute_specific_heat(self, entropy_history: List[float]) -> float:
+        """
+        Specific heat analog: C_V = ∂⟨H⟩/∂T_eff.
+        Approximated numerically from entropy history.
+        """
+        if len(entropy_history) < 3:
+            return 0.0
+        # Finite difference: dH/dt ≈ (H_{t} - H_{t-2}) / 2
+        dH = entropy_history[-1] - entropy_history[-3]
+        # Effective temperature from mean entropy
+        H_mean = sum(entropy_history[-3:]) / 3.0
+        T_eff = self.T_c * (1.0 - H_mean / self.H_max)
+        if abs(T_eff) < 1e-10:
+            return 0.0
+        return dH / (2.0 * T_eff)
+
+    def compute_jarzynski_work(self, log_prob_policy: float, log_prob_ref: float) -> float:
+        """
+        Non-equilibrium information work per token:
+        W_i = -log[π_θ(x_i|x_{<i}) / π_ref(x_i|x_{<i})]
+        """
+        return -(log_prob_policy - log_prob_ref)
+
+    def check_jarzynski_bound(self, cumulative_work: float, beta_eff: float) -> bool:
+        """
+        Jarzynski reflection criterion:
+        Returns True if trajectory violates fluctuation theorem
+        (hallucination detected with 99% confidence).
+        """
+        if beta_eff <= 0:
+            return False
+        jarzynski = math.exp(-beta_eff * cumulative_work)
+        threshold = math.exp(-beta_eff * self.jarzynski_threshold)
+        return jarzynski < threshold
+
+    def update_state(
+        self,
+        position: int,
+        probs: np.ndarray,
+        log_prob_policy: float,
+        log_prob_ref: float,
+        CG: float,                        # CK-PLUG Confidence Gain
+        visual_attention: float = 1.0,    # A_i^vis (1.0 = full attention)
+        reward_density: float = 0.0,
+        critique_confidence: float = 0.5,
+        prev_psi: float = 0.0,
+    ) -> TokenState:
+        """
+        Compute full thermodynamic state at token position i.
+        """
+        # 1. Entropy and coherence
+        H = self.compute_entropy(probs)
+        coherence = self.compute_coherence(H)
+
+        # 2. Effective temperature (stochastic resonance optimal point)
+        # T_eff = T_c * (1 - H/H_max) = T_c * coherence
+        T_eff = self.T_c * coherence
+
+        # 3. Chemical potential from CK-PLUG
+        mu_ret = self.compute_chemical_potential(CG)
+
+        # 4. Order parameter ψ
+        psi = self.compute_order_parameter(coherence, mu_ret)
+
+        # 5. Local wavenumber (gradient of ψ)
+        k_local = abs(psi - prev_psi) if prev_psi > 0 else 0.0
+
+        # 6. Free energy density
+        f_density = self.compute_free_energy_density(psi, coherence, mu_ret)
+
+        # 7. Bogoliubov excitation energy
+        E_exc = self.compute_bogoliubov_energy(psi, k_local, mu_ret)
+
+        # 8. Specific heat
+        # (requires entropy history — computed externally and passed back)
+        C_V = 0.0  # Placeholder; set by caller with history
+
+        # 9. Jarzynski work
+        W_i = self.compute_jarzynski_work(log_prob_policy, log_prob_ref)
+
+        return TokenState(
+            position=position,
+            entropy=H,
+            entropy_max=self.H_max,
+            coherence=coherence,
+            visual_attention_mass=visual_attention,
+            reward_density=reward_density,
+            critique_confidence=critique_confidence,
+            retrieval_mu=mu_ret,
+            psi=psi,
+            free_energy_density=f_density,
+            temperature_eff=T_eff,
+            specific_heat=C_V,
+            E_excitation=E_exc,
+            k_local=k_local,
+            jarzynski_work=W_i,
+        )
+
+    def evaluate_stability(self, state: TokenState) -> Dict[str, Any]:
+        """
+        Evaluate generation stability at current token.
+        Returns actionable flags for the inference loop.
+        """
+        flags = {
+            "stable": True,
+            "reflection_triggered": False,
+            "hallucination_risk": "low",
+            "action": "continue",
+        }
+
+        # Critical point: T_eff > T_c or μ_ret < 0 (adversarial retrieval)
+        if state.temperature_eff > self.T_c:
+            flags["stable"] = False
+            flags["hallucination_risk"] = "critical"
+            flags["action"] = "reflection"
+            flags["reflection_triggered"] = True
+            return flags
+
+        # Bogoliubov gap check: E_exc < μ_ret * 0.95 → near-critical
+        gap_threshold = state.retrieval_mu * 0.95 if state.retrieval_mu > 0 else 0.01
+        if state.E_excitation < gap_threshold:
+            flags["stable"] = False
+            flags["hallucination_risk"] = "high"
+            flags["action"] = "grounding_boost"
+            return flags
+
+        # Healing length divergence: ξ > 10 tokens → global perturbation
+        xi = self.compute_healing_length(state.temperature_eff, state.retrieval_mu)
+        if xi > 10.0:
+            flags["hallucination_risk"] = "elevated"
+            flags["action"] = "local_grounding"
+
+        # Visual attention collapse (RVTD detection)
+        if state.visual_attention_mass < 0.1:
+            flags["hallucination_risk"] = "high"
+            flags["action"] = "visual_reground"
+
+        return flags
+
+    def build_trajectory(self, states: List[TokenState]) -> GenerationTrajectory:
+        """Assemble tracked trajectory from individual token states."""
+        total_work = sum(s.jarzynski_work for s in states)
+        # Beta_eff from mean entropy
+        H_mean = sum(s.entropy for s in states) / max(1, len(states))
+        T_mean = self.T_c * (1.0 - H_mean / self.H_max)
+        beta_eff = 1.0 / max(T_mean, 0.01)
+
+        # Check Jarzynski bound
+        bound_violated = self.check_jarzynski_bound(total_work, beta_eff)
+
+        # Reflection triggers
+        triggers = [i for i, s in enumerate(states) if s.temperature_eff > self.T_c]
+
+        # Composite hallucination risk
+        risks = []
+        for s in states:
+            risk = 0.0
+            if s.temperature_eff > self.T_c * 0.8:
+                risk += 0.3
+            if s.retrieval_mu < 0:
+                risk += 0.3
+            if s.visual_attention_mass < 0.2:
+                risk += 0.2
+            if s.E_excitation < s.retrieval_mu * 0.95:
+                risk += 0.2
+            risks.append(min(risk, 1.0))
+        avg_risk = sum(risks) / max(1, len(risks))
+
+        # Grounding score
+        grounding = sum(max(0, s.retrieval_mu) for s in states) / max(1, len(states))
+
+        return GenerationTrajectory(
+            tokens=[0] * len(states),  # Placeholder
+            states=states,
+            total_work=total_work,
+            jarzynski_bound=math.exp(-beta_eff * self.jarzynski_threshold) if bound_violated else 0.0,
+            reflection_triggers=triggers,
+            grounded_score=grounding,
+            hallucination_risk=avg_risk,
+        )
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# Optimal Temperature Calculator (Stochastic Resonance)
+# ═══════════════════════════════════════════════════════════════════════════════
+
+class StochasticResonance:
+    """
+    Compute optimal effective temperature for reasoning tasks.
+    Based on Kramers escape rate: k_escape ∝ exp(-ΔE_barrier / k_B T_eff)
+    """
+
+    @staticmethod
+    def optimal_temperature(
+        barrier_height: float,       # ΔE_barrier — estimated from task complexity
+        target_escape_rate: float,   # Desired exploration rate (0-1)
+        T_c: float = 1.0,
+    ) -> float:
+        """
+        Solve for T_eff such that escape rate matches target.
+        k = exp(-ΔE/T) → T = ΔE / -ln(k)
+        """
+        if target_escape_rate <= 0 or target_escape_rate >= 1:
+            return 0.1 * T_c
+        T_opt = barrier_height / (-math.log(target_escape_rate))
+        # Clamp to BEC stable region: 0.1 T_c < T < 0.95 T_c
+        return max(0.1 * T_c, min(0.95 * T_c, T_opt))
+
+    @staticmethod
+    def barrier_estimate(complexity_score: float) -> float:
+        """
+        Estimate energy barrier from task complexity (0-1 from Sulphur enhancer).
+        Simple tasks: low barrier (easy to escape local minima)
+        Complex reasoning: high barrier (need more thermal energy)
+        """
+        return 0.5 + 2.0 * complexity_score  # Range: 0.5 - 2.5
+
+    @staticmethod
+    def recommend_temperature(complexity: float, T_c: float = 1.0) -> float:
+        """
+        Recommend T_eff for a given task complexity.
+        Low complexity (factual lookup): T ≈ 0.3 T_c (low exploration)
+        High complexity (creative reasoning): T ≈ 0.8 T_c (optimal SR)
+        """
+        barrier = StochasticResonance.barrier_estimate(complexity)
+        # Target escape rate: 0.1 for simple, 0.5 for complex
+        target = 0.1 + 0.4 * complexity
+        return StochasticResonance.optimal_temperature(barrier, target, T_c)