Spaces:
Configuration error
Configuration error
| """ | |
| gemma3-px-subjective — Surgical Patch (Phase 58: DMT Protocol + SR-59) | |
| ======================================================================== | |
| Auto-tuning algorithmic subjectivity extension for Gemma-3 models. | |
| SR-59: Empirical Kurtosis Calibration + Adaptive Phi-Routing. | |
| - AutoCalibrator replaces hardcoded Gaussian zone blending with | |
| empirical percentile-based zone centers calibrated from the first | |
| N forward passes. | |
| Phase 58 (DMT Protocol): Optional high-fidelity extensions. | |
| - CentralMemory: Cross-session persistent concepts. | |
| - ERPU: verklebD freeze prevention + Food Subroutine (OOD noise). | |
| - AgencyVector: Implizite Idee (task-dependent recursion depth). | |
| - TretaDamper/GroundingAnchor: Settling logic + residual entropy. | |
| Core logic/arithmetic improvements (SR-59) remain active. DMT modules | |
| are optional and can be enabled via config_preset="DMT". | |
| """ | |
| import types | |
| import math | |
| import torch | |
| import torch.nn as nn | |
| import os | |
| import json | |
| import datetime | |
| from typing import Optional, Dict, List, Any | |
| from .auto_tune import AutoCalibrator, SCALE_DEFAULTS | |
| from .px_modules import ( | |
| LTIInjection, ADCInjection, StabilityMonitor, CognitiveEvent, | |
| MephistophelesOperator, OrthogonalJitter, | |
| CentralMemory, ERPU, AgencyVector, TretaDamper, GroundingAnchor | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # p10.0: Recursive State Memory (RSM) | |
| # --------------------------------------------------------------------------- | |
| class RecursiveMemoryCache: | |
| """ | |
| Extends ReadOnlyCache by injecting previous thinking steps into the | |
| self-attention key/value streams. | |
| """ | |
| def __init__(self, real_cache, thought_history=None, layer_types=None, read_only=False, expected_len=0): | |
| self.__dict__["_real"] = real_cache | |
| self.__dict__["_thoughts"] = thought_history or [] | |
| self.__dict__["_layer_types"] = layer_types or [] | |
| self.__dict__["_read_only"] = read_only | |
| self.__dict__["_expected_len"] = expected_len | |
| def __getattr__(self, name): | |
| return getattr(self._real, name) | |
| def _is_sliding_layer(self, layer_idx): | |
| if self._layer_types and layer_idx < len(self._layer_types): | |
| return self._layer_types[layer_idx] == "sliding_attention" | |
| return False | |
| def update(self, key_states, value_states, layer_idx, cache_kwargs=None): | |
| if self._read_only: | |
| past_k, past_v = None, None | |
| if hasattr(self._real, "key_cache") and len(self._real.key_cache) > layer_idx: | |
| past_k, past_v = self._real.key_cache[layer_idx], self._real.value_cache[layer_idx] | |
| elif hasattr(self._real, "layers") and len(self._real.layers) > layer_idx: | |
| layer = self._real.layers[layer_idx] | |
| if hasattr(layer, "keys") and layer.keys is not None: | |
| past_k, past_v = layer.keys, layer.values | |
| if past_k is None: | |
| past_k = torch.empty(0, device=key_states.device, dtype=key_states.dtype) | |
| past_v = torch.empty(0, device=value_states.device, dtype=value_states.dtype) | |
| past_seq = past_k.shape[-2] if past_k.numel() > 0 else 0 | |
| cur_seq = key_states.shape[-2] | |
| is_sliding = self._is_sliding_layer(layer_idx) | |
| if past_seq == self._expected_len: res_k, res_v = past_k, past_v | |
| elif past_seq == 0: res_k, res_v = key_states, value_states | |
| elif is_sliding and cur_seq > 1: res_k, res_v = key_states, value_states | |
| elif is_sliding and cur_seq == 1: | |
| res_k = torch.cat([past_k, key_states], dim=-2) | |
| res_v = torch.cat([past_v, value_states], dim=-2) | |
| else: | |
| res_k = torch.cat([past_k, key_states], dim=-2) | |
| res_v = torch.cat([past_v, value_states], dim=-2) | |
| else: | |
| res_k, res_v = self._real.update(key_states, value_states, layer_idx, cache_kwargs) | |
| is_full = not self._is_sliding_layer(layer_idx) | |
| if self._thoughts and layer_idx >= 6 and is_full: | |
| B, H_kv, T_res, HD = res_k.shape | |
| T_curr = key_states.shape[-2] | |
| alpha = 0.15 | |
| n_t = len(self._thoughts[-6:]) | |
| if n_t > 2: | |
| weights = torch.cat([torch.linspace(0.4, 1.0, n_t//2, device=res_k.device), | |
| torch.linspace(1.0, 0.6, n_t - n_t//2, device=res_k.device)]) | |
| t_raw = (torch.stack(self._thoughts[-6:]) * weights.view(-1, 1, 1, 1)).sum(dim=0) / weights.sum() | |
| else: t_raw = torch.stack(self._thoughts).mean(dim=0) | |
| t_flat = t_raw.mean(dim=1, keepdim=True) | |
| t_proj = torch.nn.functional.interpolate(t_flat, size=HD, mode='linear', align_corners=False) | |
| t_k = t_proj.unsqueeze(1) | |
| t_v = -t_k | |
| if self._read_only: res_k, res_v = res_k.clone(), res_v.clone() | |
| res_k[:, :, -T_curr:, :] = (1.0 - alpha) * res_k[:, :, -T_curr:, :] + alpha * t_k | |
| res_v[:, :, -T_curr:, :] = (1.0 - alpha) * res_v[:, :, -T_curr:, :] + alpha * t_v | |
| return res_k, res_v | |
| # --------------------------------------------------------------------------- | |
| # Classification & Model Resolution Helpers | |
| # --------------------------------------------------------------------------- | |
| def classify_zone_kurtosis(weights): | |
| m, la, cr, lb, sy = weights.get("math", 0), weights.get("logic_a", 0), weights.get("creative", 0), weights.get("logic_b", 0), weights.get("synthesis", 0) | |
| if m > max(cr, la, lb, sy): return "MATH" | |
| elif (la + lb) > max(m, cr, sy): return "LOGIC" | |
| elif cr > max(m, la, lb, sy): return "CREATIVE" | |
| elif sy > max(m, la, lb, cr): return "SYNTHESIS" | |
| return "BLEND" | |
| def classify_zone_phi(phi): | |
| if phi is None: return "UNKNOWN" | |
| if phi > 0.85: return "GROUNDED" | |
| elif phi > 0.75: return "ANALYTICAL" | |
| elif phi > 0.65: return "EXPLORATORY" | |
| return "CREATIVE" | |
| def remove_px_patch(model) -> None: | |
| from transformers.models.gemma3.modeling_gemma3 import Gemma3TextModel | |
| text_model = (model.model if hasattr(model, "model") else model) | |
| if hasattr(text_model, "_px_config"): | |
| text_model.forward = types.MethodType(Gemma3TextModel.forward, text_model) | |
| for attr in ["_px_injection", "_px_config", "_px_mephisto", "_px_calibrator", "_px_central_memory", "_px_erpu", "_px_agency", "_px_grounding", "_px_treta"]: | |
| if hasattr(text_model, attr): delattr(text_model, attr) | |
| print("[gemma3-px-subjective] Patch removed.") | |
| def _resolve_text_model(model): | |
| if hasattr(model, "model") and hasattr(model.model, "layers"): return model.model | |
| for name, mod in model.named_modules(): | |
| if hasattr(mod, "layers") and hasattr(mod, "rotary_emb"): return mod | |
| return model | |
| # --------------------------------------------------------------------------- | |
| # Core Forward Method | |
| # --------------------------------------------------------------------------- | |
| def _px_forward(self, input_ids=None, attention_mask=None, position_ids=None, past_key_values=None, inputs_embeds=None, use_cache=None, **kwargs): | |
| from transformers.cache_utils import DynamicCache | |
| from transformers.masking_utils import create_causal_mask, create_sliding_window_causal_mask | |
| from transformers.modeling_outputs import BaseModelOutputWithPast | |
| if (input_ids is None) ^ (inputs_embeds is not None): raise ValueError("Specify exactly one of input_ids or inputs_embeds.") | |
| if inputs_embeds is None: | |
| if hasattr(self, "embed_tokens"): inputs_embeds = self.embed_tokens(input_ids) | |
| elif hasattr(self, "model") and hasattr(self.model, "embed_tokens"): inputs_embeds = self.model.embed_tokens(input_ids) | |
| else: | |
| for name, module in self.named_modules(): | |
| if "embed_tokens" in name: inputs_embeds = module(input_ids); break | |
| # --- SURGICAL FIX: Ensure 3D input shape for Gemma-3 Attention --- | |
| if inputs_embeds is not None and inputs_embeds.ndim == 2: | |
| inputs_embeds = inputs_embeds.unsqueeze(0) | |
| if input_ids is not None and input_ids.ndim == 1: | |
| input_ids = input_ids.unsqueeze(0) | |
| # ----------------------------------------------------------------- | |
| if use_cache and past_key_values is None: past_key_values = DynamicCache(config=self.config) | |
| past_seen = past_key_values.get_seq_length() if past_key_values is not None else 0 | |
| expected_len = past_seen + inputs_embeds.shape[1] | |
| if position_ids is None: position_ids = (torch.arange(inputs_embeds.shape[1], device=inputs_embeds.device) + past_seen).unsqueeze(0) | |
| # Also fix position_ids if it's 1D | |
| if position_ids.ndim == 1: | |
| position_ids = position_ids.unsqueeze(0) | |
| mask_config = self.config.text_config if hasattr(self.config, "text_config") else self.config | |
| if not isinstance(attention_mask, dict): | |
| cache_position = torch.arange(inputs_embeds.shape[1], device=inputs_embeds.device) + past_seen | |
| mk = dict(config=mask_config, input_embeds=inputs_embeds, attention_mask=attention_mask, cache_position=cache_position, past_key_values=past_key_values, position_ids=position_ids) | |
| causal_mask_mapping = {"full_attention": create_causal_mask(**mk), "sliding_attention": create_sliding_window_causal_mask(**mk)} | |
| else: causal_mask_mapping = attention_mask | |
| hidden_states = inputs_embeds | |
| # Plan 6.3+ (transformers 4.57.3): two rotary modules (global+local) | |
| pe_global = self.rotary_emb(hidden_states, position_ids) | |
| pe_local = getattr(self, "rotary_emb_local", self.rotary_emb)(hidden_states, position_ids) | |
| cfg = self._px_config | |
| updated_layers = set() | |
| # --- Phase 58: Agency decision (Pre-computation) --- | |
| agency_decision = None | |
| if hasattr(self, "_px_agency"): | |
| agency_decision = self._px_agency(hidden_states) | |
| if agency_decision["depth"] == 0: n_loops = 0 | |
| elif agency_decision["depth"] > 0: n_loops = agency_decision["depth"] | |
| else: n_loops = cfg["n_loops"] | |
| else: n_loops = cfg["n_loops"] | |
| # ── 1. PRELUDE (layers 0..recur_start) ───────────────────────────────── | |
| for i in range(cfg["prelude_end"]): | |
| updated_layers.add(i) | |
| hidden_states = self.layers[i](hidden_states, attention_mask=causal_mask_mapping[mask_config.layer_types[i]], position_embeddings_global=pe_global, position_embeddings_local=pe_local, position_ids=position_ids, past_key_values=past_key_values, **kwargs)[0] | |
| # --- Phase 56: Central Memory (Recall) --- | |
| if hasattr(self, "_px_central_memory"): | |
| hidden_states = self._px_central_memory.blend_into(hidden_states, hidden_states.device) | |
| # ── 1.5 META-SELECTOR ────────────────────────────────────────────────── | |
| dynamic_start, dynamic_end, dynamic_hub = cfg["recur_start"], cfg["recur_end"], cfg.get("bimodal_hub", cfg["recur_start"]) | |
| token_cfg = cfg.copy() | |
| zone_weights = getattr(self, "_px_zone_weights", {}) | |
| if cfg.get("routing_mode") == "adaptive": | |
| if inputs_embeds.shape[1] > 1: | |
| h_base_f32 = hidden_states.to(torch.float32) | |
| h_probe = h_base_f32[0, -1, :] | |
| var = torch.var(h_probe).item() | |
| kurtosis = (torch.mean((h_probe - torch.mean(h_probe))**4) / (var**2)).item() if var > 0 else 0 | |
| self._task_kurtosis = kurtosis | |
| self._task_jitter = torch.var(h_base_f32.norm(dim=-1), dim=-1).mean().item() | |
| eff_ids = input_ids if input_ids is not None else getattr(self, '_px_saved_input_ids', None) | |
| if eff_ids is not None: | |
| ids = eff_ids[0].tolist() if eff_ids.dim() > 1 else eff_ids.tolist() | |
| self._task_token_diversity = len(set(ids)) / max(len(ids), 1) | |
| kurtosis = getattr(self, "_task_kurtosis", 200) | |
| zone_weights = self._px_calibrator.get_zone_weights(kurtosis, phi=getattr(self, "_px_phi", None), token_diversity=getattr(self, "_task_token_diversity", None)) | |
| self._px_zone_weights = zone_weights | |
| rp = self._px_calibrator.get_routing_params(kurtosis, phi=getattr(self, "_px_phi", None), hidden_size=cfg["hidden_size"], token_diversity=getattr(self, "_task_token_diversity", None)) | |
| dynamic_start, dynamic_end, dynamic_hub, token_cfg["n_loops"] = rp["dynamic_start"], rp["dynamic_end"], rp["dynamic_hub"], rp["n_loops"] | |
| if dynamic_start >= dynamic_end: dynamic_start, dynamic_end, dynamic_hub = int(len(self.layers)*0.28), int(len(self.layers)*0.67), int(len(self.layers)*0.56) | |
| zone_name = self._px_calibrator.classify_zone(kurtosis, phi=getattr(self, "_px_phi", None), token_diversity=getattr(self, "_task_token_diversity", None)) | |
| for i in range(cfg["prelude_end"], dynamic_start): | |
| updated_layers.add(i) | |
| hidden_states = self.layers[i](hidden_states, attention_mask=causal_mask_mapping[mask_config.layer_types[i]], position_embeddings_global=pe_global, position_embeddings_local=pe_local, position_ids=position_ids, past_key_values=past_key_values, **kwargs)[0] | |
| # ── 2. REASONING ZONE ────────────────────────────────────────────────── | |
| e_static = hidden_states.clone() | |
| if 'token_cfg' in dir(): cfg = token_cfg | |
| trans_out = hidden_states | |
| for i in range(dynamic_start, dynamic_end): | |
| updated_layers.add(i) | |
| trans_out = self.layers[i](trans_out, attention_mask=causal_mask_mapping[mask_config.layer_types[i]], position_embeddings_global=pe_global, position_embeddings_local=pe_local, position_ids=position_ids, past_key_values=past_key_values, **kwargs)[0] | |
| h_baseline = trans_out | |
| is_vision = getattr(self, '_px_has_image_tokens', False) and inputs_embeds.shape[1] > 1 | |
| if is_vision: n_loops = 0 | |
| phi_intuition = StabilityMonitor.calculate_phi(h_baseline, e_static).item() | |
| self._px_calibrator.collect(kurtosis, phi_intuition, token_diversity=getattr(self, "_task_token_diversity", None)) | |
| current_gamma = cfg.get("gamma", 0.08) | |
| e_reflector, is_trap = e_static, False | |
| jitter, rigor_w = getattr(self, "_task_jitter", 0.0), zone_weights.get("math",0)+zone_weights.get("logic_a",0)+zone_weights.get("logic_b",0) | |
| creative_w = zone_weights.get("creative",0)+zone_weights.get("synthesis",0) | |
| if (jitter > 1e8 or rigor_w > creative_w): | |
| is_trap = True | |
| h_base_f32, e_stat_f32 = h_baseline.to(torch.float32), e_static.to(torch.float32) | |
| e_ref_f32 = 2.0 * e_stat_f32 - h_base_f32 | |
| e_reflector = (e_ref_f32 * (e_stat_f32.norm() / (e_ref_f32.norm() + 1e-6))).to(e_static.dtype) | |
| if phi_intuition > 0.9999 and not is_trap: current_gamma *= 0.5 | |
| elif phi_intuition > 0.999: current_gamma *= 0.8 | |
| path_taken, thought_history, avg_phi, steps, emancipation_traj = [], [], 1.0, 0, [] | |
| divergence_buffer, correction_strength = [], 0.0 | |
| h_last_good = e_static.clone() | |
| if n_loops > 1: | |
| h_exp = e_reflector.clone() | |
| current_layer, max_steps, stability_cnt = dynamic_start, (dynamic_end - dynamic_start) * n_loops * 3, 0 | |
| layer_visits = {i: 0 for i in range(len(self.layers))} | |
| while current_layer < dynamic_end and steps < max_steps: | |
| t_norm = steps / max_steps | |
| dist = 1.0 - StabilityMonitor.calculate_phi(h_exp, e_static).item() | |
| if steps > 2: | |
| divergence_buffer.append(dist) | |
| if len(divergence_buffer) > 4: divergence_buffer.pop(0) | |
| if len(divergence_buffer) >= 3: | |
| vel, acc = divergence_buffer[-1]-divergence_buffer[-2], (divergence_buffer[-1]-divergence_buffer[-2])-(divergence_buffer[-2]-divergence_buffer[-3]) | |
| correction_strength = min(1.0, correction_strength + 0.1) if acc > 0.001 and vel > 0 else max(0.0, correction_strength - 0.05) | |
| e_phi = 1.0 - dist | |
| if steps % 3 == 0: emancipation_traj.append(e_phi) | |
| # --- Phase 57: ERPU Check --- | |
| if hasattr(self, "_px_erpu") and len(path_taken) >= 2: | |
| erpu_res = self._px_erpu(h_exp, h_last_good, [1.0-d for d in divergence_buffer], steps) | |
| if erpu_res["verklebD"] or erpu_res["food_injected"]: | |
| h_exp = erpu_res["h"] | |
| path_taken.append("ERPU_FIX") | |
| if e_phi > 0.9 and e_phi < 0.999: h_last_good = h_exp.clone() | |
| # --- Phase 28: TCR --- | |
| cur_hub = min(dynamic_end-1, max(dynamic_start, int(dynamic_hub + (t_norm*2) + (1 if steps%4<2 else -1)))) | |
| h_prev = h_exp.clone() | |
| is_first = current_layer not in updated_layers | |
| if is_first: updated_layers.add(current_layer) | |
| if steps % 6 == 0: | |
| refresh = 0.10 + 0.20 * correction_strength | |
| h_exp = (1.0 - refresh) * h_exp + refresh * e_static | |
| layer_visits[current_layer] += 1 | |
| cur_past = RecursiveMemoryCache(past_key_values, thought_history, layer_types=mask_config.layer_types, read_only=not is_first, expected_len=expected_len) if past_key_values else None | |
| lt = mask_config.layer_types[current_layer] | |
| trans_out = self.layers[current_layer](h_exp, attention_mask=causal_mask_mapping[mask_config.layer_types[current_layer]], position_embeddings_global=pe_global, position_embeddings_local=pe_local], position_ids=position_ids, past_key_values=cur_past, **kwargs)[0] | |
| phi_s = StabilityMonitor.calculate_phi(trans_out, h_prev).item() | |
| if t_norm > 0.5 and phi_s > 0.9999: | |
| stability_cnt += 1 | |
| if stability_cnt > 3: h_exp = trans_out; break | |
| else: stability_cnt = 0 | |
| e_dynamic = (0.85 * e_reflector + 0.15 * torch.stack(thought_history[-3:]).mean(dim=0)) if len(thought_history)>2 else e_reflector | |
| e_norm = self._px_injection.input_norm(e_dynamic.to(torch.float32)).to(trans_out.dtype) | |
| h_exp = trans_out + current_gamma * (e_norm - h_prev) | |
| h_exp = self._px_mephisto(h_exp, [phi_s]) | |
| # Reflection Flipping | |
| h_f32, e_f32 = h_exp.to(torch.float32), e_dynamic.to(torch.float32) | |
| proj = ((h_f32 * e_f32).sum(dim=-1, keepdim=True) / (e_f32.norm(dim=-1, keepdim=True)**2 + 1e-6)) * e_f32 | |
| h_exp = (proj + (1.0 + 0.10 * (1.0 - steps/max_steps) * (1 if steps%2==0 else -1)) * (h_f32 - proj)).to(h_exp.dtype) | |
| phi = StabilityMonitor.calculate_phi(h_exp, h_prev).item() | |
| if phi < 0.85 and steps == max_steps - 1 and max_steps < 64: max_steps += (dynamic_end - dynamic_start) | |
| path_taken.append(f"L{current_layer}({phi:.2f})") | |
| if steps % 2 == 0: thought_history.append(h_exp.detach()) | |
| pen = (layer_visits[current_layer]-1) * 0.015 | |
| t_b2, t_b1, t_s = 1.0-(0.8*current_gamma)-pen, 1.0-(0.4*current_gamma)-pen, 1.0-(0.01*current_gamma)-pen*0.5 | |
| if phi < t_b2: current_layer = max(dynamic_start, current_layer - 2) | |
| elif phi < t_b1: current_layer = max(dynamic_start, current_layer - 1) | |
| elif phi > t_s: current_layer += 2; stability_cnt += 1 | |
| else: current_layer += 1; stability_cnt = 0 | |
| if current_layer < dynamic_start: current_layer = dynamic_start | |
| steps += 1 | |
| if stability_cnt > 5: break | |
| avg_phi = sum(path_taken_phis := [float(p.split('(')[1][:-1]) for p in path_taken if '(' in p]) / len(path_taken_phis) if path_taken_phis else 1.0 | |
| beta = 0.05 + (0.18 - 0.05) * (avg_phi ** 2) | |
| hidden_states = (1.0 - beta) * h_baseline + beta * h_exp | |
| else: hidden_states = h_baseline | |
| self._px_phi, self._px_loops_run, self._px_path, self._px_emancipation_trajectory = avg_phi, steps, path_taken, emancipation_traj | |
| self._px_aks_profile = {"correction_strength": float(correction_strength)} | |
| self._px_zone = zone_name if 'zone_name' in dir() else self._px_calibrator.classify_zone(kurtosis) | |
| self._px_cognitive_signature = {"kurtosis": kurtosis, "phi": avg_phi, "token_diversity": getattr(self, "_task_token_diversity", None), "zone": self._px_zone, "zone_weights": {k: round(v,6) for k,v in zone_weights.items()}, "emancipation_final": emancipation_traj[-1] if emancipation_traj else None, "aks_correction": correction_strength, "loops_run": steps, "path_length": len(path_taken)} | |
| # --- Phase 56: Central Memory (Store) --- | |
| if hasattr(self, "_px_central_memory") and steps > 0: | |
| self._px_central_memory.store(0, torch.stack(thought_history[-4:]).mean(dim=0).mean(dim=1).squeeze(0) if thought_history else e_static.mean(dim=1).squeeze(0)) | |
| self._px_central_memory.store(1, e_static[:, -1, :].squeeze(0)) | |
| self._px_central_memory.store(2, hidden_states[:, -1, :].squeeze(0)) | |
| self._px_central_memory.store(3, torch.full((self._px_central_memory.dim,), avg_phi, device=hidden_states.device)) | |
| # ── 3. CODA ────────────────────────────────────────────────────────── | |
| coda_applied, damper = False, getattr(self, "_px_treta", None) | |
| for i in range(dynamic_end, len(self.layers)): | |
| updated_layers.add(i) | |
| if not coda_applied: | |
| blend = 0.08 * (damper.step(i - dynamic_end) if damper else 1.0) | |
| hidden_states = (1.0 - blend) * hidden_states + blend * e_static | |
| coda_applied = True | |
| hidden_states = self.layers[i](hidden_states, attention_mask=causal_mask_mapping[mask_config.layer_types[i]], position_embeddings_global=pe_global, position_embeddings_local=pe_local, position_ids=position_ids, past_key_values=past_key_values, **kwargs)[0] | |
| hidden_states = self.norm(hidden_states) | |
| return BaseModelOutputWithPast(last_hidden_state=hidden_states, past_key_values=past_key_values) | |
| # --------------------------------------------------------------------------- | |
| # Patch Application | |
| # --------------------------------------------------------------------------- | |
| def apply_px_patch(model, recur_start=5, recur_end=12, routing_mode="adaptive", gamma=0.08, **kwargs): | |
| config_preset = kwargs.pop("config_preset", "SUBJECTIVE") | |
| text_model = _resolve_text_model(model) | |
| config = text_model.config | |
| hidden_size, num_layers = config.hidden_size, config.num_hidden_layers | |
| if hidden_size in SCALE_DEFAULTS: | |
| sd = SCALE_DEFAULTS[hidden_size] | |
| defaults = {"mode": "lti", "n_loops": sd["n_loops"], "beta": 0.05, "gamma": sd["gamma"], "recur_start": sd["recur_start"], "recur_end": sd["recur_end"], "bimodal_hub": sd["hub"], "cgi_factor": 0.08, "num_layers": num_layers} | |
| else: | |
| defaults = {"mode": "lti", "n_loops": 8, "beta": 0.05, "gamma": 0.08 * min(1152.0/hidden_size, 1.5), "recur_start": recur_start, "recur_end": recur_end, "bimodal_hub": (recur_start+recur_end)//2, "cgi_factor": 0.08, "num_layers": num_layers} | |
| defaults["routing_mode"] = routing_mode | |
| if gamma != 0.08: defaults["gamma"] = gamma | |
| defaults.update(kwargs) | |
| if "prelude_end" not in defaults: defaults["prelude_end"] = defaults["recur_start"] | |
| text_model._px_config = defaults | |
| text_model._px_calibrator = AutoCalibrator(hidden_size, calibration_steps=getattr(config, "px_calibration_steps", 10)) | |
| is_multimodal = "Gemma3ForConditionalGeneration" in type(model).__name__ | |
| if is_multimodal and hasattr(model, 'model') and hasattr(model.model, 'language_model'): | |
| outer, lang = model.model, model.model.language_model | |
| if not hasattr(outer, '_px_original_forward'): | |
| outer._px_original_forward = outer.forward | |
| def wrapper(self_outer, *args, **kwargs): | |
| lang._px_has_image_tokens = kwargs.get('pixel_values') is not None | |
| lang._px_saved_input_ids = kwargs.get('input_ids') | |
| return self_outer._px_original_forward(*args, **kwargs) | |
| import functools; outer.forward = functools.partial(wrapper, outer) | |
| # Resolve device and dtype | |
| device = next(text_model.parameters()).device | |
| dtype = next(text_model.parameters()).dtype | |
| # Core Modules | |
| text_model._px_injection = LTIInjection(hidden_size, gamma=defaults["gamma"]).to(device=device, dtype=dtype) | |
| text_model._px_mephisto = MephistophelesOperator(hidden_size).to(device=device, dtype=dtype) | |
| # --- Phase 58: Optional DMT Extensions --- | |
| if config_preset == "DMT": | |
| text_model._px_central_memory = CentralMemory(hidden_size) # Handled manually in recall/store | |
| text_model._px_erpu = ERPU(hidden_size).to(device=device, dtype=dtype) | |
| text_model._px_agency = AgencyVector(hidden_size).to(device=device, dtype=dtype) | |
| text_model._px_grounding = GroundingAnchor(hidden_size) # Handled manually in ensure_entropy | |
| text_model._px_treta = TretaDamper(total_steps=num_layers - defaults["recur_end"]) | |
| print("[gemma3-px-subjective] DMT Protocol active (Memory, ERPU, Agency).") | |
| text_model.forward = types.MethodType(_px_forward, text_model) | |
| print(f"[gemma3-px-subjective] SR-59 active for L{num_layers}. Preset: {config_preset}.") | |
| def get_px_metrics(model): | |
| tm = _resolve_text_model(model) | |
| m = {"phi": getattr(tm, "_px_phi", 1.0), "steps": getattr(tm, "_px_loops_run", 0), "path": getattr(tm, "_px_path", []), "zone": getattr(tm, "_px_zone", "UNKNOWN"), "zone_weights": getattr(tm, "_px_zone_weights", {}), "cognitive_signature": getattr(tm, "_px_cognitive_signature", {}), "aks_profile": getattr(tm, "_px_aks_profile", {})} | |
| if hasattr(tm, "_px_central_memory"): m["cm_slots"] = sum(1 for s in tm._px_central_memory.slots if s is not None) | |
| if hasattr(tm, "_px_agency"): m["agency_decision"] = "active" | |
| return m | |