cognitive_fingerprint.py
1 """ 2 Cognitive Fingerprint 3 4 Models an operator's characteristic patterns of thinking and attention. 5 This is the "theory of self" that the system builds over time. 6 7 Key components: 8 - Gravity Wells: Topics/concepts that consistently attract attention 9 - Pattern Preferences: Structural preferences (first-principles, narratives, contrarian) 10 - Altitude Distribution: Where they typically think (operational → philosophical) 11 - Resonance Triggers: What makes something "stick" for them 12 """ 13 14 from dataclasses import dataclass, field 15 from datetime import datetime, timedelta 16 from typing import Optional, List, Dict, Any, Set, Tuple 17 from enum import Enum 18 import math 19 20 21 class PatternType(Enum): 22 """Types of thinking patterns.""" 23 FIRST_PRINCIPLES = "first_principles" # Breaking down to fundamentals 24 SYSTEMS_THINKING = "systems_thinking" # Seeing interconnections 25 NARRATIVE_ARC = "narrative_arc" # Story and journey structure 26 CONTRARIAN = "contrarian" # Going against consensus 27 ANALOGICAL = "analogical" # Learning by analogy 28 QUANTITATIVE = "quantitative" # Numbers and metrics driven 29 HEURISTIC = "heuristic" # Rules of thumb, patterns 30 HISTORICAL = "historical" # Learning from history 31 EXPERIENTIAL = "experiential" # Learning from experience 32 PHILOSOPHICAL = "philosophical" # Meaning and purpose driven 33 34 35 class TemporalScope(Enum): 36 """How long a gravity well remains active.""" 37 PERMANENT = "permanent" # Core principles, axioms - always resonate 38 SEASONAL = "seasonal" # Projects, themes - weeks to months 39 CONTEXTUAL = "contextual" # Current focus - days to weeks 40 EPHEMERAL = "ephemeral" # Immediate task - hours to days 41 42 43 class AltitudeScope(Enum): 44 """Which thinking levels a gravity well applies to.""" 45 ALL = "all" # Resonates at every altitude 46 PHILOSOPHICAL = "philosophical" # Only at philosophical level 47 STRATEGIC = "strategic" # Strategic and above 48 TACTICAL = "tactical" # Tactical and below 49 OPERATIONAL = "operational" # Only operational 50 51 52 @dataclass 53 class GravityWell: 54 """ 55 A topic or concept that consistently attracts operator attention. 56 57 Gravity wells have mass (how strongly they attract) and 58 pull radius (how far their influence extends to related topics). 59 60 NEW: Scope dimensions that bound where/when resonance applies: 61 - temporal_scope: How long this well persists 62 - altitude_scope: Which thinking levels it applies to 63 - topic_scope: Which domains it's restricted to (empty = all) 64 - source_nodes: Which graph nodes formed this well 65 """ 66 concept: str # Core concept name 67 mass: float = 0.5 # Attraction strength (0-1) 68 pull_radius: float = 0.3 # Influence on related topics (0-1) 69 related_concepts: List[str] = field(default_factory=list) 70 71 # === SCOPE DIMENSIONS === 72 temporal_scope: TemporalScope = TemporalScope.CONTEXTUAL 73 altitude_scope: AltitudeScope = AltitudeScope.ALL 74 topic_scope: Set[str] = field(default_factory=set) # Empty = all topics 75 source_nodes: List[str] = field(default_factory=list) # UUIDs of forming nodes 76 77 # History 78 first_detected: datetime = field(default_factory=datetime.now) 79 last_activated: datetime = field(default_factory=datetime.now) 80 activation_count: int = 1 81 prediction_accuracy: float = 0.5 # How well we predict resonance near this well 82 83 # Formation tracking 84 formation_context: str = "" # What was happening when well formed 85 edge_count_at_formation: int = 0 # How connected was origin 86 87 def activate(self): 88 """Record an activation of this gravity well.""" 89 self.last_activated = datetime.now() 90 self.activation_count += 1 91 # Strengthen mass with use (capped) 92 self.mass = min(0.95, self.mass + 0.02) 93 94 def decay(self, days_since_activation: int): 95 """Apply time-based decay to gravity well.""" 96 # Decay rate depends on temporal scope 97 if self.temporal_scope == TemporalScope.PERMANENT: 98 decay_factor = 0.999 ** days_since_activation # Very slow 99 elif self.temporal_scope == TemporalScope.SEASONAL: 100 decay_factor = 0.995 ** days_since_activation # Slow 101 elif self.temporal_scope == TemporalScope.CONTEXTUAL: 102 decay_factor = 0.99 ** days_since_activation # Normal 103 else: # EPHEMERAL 104 decay_factor = 0.95 ** days_since_activation # Fast 105 106 self.mass = max(0.1, self.mass * decay_factor) 107 108 def applies_at_altitude(self, altitude: str) -> bool: 109 """Check if this well applies at a given altitude.""" 110 if self.altitude_scope == AltitudeScope.ALL: 111 return True 112 113 altitude_order = ['operational', 'tactical', 'strategic', 'philosophical'] 114 try: 115 target_idx = altitude_order.index(altitude) 116 scope_idx = altitude_order.index(self.altitude_scope.value) 117 except ValueError: 118 return True # Unknown altitude, apply 119 120 if self.altitude_scope in (AltitudeScope.PHILOSOPHICAL, AltitudeScope.STRATEGIC): 121 return target_idx >= scope_idx # Only at this level and above 122 else: 123 return target_idx <= scope_idx # Only at this level and below 124 125 def applies_to_topic(self, topic: str) -> bool: 126 """Check if this well applies to a given topic.""" 127 if not self.topic_scope: # Empty set = all topics 128 return True 129 return topic.lower() in {t.lower() for t in self.topic_scope} 130 131 def get_effective_mass(self, altitude: str = None, topic: str = None) -> float: 132 """Get mass adjusted for scope constraints.""" 133 if altitude and not self.applies_at_altitude(altitude): 134 return 0.0 135 if topic and not self.applies_to_topic(topic): 136 return 0.0 137 return self.mass 138 139 140 @dataclass 141 class PatternPreference: 142 """ 143 A preferred pattern of thinking. 144 145 Different operators gravitate toward different thinking styles. 146 """ 147 pattern_type: PatternType 148 strength: float = 0.5 # How strongly they prefer this pattern 149 contexts: List[str] = field(default_factory=list) # Where they use this pattern 150 evidence_count: int = 0 # How many times observed 151 152 153 @dataclass 154 class ResonanceTrigger: 155 """ 156 A specific trigger that makes content resonate. 157 158 These are the micro-patterns that predict resonance. 159 """ 160 trigger_type: str # 'surprise', 'validation', 'connection', etc. 161 description: str 162 strength: float = 0.5 163 examples: List[str] = field(default_factory=list) 164 165 166 @dataclass 167 class CognitiveFingerprint: 168 """ 169 Complete model of an operator's cognitive patterns. 170 171 This is the "theory of self" that enables resonance prediction. 172 """ 173 operator_id: str 174 created_at: datetime = field(default_factory=datetime.now) 175 updated_at: datetime = field(default_factory=datetime.now) 176 177 # Gravity wells - topics that attract attention 178 gravity_wells: Dict[str, GravityWell] = field(default_factory=dict) 179 180 # Pattern preferences - how they like to think 181 pattern_preferences: Dict[PatternType, PatternPreference] = field(default_factory=dict) 182 183 # Altitude distribution - where they typically operate 184 altitude_distribution: Dict[str, float] = field(default_factory=lambda: { 185 'operational': 0.25, 186 'tactical': 0.25, 187 'strategic': 0.25, 188 'philosophical': 0.25, 189 }) 190 191 # Resonance triggers - what makes things stick 192 resonance_triggers: List[ResonanceTrigger] = field(default_factory=list) 193 194 # Training metrics 195 total_predictions: int = 0 196 correct_predictions: int = 0 197 training_sessions: int = 0 198 199 @property 200 def prediction_accuracy(self) -> float: 201 """Overall prediction accuracy.""" 202 if self.total_predictions == 0: 203 return 0.5 204 return self.correct_predictions / self.total_predictions 205 206 def add_gravity_well( 207 self, 208 concept: str, 209 initial_mass: float = 0.5, 210 related_concepts: Optional[List[str]] = None 211 ) -> GravityWell: 212 """Add or strengthen a gravity well.""" 213 if concept in self.gravity_wells: 214 self.gravity_wells[concept].activate() 215 else: 216 self.gravity_wells[concept] = GravityWell( 217 concept=concept, 218 mass=initial_mass, 219 related_concepts=related_concepts or [] 220 ) 221 self.updated_at = datetime.now() 222 return self.gravity_wells[concept] 223 224 def get_gravity_for_concept(self, concept: str) -> float: 225 """ 226 Get total gravitational pull for a concept. 227 228 Includes direct gravity wells and influence from nearby wells. 229 """ 230 concept_lower = concept.lower() 231 232 total_gravity = 0.0 233 234 for well_name, well in self.gravity_wells.items(): 235 # Direct match 236 if well_name.lower() in concept_lower or concept_lower in well_name.lower(): 237 total_gravity += well.mass 238 # Related concept match 239 elif any(rel.lower() in concept_lower for rel in well.related_concepts): 240 total_gravity += well.mass * well.pull_radius 241 242 return min(1.0, total_gravity) 243 244 def update_pattern_preference( 245 self, 246 pattern_type: PatternType, 247 strength_delta: float = 0.05, 248 context: Optional[str] = None 249 ): 250 """Update preference for a thinking pattern.""" 251 if pattern_type not in self.pattern_preferences: 252 self.pattern_preferences[pattern_type] = PatternPreference( 253 pattern_type=pattern_type 254 ) 255 256 pref = self.pattern_preferences[pattern_type] 257 pref.strength = max(0.0, min(1.0, pref.strength + strength_delta)) 258 pref.evidence_count += 1 259 260 if context and context not in pref.contexts: 261 pref.contexts.append(context) 262 263 self.updated_at = datetime.now() 264 265 def update_altitude_distribution( 266 self, 267 altitude: str, 268 weight: float = 1.0 269 ): 270 """ 271 Update altitude distribution based on observed behavior. 272 273 Uses exponential moving average for stability. 274 """ 275 alpha = 0.1 # Learning rate 276 277 if altitude in self.altitude_distribution: 278 # Increase observed altitude 279 for alt in self.altitude_distribution: 280 if alt == altitude: 281 self.altitude_distribution[alt] = ( 282 (1 - alpha) * self.altitude_distribution[alt] + 283 alpha * weight 284 ) 285 else: 286 # Slight decrease for others 287 self.altitude_distribution[alt] *= (1 - alpha / 3) 288 289 # Normalize 290 total = sum(self.altitude_distribution.values()) 291 if total > 0: 292 self.altitude_distribution = { 293 k: v / total for k, v in self.altitude_distribution.items() 294 } 295 296 self.updated_at = datetime.now() 297 298 def add_resonance_trigger( 299 self, 300 trigger_type: str, 301 description: str, 302 example: Optional[str] = None 303 ): 304 """Add or strengthen a resonance trigger.""" 305 # Find existing 306 for trigger in self.resonance_triggers: 307 if trigger.trigger_type == trigger_type: 308 trigger.strength = min(1.0, trigger.strength + 0.05) 309 if example and example not in trigger.examples: 310 trigger.examples.append(example) 311 self.updated_at = datetime.now() 312 return 313 314 # Create new 315 self.resonance_triggers.append(ResonanceTrigger( 316 trigger_type=trigger_type, 317 description=description, 318 strength=0.5, 319 examples=[example] if example else [] 320 )) 321 self.updated_at = datetime.now() 322 323 def get_top_gravity_wells(self, n: int = 10) -> List[GravityWell]: 324 """Get the strongest gravity wells.""" 325 wells = list(self.gravity_wells.values()) 326 wells.sort(key=lambda w: w.mass, reverse=True) 327 return wells[:n] 328 329 def get_characteristic_patterns(self) -> List[PatternPreference]: 330 """Get patterns sorted by strength.""" 331 patterns = list(self.pattern_preferences.values()) 332 patterns.sort(key=lambda p: p.strength, reverse=True) 333 return patterns 334 335 def get_dominant_altitude(self) -> str: 336 """Get the altitude where operator spends most time.""" 337 return max(self.altitude_distribution.keys(), 338 key=lambda k: self.altitude_distribution[k]) 339 340 def record_prediction_result(self, predicted: bool, actual: bool): 341 """Record a prediction result for accuracy tracking.""" 342 self.total_predictions += 1 343 if predicted == actual: 344 self.correct_predictions += 1 345 self.updated_at = datetime.now() 346 347 def to_dict(self) -> Dict[str, Any]: 348 """Serialize for storage.""" 349 return { 350 'operator_id': self.operator_id, 351 'created_at': self.created_at.isoformat(), 352 'updated_at': self.updated_at.isoformat(), 353 'gravity_wells': { 354 name: { 355 'concept': w.concept, 356 'mass': w.mass, 357 'pull_radius': w.pull_radius, 358 'related_concepts': w.related_concepts, 359 'activation_count': w.activation_count, 360 'prediction_accuracy': w.prediction_accuracy, 361 } 362 for name, w in self.gravity_wells.items() 363 }, 364 'pattern_preferences': { 365 pt.name: { 366 'strength': p.strength, 367 'evidence_count': p.evidence_count, 368 'contexts': p.contexts, 369 } 370 for pt, p in self.pattern_preferences.items() 371 }, 372 'altitude_distribution': self.altitude_distribution, 373 'resonance_triggers': [ 374 { 375 'type': t.trigger_type, 376 'description': t.description, 377 'strength': t.strength, 378 } 379 for t in self.resonance_triggers 380 ], 381 'prediction_accuracy': self.prediction_accuracy, 382 'total_predictions': self.total_predictions, 383 'training_sessions': self.training_sessions, 384 } 385 386 @classmethod 387 def from_dict(cls, data: Dict[str, Any]) -> 'CognitiveFingerprint': 388 """Deserialize from storage.""" 389 fp = cls(operator_id=data['operator_id']) 390 fp.created_at = datetime.fromisoformat(data['created_at']) 391 fp.updated_at = datetime.fromisoformat(data['updated_at']) 392 393 for name, w_data in data.get('gravity_wells', {}).items(): 394 fp.gravity_wells[name] = GravityWell( 395 concept=w_data['concept'], 396 mass=w_data['mass'], 397 pull_radius=w_data['pull_radius'], 398 related_concepts=w_data['related_concepts'], 399 activation_count=w_data['activation_count'], 400 prediction_accuracy=w_data.get('prediction_accuracy', 0.5), 401 ) 402 403 for pt_name, p_data in data.get('pattern_preferences', {}).items(): 404 pt = PatternType[pt_name] 405 fp.pattern_preferences[pt] = PatternPreference( 406 pattern_type=pt, 407 strength=p_data['strength'], 408 evidence_count=p_data['evidence_count'], 409 contexts=p_data['contexts'], 410 ) 411 412 fp.altitude_distribution = data.get('altitude_distribution', fp.altitude_distribution) 413 fp.total_predictions = data.get('total_predictions', 0) 414 fp.correct_predictions = int(data.get('prediction_accuracy', 0.5) * fp.total_predictions) 415 fp.training_sessions = data.get('training_sessions', 0) 416 417 return fp 418 419 420 # Quick test 421 if __name__ == "__main__": 422 print("=== Cognitive Fingerprint Test ===\n") 423 424 # Create fingerprint for Rick 425 fp = CognitiveFingerprint(operator_id="rick") 426 427 # Add gravity wells based on known interests 428 fp.add_gravity_well( 429 "markov_blankets", 430 initial_mass=0.8, 431 related_concepts=["free_energy", "boundaries", "sovereignty", "emergence"] 432 ) 433 fp.add_gravity_well( 434 "architecture", 435 initial_mass=0.7, 436 related_concepts=["design", "systems", "structure", "patterns"] 437 ) 438 fp.add_gravity_well( 439 "survival", 440 initial_mass=0.6, 441 related_concepts=["resilience", "persistence", "will", "determination"] 442 ) 443 fp.add_gravity_well( 444 "compounding", 445 initial_mass=0.7, 446 related_concepts=["exponential", "growth", "leverage", "duration"] 447 ) 448 449 # Add pattern preferences 450 fp.update_pattern_preference(PatternType.FIRST_PRINCIPLES, strength_delta=0.3) 451 fp.update_pattern_preference(PatternType.SYSTEMS_THINKING, strength_delta=0.25) 452 fp.update_pattern_preference(PatternType.ANALOGICAL, strength_delta=0.2) 453 454 # Add resonance triggers 455 fp.add_resonance_trigger( 456 "pattern_recognition", 457 "When a familiar pattern appears in a new context" 458 ) 459 fp.add_resonance_trigger( 460 "first_principles_clarity", 461 "When something complex is distilled to essentials" 462 ) 463 fp.add_resonance_trigger( 464 "contrarian_wisdom", 465 "When conventional wisdom is revealed as wrong" 466 ) 467 468 # Update altitude distribution 469 fp.update_altitude_distribution('strategic', weight=1.5) 470 fp.update_altitude_distribution('philosophical', weight=1.2) 471 472 # Print results 473 print("Top Gravity Wells:") 474 for well in fp.get_top_gravity_wells(5): 475 print(f" {well.concept}: mass={well.mass:.2f}") 476 477 print("\nCharacteristic Patterns:") 478 for pattern in fp.get_characteristic_patterns()[:3]: 479 print(f" {pattern.pattern_type.name}: {pattern.strength:.2f}") 480 481 print("\nAltitude Distribution:") 482 for alt, weight in sorted(fp.altitude_distribution.items(), key=lambda x: -x[1]): 483 print(f" {alt}: {weight:.2%}") 484 485 print(f"\nDominant Altitude: {fp.get_dominant_altitude()}") 486 487 # Test gravity calculation 488 print("\nGravity Calculations:") 489 test_concepts = ["architecture", "design patterns", "survival mindset", "quantum physics"] 490 for concept in test_concepts: 491 gravity = fp.get_gravity_for_concept(concept) 492 print(f" '{concept}': gravity={gravity:.2f}") 493 494 # Serialize and deserialize 495 data = fp.to_dict() 496 fp2 = CognitiveFingerprint.from_dict(data) 497 print(f"\nSerialization roundtrip: {fp2.operator_id}, {len(fp2.gravity_wells)} wells") 498 499 print("\n=== Test Complete ===")