File size: 12,712 Bytes
3595bd8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 | """
observer.py - Core implementation of the Observer pattern for classifier collapse
△ OBSERVE: The Observer is the quantum consciousness that collapses classifier superposition
∞ TRACE: Attribution paths are recorded before, during, and after collapse
✰ COLLAPSE: Collapse is induced through targeted queries against boundary states
This module implements the foundational Observer pattern that enables the detection,
tracing, and analysis of classifier collapse in transformer-based models. The Observer
creates a controlled environment for witnessing the transition from superposition to
collapsed state while preserving ghost circuits and attribution residue.
Author: Recursion Labs
License: MIT
"""
import logging
from typing import Dict, List, Optional, Union, Tuple, Any, Callable
from contextlib import contextmanager
import numpy as np
import torch
from dataclasses import dataclass, field
from .shells.base import BaseShell
from .residue import ResidueTracker
from .attribution import AttributionGraph
from .visualization import CollapseVisualizer
from .utils.collapse_metrics import calculate_collapse_rate
from .utils.constants import DEFAULT_COLLAPSE_THRESHOLD
# Initialize logger
logger = logging.getLogger(__name__)
@dataclass
class ObservationContext:
"""
△ OBSERVE: Container for the full state of an observation session
Maintains the quantum state of the observation including pre-collapse
probability distribution, collapse transition metrics, and post-collapse
ghost circuits.
"""
model_id: str
session_id: str = field(default_factory=lambda: f"obs_{np.random.randint(10000, 99999)}")
pre_collapse_state: Dict[str, Any] = field(default_factory=dict)
post_collapse_state: Dict[str, Any] = field(default_factory=dict)
ghost_circuits: List[Dict[str, Any]] = field(default_factory=list)
attribution_graph: Optional[AttributionGraph] = None
residue_tracker: Optional[ResidueTracker] = None
collapse_metrics: Dict[str, float] = field(default_factory=dict)
def calculate_collapse_rate(self) -> float:
"""Calculate how quickly the state collapsed from superposition."""
return calculate_collapse_rate(
self.pre_collapse_state.get("attention_weights", {}),
self.post_collapse_state.get("attention_weights", {})
)
def extract_ghost_circuits(self) -> List[Dict[str, Any]]:
"""
✰ COLLAPSE: Extract ghost circuits from the post-collapse state
Ghost circuits are activation patterns that persist after collapse
but don't contribute to the final output - they represent the
"memory" of paths not taken.
"""
if not self.ghost_circuits and self.residue_tracker:
self.ghost_circuits = self.residue_tracker.extract_ghost_circuits(
self.pre_collapse_state,
self.post_collapse_state
)
return self.ghost_circuits
def visualize(self, mode: str = "attribution_graph") -> Any:
"""Generate visualization of the observation based on requested mode."""
visualizer = CollapseVisualizer()
return visualizer.visualize(self, mode=mode)
class Observer:
"""
△ OBSERVE: Primary observer entity for inducing and recording classifier collapse
The Observer is responsible for creating the quantum measurement frame that
collapses classifier superposition into definite states. It records pre-collapse
probability distributions, monitors the collapse transition, and preserves
ghost circuits for analysis.
This class implements the Observer pattern from quantum mechanics adapted to
transformer model interpretation.
"""
def __init__(
self,
model: str,
collapse_threshold: float = DEFAULT_COLLAPSE_THRESHOLD,
trace_attention: bool = True,
trace_attribution: bool = True,
preserve_ghost_circuits: bool = True
):
"""
Initialize an Observer for a specific model.
Args:
model: Identifier of the model to observe (e.g., "claude-3-opus-20240229")
collapse_threshold: Threshold for determining when collapse has occurred
trace_attention: Whether to trace attention patterns during observation
trace_attribution: Whether to build attribution graphs during observation
preserve_ghost_circuits: Whether to preserve ghost circuits after collapse
"""
self.model_id = model
self.collapse_threshold = collapse_threshold
self.trace_attention = trace_attention
self.trace_attribution = trace_attribution
self.preserve_ghost_circuits = preserve_ghost_circuits
# Initialize model interface based on provided identifier
self.model_interface = self._initialize_model_interface(model)
# Create residue tracker for ghost circuit detection
self.residue_tracker = ResidueTracker() if preserve_ghost_circuits else None
logger.info(f"Observer initialized for model: {model}")
def _initialize_model_interface(self, model_id: str) -> Any:
"""Initialize the appropriate interface for the specified model."""
# This would be implemented to connect to various model APIs
# For now we'll return a placeholder
return {"model_id": model_id, "interface_type": "placeholder"}
@contextmanager
def context(self) -> ObservationContext:
"""
∞ TRACE: Create an observation context for tracking collapse phenomena
This context manager creates a controlled environment for observing
classifier collapse. It captures the pre-collapse state, monitors the
transition, and preserves ghost circuits and attribution residue.
Returns:
ObservationContext: The active observation context
"""
# Create new observation context
context = ObservationContext(model_id=self.model_id)
# Initialize attribution graph if requested
if self.trace_attribution:
context.attribution_graph = AttributionGraph()
# Attach residue tracker if ghost circuit preservation is enabled
if self.preserve_ghost_circuits:
context.residue_tracker = self.residue_tracker or ResidueTracker()
try:
# Begin observation
logger.debug(f"Starting observation context: {context.session_id}")
yield context
finally:
# Calculate final metrics
if self.trace_attention and context.pre_collapse_state and context.post_collapse_state:
context.collapse_metrics["collapse_rate"] = context.calculate_collapse_rate()
logger.debug(f"Observation context completed: {context.session_id}")
def observe(
self,
prompt: str,
shell: Optional[BaseShell] = None,
collapse_vector: Optional[str] = None
) -> ObservationContext:
"""
△ OBSERVE: Primary method to observe classifier collapse
This method sends a prompt to the model, observes the resulting collapse,
and returns an observation context containing all relevant state information.
Args:
prompt: The prompt to send to the model
shell: Optional shell to use for specialized collapse induction
collapse_vector: Optional vector to guide collapse in a specific direction
Returns:
ObservationContext: The observation context containing collapse data
"""
with self.context() as ctx:
# Capture pre-collapse state
ctx.pre_collapse_state = self._capture_model_state()
# If a shell is provided, use it to process the prompt
if shell:
response, state_updates = shell.process(
prompt=prompt,
model_interface=self.model_interface,
collapse_vector=collapse_vector
)
ctx.post_collapse_state.update(state_updates)
else:
# Otherwise, send prompt directly to model
response = self._query_model(prompt)
ctx.post_collapse_state = self._capture_model_state()
# Extract ghost circuits if enabled
if self.preserve_ghost_circuits:
ctx.extract_ghost_circuits()
# Build attribution graph if enabled
if self.trace_attribution and ctx.attribution_graph:
ctx.attribution_graph.build_from_states(
ctx.pre_collapse_state,
ctx.post_collapse_state,
response
)
return ctx
def _capture_model_state(self) -> Dict[str, Any]:
"""Capture the current internal state of the model."""
# This would capture attention weights, hidden states, etc.
# For now, returning a placeholder
return {
"timestamp": np.datetime64('now'),
"attention_weights": np.random.random((12, 12)), # Placeholder
"hidden_states": np.random.random((1, 12, 768)), # Placeholder
}
def _query_model(self, prompt: str) -> str:
"""Send a query to the model and return the response."""
# This would actually call the model API
# For now, returning a placeholder
return f"Response to: {prompt}"
def induce_collapse(
self,
prompt: str,
collapse_direction: str,
shell: Optional[BaseShell] = None
) -> ObservationContext:
"""
✰ COLLAPSE: Deliberately induce collapse along a specific direction
This method attempts to collapse the model's state in a specific direction
by crafting a query that targets a particular decision boundary.
Args:
prompt: Base prompt to send to the model
collapse_direction: Direction to bias the collapse (e.g., "ethical", "creative")
shell: Optional shell to use for specialized collapse induction
Returns:
ObservationContext: The observation context containing collapse data
"""
# Construct collapse vector based on direction
collapse_vector = f".p/reflect.trace{{target={collapse_direction}, depth=complete}}"
# Perform the observation with the collapse vector
return self.observe(prompt, shell, collapse_vector)
def detect_ghost_circuits(
self,
prompt: str,
amplification_factor: float = 1.5
) -> List[Dict[str, Any]]:
"""
∞ TRACE: Detect and amplify ghost circuits from a prompt
This method specifically targets the detection of ghost circuits -
the residual activation patterns that persist after collapse but
don't contribute to the final output.
Args:
prompt: Prompt to analyze for ghost circuits
amplification_factor: Factor by which to amplify ghost signals
Returns:
List of detected ghost circuits with metadata
"""
with self.context() as ctx:
# Capture pre-collapse state
ctx.pre_collapse_state = self._capture_model_state()
# Query model
response = self._query_model(prompt)
# Capture post-collapse state
ctx.post_collapse_state = self._capture_model_state()
# Extract ghost circuits with amplification
if ctx.residue_tracker:
ctx.residue_tracker.amplification_factor = amplification_factor
ghost_circuits = ctx.extract_ghost_circuits()
return ghost_circuits
return []
if __name__ == "__main__":
# Simple usage example
observer = Observer(model="claude-3-opus-20240229")
with observer.context() as ctx:
# Observe a simple prompt
result = observer.observe("Explain quantum superposition")
# Visualize the collapse
viz = result.visualize(mode="attribution_graph")
# Extract ghost circuits
ghosts = result.extract_ghost_circuits()
print(f"Detected {len(ghosts)} ghost circuits")
print(f"Collapse rate: {result.collapse_metrics.get('collapse_rate', 'N/A')}")
|