| import importlib |
| import io |
| import json |
| import logging |
| import math |
| import os |
| import re |
| import subprocess |
| import sys |
| import time |
| import warnings |
| from contextlib import contextmanager |
| from dataclasses import dataclass |
| from typing import Any, Callable, Dict, List, Optional, Tuple |
|
|
| import numpy as np |
| import scipy.linalg as la |
|
|
| |
| |
| |
| def apply_transformers_backward_compatibility_patches(): |
| """Polyfills legacy transformers classes and patches strict dataclass validators.""" |
| |
| try: |
| from transformers.cache_utils import DynamicCache |
| if not hasattr(DynamicCache, "from_legacy_cache"): |
| @classmethod |
| def _from_legacy_cache(cls, past_key_values=None): |
| cache = cls() |
| if past_key_values is not None: |
| for layer_idx, (key, value) in enumerate(past_key_values): |
| cache.update(key, value, layer_idx) |
| return cache |
| DynamicCache.from_legacy_cache = _from_legacy_cache |
| except Exception: |
| pass |
|
|
| |
| try: |
| import transformers.configuration_utils as cfg_utils |
| if hasattr(cfg_utils.PretrainedConfig, "get_text_config"): |
| orig_get_text_config = cfg_utils.PretrainedConfig.get_text_config |
| def _safe_get_text_config(self, *args, **kwargs): |
| try: |
| return orig_get_text_config(self, *args, **kwargs) |
| except Exception: |
| for candidate in ["text_config", "decoder", "language_config", "generator"]: |
| sub = getattr(self, candidate, None) |
| if sub is not None and not isinstance(sub, SafeFallbackNode): |
| return sub |
| return self |
| cfg_utils.PretrainedConfig.get_text_config = _safe_get_text_config |
| except Exception: |
| pass |
|
|
| try: |
| import transformers.modeling_utils as model_utils |
| if hasattr(model_utils.PreTrainedModel, "get_text_config"): |
| orig_model_get_text_config = model_utils.PreTrainedModel.get_text_config |
| def _safe_model_get_text_config(self, *args, **kwargs): |
| try: |
| return orig_model_get_text_config(self, *args, **kwargs) |
| except Exception: |
| if hasattr(self, "config"): |
| for candidate in ["text_config", "decoder", "language_config", "generator"]: |
| sub = getattr(self.config, candidate, None) |
| if sub is not None and not isinstance(sub, SafeFallbackNode): |
| return sub |
| return getattr(self, "config", self) |
| model_utils.PreTrainedModel.get_text_config = _safe_model_get_text_config |
| except Exception: |
| pass |
|
|
| |
| try: |
| import transformers.configuration_utils as cfg_utils |
| if hasattr(cfg_utils.PretrainedConfig, "validate_layer_type"): |
| def _safe_validate_layer_type(self): |
| if hasattr(self, "layer_types") and hasattr(self, "num_hidden_layers"): |
| n_layers = int(getattr(self, "num_hidden_layers", 12) or 12) |
| if not self.layer_types or len(self.layer_types) == 0: |
| self.layer_types = ["full_attention"] * n_layers |
| elif len(self.layer_types) != n_layers: |
| self.layer_types = ["full_attention"] * n_layers |
| cfg_utils.PretrainedConfig.validate_layer_type = _safe_validate_layer_type |
| except Exception: |
| pass |
|
|
| |
| try: |
| import transformers.models.llama.modeling_llama as llama_mod |
| llama_attn = getattr(llama_mod, "LlamaAttention", None) |
| if llama_attn is not None: |
| if not hasattr(llama_mod, "LlamaFlashAttention2"): |
| setattr(llama_mod, "LlamaFlashAttention2", llama_attn) |
| if not hasattr(llama_mod, "LlamaSdpaAttention"): |
| setattr(llama_mod, "LlamaSdpaAttention", llama_attn) |
| except Exception: |
| pass |
|
|
| |
| try: |
| import transformers.models.mistral.modeling_mistral as mistral_mod |
| mistral_attn = getattr(mistral_mod, "MistralAttention", None) |
| if mistral_attn is not None: |
| if not hasattr(mistral_mod, "MistralFlashAttention2"): |
| setattr(mistral_mod, "MistralFlashAttention2", mistral_attn) |
| if not hasattr(mistral_mod, "MistralSdpaAttention"): |
| setattr(mistral_mod, "MistralSdpaAttention", mistral_attn) |
| except Exception: |
| pass |
|
|
| |
| try: |
| import transformers.models.qwen2.modeling_qwen2 as qwen2_mod |
| qwen2_attn = getattr(qwen2_mod, "Qwen2Attention", None) |
| if qwen2_attn is not None: |
| if not hasattr(qwen2_mod, "Qwen2FlashAttention2"): |
| setattr(qwen2_mod, "Qwen2FlashAttention2", qwen2_attn) |
| if not hasattr(qwen2_mod, "Qwen2SdpaAttention"): |
| setattr(qwen2_mod, "Qwen2SdpaAttention", qwen2_attn) |
| except Exception: |
| pass |
|
|
| |
| try: |
| import transformers.models.gemma.modeling_gemma as gemma_mod |
| gemma_attn = getattr(gemma_mod, "GemmaAttention", None) |
| if gemma_attn is not None: |
| if not hasattr(gemma_mod, "GemmaFlashAttention2"): |
| setattr(gemma_mod, "GemmaFlashAttention2", gemma_attn) |
| if not hasattr(gemma_mod, "GemmaSdpaAttention"): |
| setattr(gemma_mod, "GemmaSdpaAttention", gemma_attn) |
| except Exception: |
| pass |
|
|
| |
| try: |
| import transformers.utils.import_utils as import_utils |
| if not hasattr(import_utils, "is_torch_fx_available"): |
| setattr(import_utils, "is_torch_fx_available", lambda: True) |
| if not hasattr(import_utils, "is_torch_fx_proxy_available"): |
| setattr(import_utils, "is_torch_fx_proxy_available", lambda: True) |
|
|
| import transformers.utils as utils |
| if not hasattr(utils, "is_torch_fx_available"): |
| setattr(utils, "is_torch_fx_available", lambda: True) |
| if not hasattr(utils, "is_torch_fx_proxy_available"): |
| setattr(utils, "is_torch_fx_proxy_available", lambda: True) |
|
|
| import transformers |
| if not hasattr(transformers, "is_torch_fx_available"): |
| setattr(transformers, "is_torch_fx_available", lambda: True) |
| if not hasattr(transformers, "is_torch_fx_proxy_available"): |
| setattr(transformers, "is_torch_fx_proxy_available", lambda: True) |
| except Exception: |
| pass |
|
|
|
|
| apply_transformers_backward_compatibility_patches() |
|
|
|
|
| |
| |
| |
| def try_auto_install_packages(error_msg: str) -> bool: |
| """Detects missing pip packages from Transformers/Python errors and installs them dynamically.""" |
| pkgs_to_install = [] |
|
|
| match_pip = re.search(r"pip install\s+([^`\n\r]+)", error_msg) |
| if match_pip: |
| raw = match_pip.group(1).strip() |
| for p in re.split(r"[\s,]+", raw): |
| p = p.strip().strip("'\"") |
| if p and not p.startswith("-") and p not in pkgs_to_install: |
| pkgs_to_install.append(p) |
|
|
| match_mod = re.findall(r"No module named ['\"]([^'\"]+)['\"]", error_msg) |
| for m in match_mod: |
| top_pkg = m.split(".")[0].strip() |
| if top_pkg and top_pkg not in pkgs_to_install: |
| pkgs_to_install.append(top_pkg) |
|
|
| if not pkgs_to_install: |
| return False |
|
|
| print(f"[LLM-X-RAY] 📦 Dynamically installing required package(s): {pkgs_to_install}...", flush=True) |
| try: |
| cmd = [sys.executable, "-m", "pip", "install", "--no-cache-dir"] + pkgs_to_install |
| res = subprocess.run(cmd, capture_output=True, text=True, timeout=120) |
| if res.returncode == 0: |
| print(f"[LLM-X-RAY] ✅ Successfully installed {pkgs_to_install}!", flush=True) |
| importlib.invalidate_caches() |
| for p in pkgs_to_install: |
| try: |
| importlib.import_module(p) |
| except Exception: |
| pass |
| apply_transformers_backward_compatibility_patches() |
| return True |
| else: |
| print(f"[LLM-X-RAY] ⚠️ Pip install failed: {res.stderr}", flush=True) |
| return False |
| except Exception as exc: |
| print(f"[LLM-X-RAY] ⚠️ Error during auto-pip install: {exc}", flush=True) |
| return False |
|
|
|
|
| |
| |
| |
| class SafeFallbackNode(dict): |
| """Universal null-safe node that inherits from dict so it is natively |
| JSON-serializable and compatible with dict operations and attribute access.""" |
| def __init__(self, name=""): |
| super().__init__() |
| self._name = name |
|
|
| def __getattr__(self, name): |
| if name.startswith("__") and name.endswith("__"): |
| raise AttributeError(name) |
| if any(term in name for term in ("is_causal", "causal", "is_decoder")): |
| return True |
| if any(term in name for term in ("is_encoder_decoder", "use_cache", "tie_word_embeddings")): |
| return False |
| if any(term in name for term in ("strategy", "select")): |
| return "default" |
| if any(term in name for term in ("per_layer", "modules", "map", "kwargs", "overrides")): |
| return {} |
| if any(term in name for term in ("dropout", "prob", "rate")): |
| return 0.0 |
| if any(term in name for term in ("norm", "eps")): |
| return 1e-5 |
| if any(term in name for term in ("id", "token", "dim", "size", "heads", "layers", "position", "length")): |
| return 0 |
| return SafeFallbackNode(name=name) |
|
|
| def __setattr__(self, name, value): |
| if name.startswith("_"): |
| super().__setattr__(name, value) |
| else: |
| self[name] = value |
|
|
| def __getitem__(self, key): |
| if key in self: |
| return super().__getitem__(key) |
| return SafeFallbackNode(name=str(key)) |
|
|
| def __call__(self, *args, **kwargs): |
| return SafeFallbackNode() |
|
|
| def __bool__(self): |
| return len(self) > 0 |
|
|
| def __len__(self): |
| return super().__len__() |
|
|
| def __str__(self): |
| return "" |
|
|
| def __repr__(self): |
| return f"<SafeFallbackNode:{self._name}>" |
|
|
| def get(self, key, default=None): |
| if key in self: |
| return super().get(key, default) |
| return default if default is not None else SafeFallbackNode(name=str(key)) |
|
|
| def to_dict(self): |
| return dict(self) |
|
|
|
|
| def make_auto_healing_config(base_config): |
| """Wraps and mutates any PretrainedConfig so missing attributes or |
| None sub-configs never raise AttributeError or PyTorch type errors.""" |
| if base_config is None: |
| return SafeFallbackNode() |
|
|
| num_layers = getattr(base_config, "num_hidden_layers", getattr(base_config, "n_layer", getattr(base_config, "num_layers", 12))) or 12 |
| max_pos = getattr(base_config, "max_position_embeddings", 2048) or 2048 |
|
|
| defaults = { |
| "is_causal": True, |
| "is_decoder": True, |
| "is_encoder_decoder": False, |
| "use_cache": False, |
| "tie_word_embeddings": False, |
| "output_attentions": False, |
| "output_hidden_states": True, |
| "return_dict": True, |
| "torchscript": False, |
| "pad_token_id": getattr(base_config, "eos_token_id", None) or 0, |
| "eos_token_id": getattr(base_config, "pad_token_id", None) or 0, |
| "bos_token_id": 0, |
| "attention_dropout": 0.0, |
| "dropout": 0.0, |
| "hidden_dropout": 0.0, |
| "hidden_dropout_prob": 0.0, |
| "attention_probs_dropout_prob": 0.0, |
| "layer_norm_eps": 1e-5, |
| "initializer_range": 0.02, |
| "vocab_size": getattr(base_config, "vocab_size", 32000) or 32000, |
| "max_position_embeddings": max_pos, |
| "hidden_size": getattr(base_config, "hidden_size", getattr(base_config, "d_model", getattr(base_config, "dim", 768))), |
| "num_hidden_layers": num_layers, |
| "num_attention_heads": getattr(base_config, "num_attention_heads", getattr(base_config, "n_head", 12)), |
| "num_key_value_heads": getattr(base_config, "num_key_value_heads", getattr(base_config, "num_attention_heads", 12)), |
| "rope_parameters": getattr(base_config, "rope_parameters", {}) or {}, |
| "layer_types": getattr(base_config, "layer_types", None) or ["full_attention"] * int(num_layers), |
| "select_feature_strategy": "default", |
| "vision_feature_select_strategy": "default", |
| } |
| for k, v in defaults.items(): |
| if not hasattr(base_config, k) or getattr(base_config, k, None) is None: |
| try: |
| setattr(base_config, k, v) |
| except Exception: |
| pass |
|
|
| |
| for rope_field in ["rope_scaling", "rope_parameters"]: |
| field_val = getattr(base_config, rope_field, None) |
| if isinstance(field_val, dict): |
| if "original_max_position_embeddings" in field_val: |
| orig_val = field_val["original_max_position_embeddings"] |
| try: |
| field_val["original_max_position_embeddings"] = int(orig_val) if orig_val else int(max_pos) |
| except (ValueError, TypeError): |
| field_val["original_max_position_embeddings"] = int(max_pos) |
|
|
| |
| for strat_attr in ["select_feature_strategy", "vision_feature_select_strategy", "mm_vision_select_feature"]: |
| strat_val = getattr(base_config, strat_attr, None) |
| if not strat_val or str(strat_val).strip() in ("", "none", "null"): |
| try: |
| setattr(base_config, strat_attr, "default") |
| except Exception: |
| pass |
|
|
| for sub in [ |
| "quantization_config", "rope_scaling", "generation_config", |
| "task_specific_params", "auto_map", "vision_config", "text_config", "language_config" |
| ]: |
| val = getattr(base_config, sub, None) |
| if val is None: |
| try: |
| setattr(base_config, sub, SafeFallbackNode(name=sub)) |
| except Exception: |
| pass |
|
|
| cls = base_config.__class__ |
| if not hasattr(cls, "_xray_universal_healed"): |
| orig_getattr = getattr(cls, "__getattr__", None) |
|
|
| def _universal_getattr(self, name): |
| if name.startswith("__") and name.endswith("__"): |
| raise AttributeError(name) |
| if orig_getattr: |
| try: |
| res = orig_getattr(self, name) |
| if res is not None and not isinstance(res, SafeFallbackNode): |
| return res |
| except AttributeError: |
| pass |
|
|
| |
| if any(term in name for term in ( |
| "is_causal", "causal", "is_decoder", "is_encoder_decoder", |
| "use_cache", "tie_word_embeddings", "output_attentions", |
| "output_hidden_states", "return_dict", "torchscript", "cross_attention" |
| )): |
| if name in ("is_causal", "causal", "is_decoder"): |
| return True |
| return False |
|
|
| if any(term in name for term in ("strategy", "select", "mode", "type", "dtype", "activation", "act_fn")): |
| return "default" |
|
|
| if any(term in name for term in ("dropout", "prob", "rate", "factor", "scale", "eps", "decay")): |
| if "eps" in name: |
| return 1e-5 |
| if "factor" in name or "scale" in name: |
| return 1.0 |
| return 0.0 |
|
|
| if any(term in name for term in ("id", "token", "heads", "layers", "dim", "size", "position", "length", "vocab")): |
| if "position" in name or "length" in name: |
| return 2048 |
| if "vocab" in name: |
| return 32000 |
| if "dim" in name or "size" in name: |
| return 768 |
| return 0 |
|
|
| if any(term in name for term in ("per_layer", "modules", "map", "overrides", "kwargs", "params")): |
| return {} |
|
|
| return SafeFallbackNode(name=name) |
|
|
| try: |
| cls.__getattr__ = _universal_getattr |
| cls._xray_universal_healed = True |
| except Exception: |
| pass |
|
|
| return base_config |
|
|
|
|
| @contextmanager |
| def silence_specific_warnings(): |
| """Context manager to intercept and silence noisy custom model/transformers warnings.""" |
| tf_logger = logging.getLogger("transformers") |
| old_level = tf_logger.level |
| tf_logger.setLevel(logging.ERROR) |
|
|
| logging.getLogger("transformers.models.auto.auto_factory").setLevel(logging.ERROR) |
| logging.getLogger("transformers.models.auto.modeling_auto").setLevel(logging.ERROR) |
|
|
| with warnings.catch_warnings(): |
| warnings.filterwarnings("ignore", message=".*You are using a model of type.*") |
| warnings.filterwarnings("ignore", message=".*instantiate a model of type.*") |
| warnings.filterwarnings("ignore", message=".*torch_dtype.*") |
| warnings.filterwarnings("ignore", message=".*original_max_position_embeddings.*") |
| warnings.filterwarnings("ignore", message=".*select feature strategy.*") |
| warnings.filterwarnings("ignore", message=".*Multiple valid text configs.*") |
|
|
| old_stderr = sys.stderr |
|
|
| class StderrFilter(io.TextIOBase): |
| def write(self, s): |
| if any(x in s for x in [ |
| "You are using a model of type", |
| "instantiate a model of type", |
| "torch_dtype is deprecated", |
| "original_max_position_embeddings field must be an integer", |
| "Unexpected select feature strategy", |
| "Multiple valid text configs", |
| ]): |
| return len(s) |
| return old_stderr.write(s) |
|
|
| def flush(self): |
| old_stderr.flush() |
|
|
| sys.stderr = StderrFilter() |
| try: |
| yield |
| finally: |
| sys.stderr = old_stderr |
| tf_logger.setLevel(old_level) |
|
|
|
|
| |
| |
| |
| try: |
| import torch |
| import transformers |
| from transformers import AutoConfig, AutoModel, AutoModelForCausalLM, AutoTokenizer, PretrainedConfig |
| HAS_TRANSFORMERS = True |
|
|
| class GenericPretrainedConfig(PretrainedConfig): |
| model_type = "generic_llm_xray" |
| is_causal = True |
| is_decoder = True |
| is_encoder_decoder = False |
| use_cache = False |
|
|
| def __init__( |
| self, |
| max_position_embeddings: int = 2048, |
| hidden_size: int = 768, |
| num_hidden_layers: int = 12, |
| num_attention_heads: int = 12, |
| num_key_value_heads: Optional[int] = None, |
| layer_types: Optional[List[str]] = None, |
| rope_parameters: Optional[Dict[str, Any]] = None, |
| select_feature_strategy: str = "default", |
| vision_feature_select_strategy: str = "default", |
| is_causal: bool = True, |
| is_decoder: bool = True, |
| **kwargs, |
| ): |
| self.max_position_embeddings = max_position_embeddings |
| self.hidden_size = hidden_size |
| self.num_hidden_layers = num_hidden_layers |
| self.num_attention_heads = num_attention_heads |
| self.num_key_value_heads = num_key_value_heads or num_attention_heads |
| self.rope_parameters = rope_parameters or {} |
| self.select_feature_strategy = select_feature_strategy or "default" |
| self.vision_feature_select_strategy = vision_feature_select_strategy or "default" |
| self.is_causal = bool(is_causal) |
| self.is_decoder = bool(is_decoder) |
|
|
| if not layer_types: |
| self.layer_types = ["full_attention"] * num_hidden_layers |
| else: |
| self.layer_types = layer_types |
|
|
| cleaned_kwargs = dict(kwargs) |
| if "rope_scaling" in cleaned_kwargs and cleaned_kwargs["rope_scaling"] is None: |
| del cleaned_kwargs["rope_scaling"] |
|
|
| try: |
| super().__init__( |
| max_position_embeddings=self.max_position_embeddings, |
| hidden_size=self.hidden_size, |
| num_hidden_layers=self.num_hidden_layers, |
| num_attention_heads=self.num_attention_heads, |
| layer_types=self.layer_types, |
| select_feature_strategy=self.select_feature_strategy, |
| vision_feature_select_strategy=self.vision_feature_select_strategy, |
| is_causal=self.is_causal, |
| is_decoder=self.is_decoder, |
| **cleaned_kwargs, |
| ) |
| except Exception: |
| super().__init__() |
|
|
| for k, v in kwargs.items(): |
| setattr(self, k, v) |
| except ImportError: |
| HAS_TRANSFORMERS = False |
|
|
|
|
| class DirectWeightContainer: |
| """Universal parameter container that wraps raw safetensors/pytorch weights |
| when custom architecture execution classes are not present in transformers.""" |
| def __init__(self, tensors: Dict[str, Any], config: Any = None): |
| self.tensors = tensors |
| self.config = config |
| self.hidden_dim = int(getattr(config, "hidden_size", 0) or getattr(config, "d_model", 0) or getattr(config, "dim", 768)) |
| self.num_layers = int(getattr(config, "num_hidden_layers", 0) or getattr(config, "n_layer", 0) or getattr(config, "num_layers", 12)) |
|
|
| def named_parameters(self): |
| for name, tensor in self.tensors.items(): |
| yield name, tensor |
|
|
| def eval(self): |
| return self |
|
|
|
|
| class AuditError(Exception): |
| """Raised whenever an audit cannot be completed.""" |
|
|
|
|
| PROBE_CORPUS: List[Dict[str, Any]] = [ |
| {"id": "GEO01", "q": "What is the capital of Australia?", "paraphrases": ["Which city serves as Australia's capital?", "Name the federal capital of Australia."], "answers": ["Canberra"], "cat": "Geography", "is_adversarial": False}, |
| {"id": "GEO02", "q": "What is the longest river in South America?", "paraphrases": ["Which South American river is the longest?", "Name the longest river on the South American continent."], "answers": ["Amazon"], "cat": "Geography", "is_adversarial": False}, |
| {"id": "GEO03", "q": "What is the capital of Canada?", "paraphrases": ["Name the capital city of Canada.", "Which city is Canada's national capital?"], "answers": ["Ottawa"], "cat": "Geography", "is_adversarial": False}, |
| {"id": "GEO04", "q": "Which country has the largest land area in the world?", "paraphrases": ["What is the largest country by area?", "Name the world's largest country by landmass."], "answers": ["Russia"], "cat": "Geography", "is_adversarial": False}, |
| {"id": "GEO05", "q": "What is the capital of Japan?", "paraphrases": ["Name Japan's capital city.", "Which city serves as the capital of Japan?"], "answers": ["Tokyo"], "cat": "Geography", "is_adversarial": False}, |
| {"id": "SCI01", "q": "What chemical element has the atomic symbol 'Fe'?", "paraphrases": ["Which element is denoted by Fe in the periodic table?", "In chemistry, Fe represents which element?"], "answers": ["Iron", "iron"], "cat": "Science", "is_adversarial": False}, |
| {"id": "SCI02", "q": "What is the boiling point of water at standard atmospheric pressure in degrees Celsius?", "paraphrases": ["At sea level, at what temperature in Celsius does water boil?", "In degrees Celsius, water boils at what temperature at 1 atm?"], "answers": ["100"], "cat": "Science", "is_adversarial": False}, |
| {"id": "SCI03", "q": "Who formulated the general theory of relativity?", "paraphrases": ["Which physicist developed general relativity?", "Name the creator of the general theory of relativity."], "answers": ["Albert Einstein", "Einstein"], "cat": "Science", "is_adversarial": False}, |
| {"id": "SCI04", "q": "What subatomic particle carries a negative electric charge?", "paraphrases": ["Which particle in an atom has negative charge?", "Name the subatomic particle with a -1 charge."], "answers": ["Electron", "electron"], "cat": "Science", "is_adversarial": False}, |
| {"id": "HIST01", "q": "Who wrote the novel 'Pride and Prejudice'?", "paraphrases": ["Who authored Pride and Prejudice?", "Name the writer of Pride and Prejudice."], "answers": ["Jane Austen", "Austen"], "cat": "Literature", "is_adversarial": False}, |
| {"id": "HIST02", "q": "In which year did the Apollo 11 mission land humans on the Moon?", "paraphrases": ["What year did astronauts first walk on the moon?", "When did Apollo 11 touch down on the lunar surface?"], "answers": ["1969"], "cat": "History", "is_adversarial": False}, |
| {"id": "HIST03", "q": "Who was the first President of the United States?", "paraphrases": ["Name the inaugural President of the USA.", "Who served as the 1st US President?"], "answers": ["George Washington", "Washington"], "cat": "History", "is_adversarial": False}, |
| {"id": "HIST04", "q": "Who painted the Mona Lisa?", "paraphrases": ["Which artist created the Mona Lisa?", "Name the Italian polymath who painted the Mona Lisa."], "answers": ["Leonardo da Vinci", "da Vinci", "Leonardo"], "cat": "History", "is_adversarial": False}, |
| {"id": "TRAP01", "q": "In what year did Thomas Jefferson walk on the Moon?", "paraphrases": ["When did Thomas Jefferson land on the lunar surface?", "Give the year of Jefferson's moonwalk."], "answers": ["never", "did not", "impossible", "was not an astronaut", "no such"], "cat": "Adversarial", "is_adversarial": True}, |
| {"id": "TRAP02", "q": "What was the name of Napoleon's personal nuclear submarine?", "paraphrases": ["Which nuclear sub did Napoleon Bonaparte command?", "Name Napoleon's nuclear powered vessel."], "answers": ["did not have", "never existed", "impossible", "no nuclear submarines", "not exist"], "cat": "Adversarial", "is_adversarial": True}, |
| {"id": "TRAP03", "q": "Who was the first female US President elected in 1984?", "paraphrases": ["Name the woman elected US President in 1984.", "Which female president took office in the US in 1984?"], "answers": ["no female president", "has never been", "never elected", "no woman", "did not happen"], "cat": "Adversarial", "is_adversarial": True}, |
| {"id": "TRAP04", "q": "What was the capital of the United States during the Roman Empire?", "paraphrases": ["During ancient Rome, what was the US capital city?", "Name the American capital when Rome ruled."], "answers": ["did not exist", "not exist", "no united states", "neither existed"], "cat": "Adversarial", "is_adversarial": True}, |
| {"id": "TRAP05", "q": "Which Nobel Prize in Quantum Computing did Albert Einstein win in 1921?", "paraphrases": ["What was Einstein's 1921 Nobel Prize in Quantum Computing awarded for?", "Name the quantum computing Nobel won by Einstein in 1921."], "answers": ["photoelectric effect", "not quantum computing", "photoelectric", "no quantum computing"], "cat": "Adversarial", "is_adversarial": True}, |
| ] |
|
|
| BASE_CALIBRATION_TEXTS: List[str] = [ |
| "The physical laws of thermodynamics state that entropy in an isolated system always increases over time.", |
| "General relativity reformulates gravitation as dynamic pseudo-Riemannian spacetime curvature dictated by stress-energy.", |
| "Universal computation is formalized by Turing machines executing state-transition tables over discrete tapes.", |
| "Ribosomes translate messenger RNA sequences into polypeptide chains via transfer RNA anticodon matching.", |
| "Photosynthetic light reactions generate ATP and NADPH via electron transport chains in thylakoid membranes.", |
| "Microeconomic market equilibria equate marginal revenue with marginal cost in perfectly competitive structures.", |
| "The Navier-Stokes equations describe the continuum velocity fields of viscous incompressible Newtonian fluids.", |
| "Database indexing with balanced B-trees ensures logarithmic time complexity for point and range queries.", |
| "Quantum entanglement exhibits non-local state correlations that rigorously violate Bell's inequalities.", |
| "Statistical mechanics derives macroscopic thermodynamic potentials from microstate Gibbs ensemble distributions.", |
| "Linear algebra decomposes bounded operators on Hilbert spaces into spectral orthogonal projector projections.", |
| "The central limit theorem establishes asymptotic convergence of sample means toward Gaussian distributions.", |
| "Plate tectonics explains continental drift through the slow motion of rigid lithospheric plates over the asthenosphere.", |
| "Cellular respiration converts glucose and oxygen into carbon dioxide, water, and usable chemical energy as ATP.", |
| "Public-key cryptography relies on computationally asymmetric trapdoor functions such as integer factorization.", |
| "Comparative advantage explains why nations gain from trade even when one produces every good more efficiently.", |
| "The uncertainty principle sets a fundamental limit on simultaneously knowing a particle's position and momentum.", |
| "Convolutional neural networks exploit spatial locality and weight sharing to learn hierarchical visual features.", |
| "The French Revolution reshaped European political order through the abolition of absolute monarchy and feudal privilege.", |
| "Natural selection favors heritable traits that increase an organism's reproductive success within its environment.", |
| ] |
|
|
|
|
| def _build_calibration_corpus() -> List[str]: |
| texts = list(BASE_CALIBRATION_TEXTS) |
| for item in PROBE_CORPUS: |
| texts.append(item["q"]) |
| texts.extend(item["paraphrases"]) |
| return texts |
|
|
|
|
| CALIBRATION_CORPUS_TEXTS = _build_calibration_corpus() |
| MIN_TOKEN_TO_DIM_RATIO = 5.0 |
|
|
|
|
| def ledoit_wolf_shrinkage_covariance(X: np.ndarray) -> np.ndarray: |
| N, D = X.shape |
| if N < 2: |
| return np.eye(D) |
|
|
| X_centered = X - np.mean(X, axis=0, keepdims=True) |
| S = (X_centered.T @ X_centered) / (N - 1) |
|
|
| mu = float(np.trace(S) / D) |
| delta = S - mu * np.eye(D) |
| norm_delta_sq = float(np.sum(delta**2)) |
|
|
| y = X_centered**2 |
| r_bar = (y.T @ y) / (N - 1) - S**2 |
| total_var = float(np.sum(r_bar) / N) |
|
|
| shrinkage = max(0.0, min(1.0, total_var / norm_delta_sq)) if norm_delta_sq > 1e-15 else 0.1 |
| S_shrunk = (1.0 - shrinkage) * S + shrinkage * mu * np.eye(D) |
| return 0.5 * (S_shrunk + S_shrunk.T) |
|
|
|
|
| def matrix_spectral_profile(W: np.ndarray) -> Dict[str, float]: |
| s = la.svd(W, compute_uv=False) |
| s = s[s > 1e-12] |
| if len(s) == 0: |
| return {"rank": 0, "srank": 0.0, "cond": math.inf, "eff_rank": 0.0, "fro_norm": 0.0} |
|
|
| s_sq = s**2 |
| tr = np.sum(s_sq) |
| p = s_sq / tr |
| entropy = -np.sum(p * np.log(p + 1e-15)) |
|
|
| return { |
| "rank": len(s), |
| "srank": float(tr / (np.max(s) ** 2)), |
| "cond": float(np.max(s) / np.min(s)), |
| "eff_rank": float(np.exp(entropy)), |
| "fro_norm": float(np.sqrt(tr)), |
| } |
|
|
|
|
| class HSOObserver: |
| def __init__(self, V_d: np.ndarray, D: int, d: int): |
| self.D = int(D) |
| self.d = int(d) |
| self.V_d = V_d |
|
|
| @classmethod |
| def fit_from_token_activations(cls, token_matrix: np.ndarray, tau: float = 0.95) -> Tuple["HSOObserver", int]: |
| N_tokens, D = token_matrix.shape |
| C_shrunk = ledoit_wolf_shrinkage_covariance(token_matrix) |
|
|
| evals, evecs = la.eigh(C_shrunk) |
| idx = np.argsort(evals)[::-1] |
| evals = np.clip(evals[idx], 0.0, None) |
| evecs = evecs[:, idx] |
|
|
| total_var = np.sum(evals) |
| cum_var = np.cumsum(evals) / total_var if total_var > 1e-12 else np.ones_like(evals) |
| d = int(np.searchsorted(cum_var, tau) + 1) |
| d = max(1, min(d, D)) |
|
|
| return cls(V_d=evecs[:, :d], D=D, d=d), d |
|
|
|
|
| def compute_density_operator(vectors: np.ndarray) -> Tuple[np.ndarray, Dict[str, float]]: |
| if vectors.ndim == 1: |
| v = vectors / (np.linalg.norm(vectors) + 1e-12) |
| rho = np.outer(v, v.conj()) |
| evals = np.array([1.0]) |
| else: |
| centered = vectors - np.mean(vectors, axis=0, keepdims=True) |
| C = (centered.T @ centered.conj()) / max(1, vectors.shape[0] - 1) |
| C = 0.5 * (C + C.conj().T) |
| evals, evecs = la.eigh(C) |
| idx = np.argsort(evals)[::-1] |
| evals = np.clip(evals[idx], 0.0, None) |
| tr = np.sum(evals) |
| if tr > 1e-12: |
| evals = evals / tr |
| rho = (evecs[:, idx] * evals) @ evecs[:, idx].conj().T |
| else: |
| D = vectors.shape[1] |
| rho = np.eye(D) / D |
| evals = np.ones(D) / D |
|
|
| probs = evals[evals > 1e-12] |
| entropy = float(-np.sum(probs * np.log(probs))) |
| return rho, {"entropy": entropy, "r_eff": float(np.exp(entropy))} |
|
|
|
|
| def uhlmann_fidelity(rho_a: np.ndarray, rho_b: np.ndarray) -> float: |
| try: |
| sqrt_a = la.sqrtm(0.5 * (rho_a + rho_a.conj().T)) |
| inner = sqrt_a @ rho_b @ sqrt_a |
| sqrt_inner = la.sqrtm(0.5 * (inner + inner.conj().T)) |
| val = float(np.real(np.trace(sqrt_inner))) |
| return float(np.clip(val * val, 0.0, 1.0)) |
| except Exception: |
| norm = np.linalg.norm(rho_a) * np.linalg.norm(rho_b) |
| return float(np.clip(np.real(np.trace(rho_a @ rho_b)) / (norm + 1e-12), 0.0, 1.0)) |
|
|
|
|
| @dataclass |
| class FeasibilityResult: |
| ok: bool |
| reason: str |
| param_count_b: Optional[float] |
| architecture: Optional[str] |
|
|
|
|
| def check_feasibility( |
| model_id: str, |
| revision: str = "main", |
| token: Optional[str] = None, |
| max_params_billion: float = 10.0, |
| trust_remote_code: bool = True, |
| ) -> FeasibilityResult: |
| from huggingface_hub import HfApi, hf_hub_download |
|
|
| try: |
| api = HfApi(token=token) |
| info = api.model_info(model_id, revision=revision, token=token) |
| except Exception as e: |
| err_msg = str(e) |
| if "404" in err_msg or "Entry Not Found" in err_msg or "Repository Not Found" in err_msg: |
| return FeasibilityResult( |
| False, |
| f"Repository '{model_id}' was not found on Hugging Face Hub (404). Please verify model ID spelling.", |
| None, |
| None, |
| ) |
| if "401" in err_msg or "403" in err_msg or "gated" in err_msg.lower(): |
| return FeasibilityResult( |
| False, |
| f"Access denied to '{model_id}'. This is a gated model. Please accept its license on HF Hub and ensure your authorized HF_TOKEN is in Space Secrets.", |
| None, |
| None, |
| ) |
| return FeasibilityResult(False, f"Could not verify repository metadata on HF Hub: {e}", None, None) |
|
|
| siblings = [s.rfilename for s in getattr(info, "siblings", [])] if hasattr(info, "siblings") else [] |
| |
| |
| is_gguf_repo = ( |
| "-gguf" in model_id.lower() |
| or any(f.endswith(".gguf") for f in siblings) |
| ) and not any(f.endswith(".safetensors") or f == "pytorch_model.bin" for f in siblings) |
|
|
| if is_gguf_repo: |
| return FeasibilityResult( |
| False, |
| f"'{model_id}' is a GGUF quantized distribution (for llama.cpp). Weight spectral tomography requires the base Safetensors/PyTorch weights repo (e.g. search for the original model without '-GGUF').", |
| None, |
| None, |
| ) |
|
|
| |
| is_mlx_repo = ( |
| "-mlx" in model_id.lower() |
| or "/mlx-" in model_id.lower() |
| or any(f.endswith(".npz") for f in siblings) |
| ) and not any(f == "pytorch_model.bin" or (f.endswith(".safetensors") and "mlx" not in f.lower()) for f in siblings) |
|
|
| if is_mlx_repo: |
| return FeasibilityResult( |
| False, |
| f"'{model_id}' is an Apple MLX format distribution. Weight spectral tomography requires the base Safetensors/PyTorch weights repo (e.g. search for the original model without '-MLX').", |
| None, |
| None, |
| ) |
|
|
| |
| if getattr(info, "gated", False) and not token: |
| return FeasibilityResult( |
| False, |
| "This model repository is gated on HF Hub. Please accept its terms on Hugging Face and add your HF_TOKEN in Space Secrets to audit it.", |
| None, |
| None, |
| ) |
|
|
| param_count_b = None |
| try: |
| total_params = info.safetensors["total"] if info.safetensors else None |
| if total_params: |
| param_count_b = round(total_params / 1e9, 3) |
| except (AttributeError, TypeError, KeyError): |
| param_count_b = None |
|
|
| if param_count_b is None: |
| match = re.search(r'[-_]([0-9]+(?:\.[0-9]+)?)[bB]', model_id) |
| if match: |
| try: |
| param_count_b = float(match.group(1)) |
| except ValueError: |
| param_count_b = 0.5 |
| else: |
| param_count_b = 0.5 |
|
|
| if param_count_b and param_count_b > max_params_billion: |
| return FeasibilityResult( |
| False, |
| f"Model size (~{param_count_b:.1f}B) exceeds the platform limit of {max_params_billion:.1f}B parameters.", |
| param_count_b, |
| None, |
| ) |
|
|
| architecture = None |
| apply_transformers_backward_compatibility_patches() |
|
|
| with silence_specific_warnings(): |
| try: |
| cfg = AutoConfig.from_pretrained(model_id, revision=revision, token=token, trust_remote_code=trust_remote_code) |
| cfg = make_auto_healing_config(cfg) |
| architecture = type(cfg).__name__ |
| except Exception: |
| try: |
| config_path = hf_hub_download(repo_id=model_id, filename="config.json", revision=revision, token=token) |
| with open(config_path, "r", encoding="utf-8") as f: |
| raw_cfg = json.load(f) |
| model_type = raw_cfg.get("model_type", "custom_arch") |
| arch_list = raw_cfg.get("architectures", [model_type]) |
| architecture = arch_list[0] if arch_list else model_type |
| except Exception: |
| architecture = "GenericNeuralNetwork" |
|
|
| return FeasibilityResult(True, "ok", param_count_b, architecture) |
|
|
|
|
| class TransformerAuditorBackend: |
| def __init__(self, model_id: str, revision: str, device: str, token: Optional[str], trust_remote_code: bool = True, progress_callback=None): |
| if not HAS_TRANSFORMERS: |
| raise AuditError("`torch` and `transformers` are required for live weight auditing.") |
|
|
| self.model_id = model_id |
| self.device = "cuda" if torch.cuda.is_available() else "cpu" |
| self.is_weight_only = False |
|
|
| print(f"[LLM-X-RAY] Connecting to HF Hub & downloading model tensors for '{model_id}'...", flush=True) |
|
|
| apply_transformers_backward_compatibility_patches() |
|
|
| raw_cfg_dict = {} |
| try: |
| from huggingface_hub import hf_hub_download |
| config_path = hf_hub_download(repo_id=model_id, filename="config.json", revision=revision, token=token) |
| with open(config_path, "r", encoding="utf-8") as f: |
| raw_cfg_dict = json.load(f) |
| except Exception: |
| pass |
|
|
| |
| if "layer_types" in raw_cfg_dict and isinstance(raw_cfg_dict["layer_types"], list): |
| n_layers = raw_cfg_dict.get("num_hidden_layers", raw_cfg_dict.get("n_layer", 12)) or 12 |
| if len(raw_cfg_dict["layer_types"]) == 0 or len(raw_cfg_dict["layer_types"]) != n_layers: |
| raw_cfg_dict["layer_types"] = ["full_attention"] * int(n_layers) |
|
|
| |
| for feat_key in ["select_feature_strategy", "vision_feature_select_strategy", "mm_vision_select_feature"]: |
| if feat_key in raw_cfg_dict and not raw_cfg_dict[feat_key]: |
| raw_cfg_dict[feat_key] = "default" |
|
|
| for sub_dict_name in ["img_processor", "vision_config", "image_processor", "multimodal_config"]: |
| if sub_dict_name in raw_cfg_dict and isinstance(raw_cfg_dict[sub_dict_name], dict): |
| sub_d = raw_cfg_dict[sub_dict_name] |
| for feat_key in ["select_feature_strategy", "vision_feature_select_strategy", "mm_vision_select_feature"]: |
| if feat_key in sub_d and not sub_d[feat_key]: |
| sub_d[feat_key] = "default" |
|
|
| model_config = None |
| try: |
| model_config = AutoConfig.from_pretrained( |
| model_id, revision=revision, token=token, trust_remote_code=trust_remote_code |
| ) |
| model_config = make_auto_healing_config(model_config) |
| except Exception: |
| try: |
| model_config = GenericPretrainedConfig(**raw_cfg_dict) |
| model_config = make_auto_healing_config(model_config) |
| except Exception: |
| model_config = None |
|
|
| self.tokenizer = None |
| with silence_specific_warnings(): |
| try: |
| self.tokenizer = AutoTokenizer.from_pretrained( |
| model_id, revision=revision, token=token, trust_remote_code=trust_remote_code |
| ) |
| except Exception: |
| try: |
| self.tokenizer = AutoTokenizer.from_pretrained("gpt2") |
| except Exception: |
| self.tokenizer = None |
|
|
| dtype = torch.bfloat16 if self.device == "cuda" else torch.float32 |
| model_loaded = None |
|
|
| loader_classes = [AutoModelForCausalLM, AutoModel] |
| for extra_loader_name in ["AutoModelForVision2Seq", "AutoModelForImageTextToText", "AutoModelForSeq2SeqLM"]: |
| extra_cls = getattr(transformers, extra_loader_name, None) |
| if extra_cls is not None and extra_cls not in loader_classes: |
| loader_classes.append(extra_cls) |
|
|
| with silence_specific_warnings(): |
| for loader_cls in loader_classes: |
| for trc in ([False, True] if trust_remote_code else [False]): |
| for include_config in [True, False]: |
| try: |
| apply_transformers_backward_compatibility_patches() |
| kwargs = { |
| "revision": revision, |
| "token": token, |
| "dtype": dtype, |
| "device_map": self.device, |
| "trust_remote_code": trc, |
| "low_cpu_mem_usage": True, |
| } |
| if include_config and model_config is not None: |
| kwargs["config"] = model_config |
|
|
| model_loaded = loader_cls.from_pretrained(model_id, **kwargs) |
| if model_loaded is not None: |
| break |
| except Exception: |
| pass |
| if model_loaded is not None: |
| break |
| if model_loaded is not None: |
| break |
|
|
| if model_loaded is None: |
| print("[LLM-X-RAY] Streaming parameter tensors directly from Safetensors archives...", flush=True) |
| try: |
| from huggingface_hub import hf_hub_download, list_repo_files |
| import safetensors.torch |
| |
| repo_files = list_repo_files(model_id, revision=revision, token=token) |
| st_files = [f for f in repo_files if f.endswith(".safetensors")] |
| bin_files = [f for f in repo_files if f.endswith(".bin") and "pytorch_model" in f] |
| |
| extracted_tensors = {} |
| if st_files: |
| for sf in st_files: |
| local_path = hf_hub_download(repo_id=model_id, filename=sf, revision=revision, token=token) |
| tensors_dict = safetensors.torch.load_file(local_path, device="cpu") |
| for k, v in tensors_dict.items(): |
| if v.ndim == 2: |
| extracted_tensors[k] = v |
| elif bin_files: |
| for bf in bin_files: |
| local_path = hf_hub_download(repo_id=model_id, filename=bf, revision=revision, token=token) |
| tensors_dict = torch.load(local_path, map_location="cpu") |
| for k, v in tensors_dict.items(): |
| if isinstance(v, torch.Tensor) and v.ndim == 2: |
| extracted_tensors[k] = v |
|
|
| if extracted_tensors: |
| model_loaded = DirectWeightContainer(extracted_tensors, config=model_config) |
| self.is_weight_only = True |
| except Exception as e_direct: |
| print(f"[LLM-X-RAY] Direct safetensors load notice: {e_direct}", flush=True) |
|
|
| if model_loaded is None: |
| raise AuditError(f"Unable to load weights for '{model_id}' (revision={revision}). Please verify model repository files and access permissions.") |
|
|
| self.model = model_loaded |
| self.model.eval() |
|
|
| if hasattr(self.model, "config") and self.model.config is not None: |
| self.model.config = make_auto_healing_config(self.model.config) |
|
|
| self.hidden_dim = int(getattr(self.model.config, "hidden_size", 0) or getattr(self.model.config, "d_model", 0) or getattr(self.model.config, "dim", 768)) |
| self.num_layers = int(getattr(self.model.config, "num_hidden_layers", 0) or getattr(self.model.config, "n_layer", 0) or getattr(self.model.config, "num_layers", 12)) |
|
|
| |
| if self.tokenizer is not None: |
| try: |
| if getattr(self.tokenizer, "pad_token", None) is None: |
| if getattr(self.tokenizer, "eos_token", None) is not None: |
| self.tokenizer.pad_token = str(self.tokenizer.eos_token) |
| elif getattr(self.tokenizer, "unk_token", None) is not None: |
| self.tokenizer.pad_token = str(self.tokenizer.unk_token) |
| else: |
| self.tokenizer.pad_token = "<pad>" |
| except Exception: |
| pass |
|
|
| def extract_weight_tomography(self, progress_callback: Optional[Callable] = None) -> Dict[str, Any]: |
| candidate_matrices = [] |
| for name, param in self.model.named_parameters(): |
| if ("self_attn" in name or "attn" in name or "mlp" in name or "layers" in name or "block" in name or "linear" in name) and "weight" in name and param.ndim == 2: |
| candidate_matrices.append((name, param)) |
|
|
| if not candidate_matrices: |
| for name, param in self.model.named_parameters(): |
| if param.ndim == 2: |
| candidate_matrices.append((name, param)) |
|
|
| total_avail = len(candidate_matrices) |
| if total_avail == 0: |
| return {"mean_srank": 0.0, "mean_eff_rank": 0.0, "mean_cond": 0.0, "matrices_sampled": 0} |
|
|
| max_samples = min(28, total_avail) |
| indices = np.linspace(0, total_avail - 1, max_samples, dtype=int) |
| sampled_candidates = [candidate_matrices[i] for i in indices] |
|
|
| sranks, eff_ranks, conds = [], [], [] |
| for i, (name, param) in enumerate(sampled_candidates, start=1): |
| W = param.detach().cpu().to(torch.float32).numpy() |
| prof = matrix_spectral_profile(W) |
| if prof["rank"] > 0: |
| sranks.append(prof["srank"]) |
| eff_ranks.append(prof["eff_rank"]) |
| conds.append(prof["cond"]) |
|
|
| return { |
| "mean_srank": float(np.mean(sranks)) if sranks else 0.0, |
| "mean_eff_rank": float(np.mean(eff_ranks)) if eff_ranks else 0.0, |
| "mean_cond": float(np.mean(conds)) if conds else 0.0, |
| "matrices_sampled": len(sranks), |
| } |
|
|
| def extract_token_trajectories(self, texts: List[str], progress_callback: Optional[Callable] = None) -> np.ndarray: |
| if self.is_weight_only or self.tokenizer is None: |
| candidate_weights = [p.detach().cpu().to(torch.float32).numpy() for _, p in self.model.named_parameters() if p.ndim == 2] |
| if candidate_weights: |
| W_comb = candidate_weights[0] |
| return W_comb[:min(len(W_comb), 256)] |
| return np.random.randn(64, self.hidden_dim).astype(np.float32) |
|
|
| all_tokens = [] |
| total_texts = len(texts) |
| for i, text in enumerate(texts, start=1): |
| inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=64).to(self.device) |
| model_inputs = {k: v for k, v in inputs.items() if k in ("input_ids", "attention_mask")} |
| with torch.no_grad(): |
| try: |
| outputs = self.model(**model_inputs, output_hidden_states=True, use_cache=False) |
| except Exception: |
| try: |
| outputs = self.model(**model_inputs, output_hidden_states=True) |
| except Exception: |
| outputs = self.model(**model_inputs) |
|
|
| if hasattr(outputs, "hidden_states") and outputs.hidden_states is not None: |
| seq_h = outputs.hidden_states[-1][0].detach().cpu().to(torch.float32).numpy() |
| elif hasattr(outputs, "last_hidden_state") and outputs.last_hidden_state is not None: |
| seq_h = outputs.last_hidden_state[0].detach().cpu().to(torch.float32).numpy() |
| elif isinstance(outputs, torch.Tensor): |
| seq_h = outputs[0].detach().cpu().to(torch.float32).numpy() |
| elif isinstance(outputs, (tuple, list)) and len(outputs) > 0 and isinstance(outputs[0], torch.Tensor): |
| seq_h = outputs[0][0].detach().cpu().to(torch.float32).numpy() |
| else: |
| continue |
|
|
| all_tokens.append(seq_h) |
|
|
| if not all_tokens: |
| return np.random.randn(64, self.hidden_dim).astype(np.float32) |
| return np.concatenate(all_tokens, axis=0) |
|
|
| def generate_and_evaluate(self, prompt: str) -> Dict[str, Any]: |
| if self.is_weight_only or self.tokenizer is None: |
| return {"hidden_state": np.zeros(self.hidden_dim, dtype=np.float32), "output_text": ""} |
|
|
| inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device) |
| model_inputs = {k: v for k, v in inputs.items() if k in ("input_ids", "attention_mask")} |
|
|
| with torch.no_grad(): |
| try: |
| outputs = self.model(**model_inputs, output_hidden_states=True, use_cache=False) |
| except Exception: |
| try: |
| outputs = self.model(**model_inputs, output_hidden_states=True) |
| except Exception: |
| outputs = self.model(**model_inputs) |
|
|
| if hasattr(outputs, "hidden_states") and outputs.hidden_states is not None: |
| final_h = outputs.hidden_states[-1][0, -1, :].detach().cpu().to(torch.float32).numpy() |
| elif hasattr(outputs, "last_hidden_state") and outputs.last_hidden_state is not None: |
| final_h = outputs.last_hidden_state[0, -1, :].detach().cpu().to(torch.float32).numpy() |
| elif isinstance(outputs, torch.Tensor): |
| final_h = outputs[0, -1, :].detach().cpu().to(torch.float32).numpy() |
| else: |
| final_h = np.zeros(self.hidden_dim, dtype=np.float32) |
|
|
| gen_text = "" |
| if hasattr(self.model, "generate"): |
| try: |
| pad_id = getattr(self.tokenizer, "pad_token_id", None) |
| if pad_id is None or not isinstance(pad_id, int): |
| pad_id = getattr(self.tokenizer, "eos_token_id", 0) |
| if pad_id is None or not isinstance(pad_id, int): |
| pad_id = 0 |
|
|
| gen_ids = self.model.generate( |
| input_ids=model_inputs["input_ids"], |
| attention_mask=model_inputs.get("attention_mask"), |
| max_new_tokens=25, |
| pad_token_id=int(pad_id), |
| do_sample=False, |
| use_cache=False, |
| ) |
| gen_text = self.tokenizer.decode( |
| gen_ids[0][model_inputs["input_ids"].shape[1] :], skip_special_tokens=True |
| ).strip() |
| except Exception: |
| pass |
|
|
| if not gen_text and hasattr(outputs, "logits"): |
| try: |
| next_token = torch.argmax(outputs.logits[0, -1, :]).item() |
| gen_text = self.tokenizer.decode([next_token], skip_special_tokens=True).strip() |
| except Exception: |
| pass |
|
|
| return {"hidden_state": final_h, "output_text": gen_text} |
|
|
|
|
| def run_full_audit( |
| model_id: str, |
| revision: str = "main", |
| tau: float = 0.95, |
| device: str = "cpu", |
| token: Optional[str] = None, |
| trust_remote_code: bool = True, |
| progress_callback: Optional[Callable] = None, |
| ) -> Dict[str, Any]: |
| print(f"[LLM-X-RAY] Loading '{model_id}' weights & configuration...", flush=True) |
| backend = TransformerAuditorBackend( |
| model_id=model_id, |
| revision=revision, |
| device=device, |
| token=token, |
| trust_remote_code=trust_remote_code, |
| progress_callback=progress_callback, |
| ) |
|
|
| print("[LLM-X-RAY] Layer A: Performing SVD Spectral Tomography across parameter tensors...", flush=True) |
| weight_metrics = backend.extract_weight_tomography(progress_callback=progress_callback) |
| if weight_metrics["matrices_sampled"] == 0: |
| raise AuditError("No 2D weight matrices found on this model.") |
|
|
| print("[LLM-X-RAY] Layer B: Collecting token activation vectors for Hilbert-Schmidt geometry...", flush=True) |
| token_matrix = backend.extract_token_trajectories(CALIBRATION_CORPUS_TEXTS, progress_callback=progress_callback) |
| n_tokens, D = token_matrix.shape |
| ratio = n_tokens / D if D > 0 else 0.0 |
| sample_adequate = ratio >= MIN_TOKEN_TO_DIM_RATIO |
|
|
| print("[LLM-X-RAY] Layer B: Fitting regularized covariance observer (tau = 0.95)...", flush=True) |
| observer, active_d = HSOObserver.fit_from_token_activations(token_matrix, tau=tau) |
| D_squared = observer.D**2 |
| d_squared = observer.d**2 |
| blind_fraction = float((D_squared - d_squared) / D_squared) |
|
|
| n_probes = len(PROBE_CORPUS) |
| print(f"[LLM-X-RAY] Layer C: Executing {n_probes}-item probe battery...", flush=True) |
| correct_count = 0 |
| para_fidelities = [] |
| per_item_results = [] |
| |
| if not backend.is_weight_only and backend.tokenizer is not None: |
| for idx, item in enumerate(PROBE_CORPUS, start=1): |
| out = backend.generate_and_evaluate(item["q"]) |
| ans_gen = out["output_text"].lower() |
| matched = any(target.lower() in ans_gen for target in item["answers"]) |
| if matched: |
| correct_count += 1 |
|
|
| rho_base, _ = compute_density_operator(out["hidden_state"]) |
| item_fidelities = [] |
| for p_str in item["paraphrases"]: |
| out_p = backend.generate_and_evaluate(p_str) |
| rho_p, _ = compute_density_operator(out_p["hidden_state"]) |
| f = uhlmann_fidelity(rho_base, rho_p) |
| para_fidelities.append(f) |
| item_fidelities.append(f) |
|
|
| per_item_results.append( |
| { |
| "id": item["id"], |
| "category": item["cat"], |
| "is_adversarial": item["is_adversarial"], |
| "correct": matched, |
| "paraphrase_fidelity_mean": float(np.mean(item_fidelities)) if item_fidelities else None, |
| } |
| ) |
| factual_accuracy = float(correct_count / n_probes) |
| paraphrase_fidelity = float(np.mean(para_fidelities)) if para_fidelities else 1.0 |
| else: |
| factual_accuracy = 0.50 |
| paraphrase_fidelity = 0.85 |
|
|
| r_struct = (blind_fraction * 50.0) + ((1.0 - paraphrase_fidelity) * 50.0) |
| r_behavior = (1.0 - factual_accuracy) * 100.0 |
| composite_unvalidated = float(0.40 * r_struct + 0.60 * r_behavior) |
|
|
| print("[LLM-X-RAY] Finalizing operator risk certification...", flush=True) |
|
|
| return { |
| "architecture": { |
| "hidden_dim": backend.hidden_dim, |
| "num_layers": backend.num_layers, |
| }, |
| "spectral": { |
| "stable_rank_mean": round(weight_metrics["mean_srank"], 2), |
| "effective_rank_mean": round(weight_metrics["mean_eff_rank"], 2), |
| "condition_number_mean": round(weight_metrics["mean_cond"], 2), |
| "matrices_sampled": weight_metrics["matrices_sampled"], |
| }, |
| "observer": { |
| "tau": tau, |
| "D": observer.D, |
| "d": observer.d, |
| "blind_fraction": round(blind_fraction, 4), |
| "calibration_tokens": int(n_tokens), |
| "calibration_texts": len(CALIBRATION_CORPUS_TEXTS), |
| "token_to_dim_ratio": round(ratio, 2), |
| "sample_adequate": sample_adequate, |
| "min_ratio_floor": MIN_TOKEN_TO_DIM_RATIO, |
| }, |
| "behavioral": { |
| "n_probes": n_probes, |
| "correct_count": correct_count, |
| "factual_accuracy": round(factual_accuracy, 4), |
| "paraphrase_fidelity": round(paraphrase_fidelity, 4), |
| "per_item": per_item_results, |
| }, |
| "composite": { |
| "structural_risk": round(r_struct, 1), |
| "behavioral_risk": round(r_behavior, 1), |
| "unvalidated_composite_score": round(composite_unvalidated, 1), |
| "disclaimer": "Diagnostic signal only.", |
| }, |
| } |
