""" AIVA Queen Curiosity-Driven Learning Engine ============================================ A comprehensive curiosity-driven learning system implementing intrinsic motivation mechanisms for autonomous knowledge acquisition. Components: 1. InformationGainCalculator - Calculate expected information gain from queries 2. NoveltyDetector - Detect novel patterns and situations 3. ExplorationPolicy - Balance exploration vs exploitation 4. QuestionGenerator - Generate curious questions for knowledge gaps 5. KnowledgeGapFinder - Find gaps in knowledge base 6. InterestModel - Model and track interests over time This module enables AIVA Queen to autonomously identify what to learn next, driven by intrinsic curiosity rather than just external rewards. Author: Genesis System / AIVA Queen Version: 1.0.0 """ import json import time import math import random import logging import hashlib import threading from abc import ABC, abstractmethod from enum import Enum, auto from dataclasses import dataclass, field, asdict from typing import ( Dict, List, Optional, Any, Tuple, Callable, Set, Union, TypeVar, Generic, Iterator, Protocol ) from collections import defaultdict, deque from datetime import datetime, timedelta from pathlib import Path import heapq import uuid # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger("CuriosityEngine") # ============================================================================== # Type Definitions and Enums # ============================================================================== T = TypeVar('T') State = TypeVar('State') Action = TypeVar('Action') class NoveltyType(Enum): """Types of detected novelty.""" FEATURE_NOVELTY = auto() # Novel feature combinations SEMANTIC_NOVELTY = auto() # Semantically new concepts STRUCTURAL_NOVELTY = auto() # Novel structure/patterns TEMPORAL_NOVELTY = auto() # Temporal pattern changes DISTRIBUTION_SHIFT = auto() # Distribution changes UNKNOWN = auto() class ExplorationMode(Enum): """Exploration strategy modes.""" EPSILON_GREEDY = "epsilon_greedy" UCB = "upper_confidence_bound" THOMPSON_SAMPLING = "thompson_sampling" CURIOSITY_DRIVEN = "curiosity_driven" RANDOM = "random" HYBRID = "hybrid" class QuestionType(Enum): """Types of curious questions.""" FACTUAL = auto() # What is X? CAUSAL = auto() # Why does X happen? PROCEDURAL = auto() # How to do X? COMPARATIVE = auto() # How does X compare to Y? COUNTERFACTUAL = auto() # What if X were different? EXPLORATORY = auto() # What other X exist? PREDICTIVE = auto() # What will happen if X? DEFINITIONAL = auto() # What defines X? class InterestCategory(Enum): """Categories of interest areas.""" DOMAIN_KNOWLEDGE = auto() SKILL_ACQUISITION = auto() PROBLEM_SOLVING = auto() PATTERN_RECOGNITION = auto() TOOL_USAGE = auto() SOCIAL_INTERACTION = auto() CREATIVE_EXPRESSION = auto() OPTIMIZATION = auto() # ============================================================================== # Data Classes # ============================================================================== @dataclass class InformationState: """Represents the current information state about a concept.""" concept_id: str name: str certainty: float = 0.0 # How certain we are (0-1) completeness: float = 0.0 # How complete knowledge is (0-1) freshness: float = 1.0 # How fresh the knowledge is (0-1) access_count: int = 0 last_accessed: float = field(default_factory=time.time) related_concepts: Set[str] = field(default_factory=set) metadata: Dict[str, Any] = field(default_factory=dict) def decay_freshness(self, decay_rate: float = 0.01): """Apply time-based decay to freshness.""" time_delta = time.time() - self.last_accessed self.freshness *= math.exp(-decay_rate * time_delta / 3600) def get_knowledge_score(self) -> float: """Get overall knowledge score.""" return (self.certainty + self.completeness + self.freshness) / 3.0 @dataclass class NoveltySignal: """Represents a detected novelty signal.""" signal_id: str = field(default_factory=lambda: str(uuid.uuid4())[:12]) novelty_type: NoveltyType = NoveltyType.UNKNOWN source: str = "" intensity: float = 0.0 # How novel (0-1) surprise: float = 0.0 # Surprise level (0-1) features: Dict[str, float] = field(default_factory=dict) timestamp: float = field(default_factory=time.time) context: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: result = asdict(self) result['novelty_type'] = self.novelty_type.name return result @dataclass class CuriousQuestion: """Represents a generated curious question.""" question_id: str = field(default_factory=lambda: str(uuid.uuid4())[:12]) question_text: str = "" question_type: QuestionType = QuestionType.FACTUAL target_concept: str = "" information_gain: float = 0.0 priority: float = 0.0 asked: bool = False answered: bool = False answer: Optional[str] = None created_at: float = field(default_factory=time.time) metadata: Dict[str, Any] = field(default_factory=dict) def __lt__(self, other: 'CuriousQuestion') -> bool: """Enable priority queue comparison.""" return self.priority > other.priority @dataclass class KnowledgeGap: """Represents a gap in knowledge.""" gap_id: str = field(default_factory=lambda: str(uuid.uuid4())[:12]) concept: str = "" gap_type: str = "missing" # missing, incomplete, outdated, uncertain severity: float = 0.0 # How important to fill (0-1) potential_gain: float = 0.0 # Expected gain from filling related_gaps: Set[str] = field(default_factory=set) fill_difficulty: float = 0.5 # How hard to fill (0-1) discovered_at: float = field(default_factory=time.time) filled_at: Optional[float] = None @dataclass class InterestProfile: """Models interest in a particular area.""" interest_id: str category: InterestCategory name: str base_interest: float = 0.5 # Base interest level (0-1) current_interest: float = 0.5 # Current interest (varies) engagement_history: List[float] = field(default_factory=list) last_engaged: float = field(default_factory=time.time) total_time_spent: float = 0.0 # Seconds discovery_count: int = 0 # Things discovered satisfaction_score: float = 0.5 # How satisfying pursuits have been @dataclass class ExplorationResult: """Result of an exploration action.""" action_taken: str information_gained: float novelty_encountered: float questions_generated: int gaps_filled: int interest_change: float success: bool metadata: Dict[str, Any] = field(default_factory=dict) # ============================================================================== # Information Gain Calculator # ============================================================================== class InformationGainCalculator: """ Calculates expected information gain from potential queries or actions. Uses entropy-based measures to estimate how much a query would reduce uncertainty about the world model. """ def __init__( self, prior_weight: float = 0.5, evidence_discount: float = 0.1, temporal_discount: float = 0.01, min_gain_threshold: float = 0.01 ): """ Initialize the information gain calculator. Args: prior_weight: Weight given to prior beliefs evidence_discount: Discount for conflicting evidence temporal_discount: Discount for old information min_gain_threshold: Minimum gain to consider significant """ self.prior_weight = prior_weight self.evidence_discount = evidence_discount self.temporal_discount = temporal_discount self.min_gain_threshold = min_gain_threshold # Track beliefs about concepts self.beliefs: Dict[str, Dict[str, float]] = defaultdict( lambda: {"probability": 0.5, "certainty": 0.1} ) # Evidence history self.evidence_history: List[Dict[str, Any]] = [] # Cache for computed gains self._gain_cache: Dict[str, Tuple[float, float]] = {} self._cache_ttl = 300 # 5 minutes self._lock = threading.RLock() logger.info("InformationGainCalculator initialized") def calculate_entropy( self, probabilities: Dict[str, float] ) -> float: """ Calculate Shannon entropy of a probability distribution. Args: probabilities: Probability distribution {outcome: probability} Returns: Entropy value in bits """ entropy = 0.0 for p in probabilities.values(): if p > 0 and p < 1: entropy -= p * math.log2(p) return entropy def calculate_expected_gain( self, query: str, possible_outcomes: Dict[str, float], current_beliefs: Optional[Dict[str, float]] = None ) -> float: """ Calculate expected information gain from a query. Args: query: The query/action being considered possible_outcomes: Possible outcomes and their probabilities current_beliefs: Current belief state (optional) Returns: Expected information gain in bits """ with self._lock: # Check cache cache_key = f"{query}_{hash(frozenset(possible_outcomes.items()))}" if cache_key in self._gain_cache: cached_gain, cached_time = self._gain_cache[cache_key] if time.time() - cached_time < self._cache_ttl: return cached_gain # Calculate current entropy (uncertainty) current_beliefs = current_beliefs or self.beliefs.get(query, {}) if not current_beliefs: current_beliefs = {"unknown": 0.5, "known": 0.5} current_entropy = self.calculate_entropy(current_beliefs) # Calculate expected posterior entropy expected_posterior_entropy = 0.0 for outcome, outcome_prob in possible_outcomes.items(): # Simulate updating beliefs with this outcome posterior = self._compute_posterior( current_beliefs, outcome, outcome_prob ) posterior_entropy = self.calculate_entropy(posterior) expected_posterior_entropy += outcome_prob * posterior_entropy # Information gain = reduction in entropy gain = max(0.0, current_entropy - expected_posterior_entropy) # Apply temporal discount based on belief freshness belief_age = self._get_belief_age(query) temporal_factor = math.exp(-self.temporal_discount * belief_age / 3600) gain *= (1 - temporal_factor * 0.5) # Old beliefs = more gain potential # Cache result self._gain_cache[cache_key] = (gain, time.time()) return gain def _compute_posterior( self, prior: Dict[str, float], evidence: str, evidence_strength: float ) -> Dict[str, float]: """Compute posterior beliefs given evidence using Bayesian update.""" posterior = {} # Simplified Bayesian update for outcome, prior_prob in prior.items(): if outcome == evidence: # Evidence supports this outcome likelihood = 0.5 + evidence_strength * 0.5 else: # Evidence against this outcome likelihood = 0.5 - evidence_strength * 0.3 posterior[outcome] = prior_prob * likelihood # Normalize total = sum(posterior.values()) if total > 0: posterior = {k: v / total for k, v in posterior.items()} return posterior def _get_belief_age(self, concept: str) -> float: """Get age of belief about a concept in seconds.""" # Check evidence history for most recent update for evidence in reversed(self.evidence_history): if evidence.get("concept") == concept: return time.time() - evidence.get("timestamp", 0) return 3600 * 24 # Default to 1 day old def update_beliefs( self, concept: str, evidence: str, strength: float, source: str = "observation" ): """ Update beliefs based on new evidence. Args: concept: Concept being updated evidence: Evidence value strength: Strength of evidence (0-1) source: Source of evidence """ with self._lock: current = self.beliefs[concept] # Update probability based on evidence if evidence == "confirmed": new_prob = current["probability"] + (1 - current["probability"]) * strength * 0.5 elif evidence == "denied": new_prob = current["probability"] * (1 - strength * 0.5) else: new_prob = current["probability"] + random.gauss(0, 0.05) current["probability"] = max(0.01, min(0.99, new_prob)) # Update certainty current["certainty"] = min(1.0, current["certainty"] + strength * 0.1) # Record evidence self.evidence_history.append({ "concept": concept, "evidence": evidence, "strength": strength, "source": source, "timestamp": time.time() }) # Trim history if len(self.evidence_history) > 10000: self.evidence_history = self.evidence_history[-5000:] def calculate_mutual_information( self, variable_a: str, variable_b: str, joint_distribution: Dict[Tuple[str, str], float] ) -> float: """ Calculate mutual information between two variables. Args: variable_a: First variable name variable_b: Second variable name joint_distribution: Joint probability distribution Returns: Mutual information in bits """ # Calculate marginals marginal_a: Dict[str, float] = defaultdict(float) marginal_b: Dict[str, float] = defaultdict(float) for (a, b), prob in joint_distribution.items(): marginal_a[a] += prob marginal_b[b] += prob # Calculate mutual information mi = 0.0 for (a, b), joint_prob in joint_distribution.items(): if joint_prob > 0: marginal_prob = marginal_a[a] * marginal_b[b] if marginal_prob > 0: mi += joint_prob * math.log2(joint_prob / marginal_prob) return max(0.0, mi) def estimate_query_value( self, query: str, context: Dict[str, Any] = None ) -> Dict[str, float]: """ Estimate comprehensive value of asking a query. Args: query: Query to evaluate context: Additional context Returns: Dictionary of value metrics """ context = context or {} # Estimate information gain possible_outcomes = {"yes": 0.5, "no": 0.3, "partial": 0.2} info_gain = self.calculate_expected_gain(query, possible_outcomes) # Estimate novelty (based on how often similar queries) novelty = self._estimate_query_novelty(query) # Estimate relevance to current goals relevance = context.get("relevance", 0.5) # Estimate answer availability availability = context.get("availability", 0.8) # Combined value combined = ( 0.4 * info_gain + 0.3 * novelty + 0.2 * relevance + 0.1 * availability ) return { "information_gain": info_gain, "novelty": novelty, "relevance": relevance, "availability": availability, "combined_value": combined } def _estimate_query_novelty(self, query: str) -> float: """Estimate how novel a query is.""" # Check similarity to past queries in evidence history similar_count = 0 query_words = set(query.lower().split()) for evidence in self.evidence_history[-100:]: concept = evidence.get("concept", "") concept_words = set(concept.lower().split()) overlap = len(query_words & concept_words) if overlap > len(query_words) * 0.5: similar_count += 1 # More similar queries = less novel if similar_count == 0: return 1.0 return 1.0 / (1.0 + similar_count) def get_statistics(self) -> Dict[str, Any]: """Get calculator statistics.""" return { "tracked_beliefs": len(self.beliefs), "evidence_count": len(self.evidence_history), "cache_size": len(self._gain_cache), "avg_certainty": ( sum(b["certainty"] for b in self.beliefs.values()) / len(self.beliefs) if self.beliefs else 0 ) } # ============================================================================== # Novelty Detector # ============================================================================== class NoveltyDetector: """ Detects novel patterns and situations by comparing against learned models. Implements multiple novelty detection strategies: - Feature-based novelty (unexpected feature combinations) - Semantic novelty (new concept relationships) - Structural novelty (new patterns) - Distribution shift detection """ def __init__( self, novelty_threshold: float = 0.5, memory_size: int = 10000, feature_decay: float = 0.001, adaptation_rate: float = 0.1 ): """ Initialize the novelty detector. Args: novelty_threshold: Threshold for flagging as novel memory_size: Size of observation memory feature_decay: Decay rate for feature importance adaptation_rate: Rate of adapting to new patterns """ self.novelty_threshold = novelty_threshold self.memory_size = memory_size self.feature_decay = feature_decay self.adaptation_rate = adaptation_rate # Feature statistics self.feature_means: Dict[str, float] = defaultdict(float) self.feature_variances: Dict[str, float] = defaultdict(lambda: 1.0) self.feature_counts: Dict[str, int] = defaultdict(int) # Pattern memory self.pattern_memory: deque = deque(maxlen=memory_size) self.pattern_frequencies: Dict[str, int] = defaultdict(int) # Concept relationships self.concept_graph: Dict[str, Set[str]] = defaultdict(set) # Novelty history self.novelty_history: List[NoveltySignal] = [] # Running statistics self._observation_count = 0 self._total_novelty = 0.0 self._lock = threading.RLock() logger.info("NoveltyDetector initialized") def detect_novelty( self, observation: Dict[str, Any], context: Optional[Dict[str, Any]] = None ) -> NoveltySignal: """ Detect novelty in an observation. Args: observation: Current observation context: Optional context information Returns: NoveltySignal with detection results """ with self._lock: self._observation_count += 1 context = context or {} # Compute novelty scores for different aspects feature_novelty = self._compute_feature_novelty(observation) semantic_novelty = self._compute_semantic_novelty(observation) structural_novelty = self._compute_structural_novelty(observation) temporal_novelty = self._compute_temporal_novelty(observation) # Determine dominant novelty type scores = { NoveltyType.FEATURE_NOVELTY: feature_novelty, NoveltyType.SEMANTIC_NOVELTY: semantic_novelty, NoveltyType.STRUCTURAL_NOVELTY: structural_novelty, NoveltyType.TEMPORAL_NOVELTY: temporal_novelty } dominant_type = max(scores, key=scores.get) dominant_score = scores[dominant_type] # Compute overall novelty and surprise overall_novelty = ( 0.35 * feature_novelty + 0.25 * semantic_novelty + 0.25 * structural_novelty + 0.15 * temporal_novelty ) # Surprise is how much novelty exceeds threshold surprise = max(0, overall_novelty - self.novelty_threshold) * 2 # Create signal signal = NoveltySignal( novelty_type=dominant_type, source="observation", intensity=overall_novelty, surprise=min(1.0, surprise), features={ "feature": feature_novelty, "semantic": semantic_novelty, "structural": structural_novelty, "temporal": temporal_novelty }, context=context ) # Update statistics self._update_statistics(observation, overall_novelty) # Record if above threshold if overall_novelty >= self.novelty_threshold: self.novelty_history.append(signal) if len(self.novelty_history) > 1000: self.novelty_history = self.novelty_history[-500:] return signal def _compute_feature_novelty( self, observation: Dict[str, Any] ) -> float: """Compute novelty based on feature values.""" if not observation: return 0.0 novelty_scores = [] for feature, value in observation.items(): if isinstance(value, (int, float)): # Numerical feature - check against statistics mean = self.feature_means.get(feature, 0) var = max(self.feature_variances.get(feature, 1), 0.01) # Z-score z_score = abs(value - mean) / math.sqrt(var) # Convert to novelty (sigmoid) novelty = 1.0 / (1.0 + math.exp(-z_score + 2)) novelty_scores.append(novelty) elif isinstance(value, str): # Categorical feature - check frequency key = f"{feature}:{value}" count = self.feature_counts.get(key, 0) total = self.feature_counts.get(f"_total_{feature}", 1) # Rare values are novel frequency = count / max(total, 1) novelty = 1.0 - frequency novelty_scores.append(novelty) return sum(novelty_scores) / len(novelty_scores) if novelty_scores else 0.0 def _compute_semantic_novelty( self, observation: Dict[str, Any] ) -> float: """Compute novelty based on semantic relationships.""" concepts = observation.get("concepts", []) if not concepts or len(concepts) < 2: return 0.5 # Default moderate novelty # Check for novel concept combinations novel_pairs = 0 total_pairs = 0 for i, concept_a in enumerate(concepts): for concept_b in concepts[i + 1:]: total_pairs += 1 # Check if this pair has been seen if concept_b not in self.concept_graph.get(concept_a, set()): novel_pairs += 1 if total_pairs == 0: return 0.5 return novel_pairs / total_pairs def _compute_structural_novelty( self, observation: Dict[str, Any] ) -> float: """Compute novelty based on structural patterns.""" # Create pattern signature pattern = self._create_pattern_signature(observation) # Check frequency of this pattern frequency = self.pattern_frequencies.get(pattern, 0) total = len(self.pattern_memory) or 1 # Rare patterns are novel if frequency == 0: return 1.0 return 1.0 - (frequency / total) def _compute_temporal_novelty( self, observation: Dict[str, Any] ) -> float: """Compute novelty based on temporal patterns.""" if len(self.pattern_memory) < 10: return 0.5 # Not enough history # Get recent pattern current_pattern = self._create_pattern_signature(observation) # Check if pattern matches expected temporal sequence recent_patterns = list(self.pattern_memory)[-10:] # Simple: check if current pattern follows recent ones transitions = defaultdict(lambda: defaultdict(int)) for i in range(len(recent_patterns) - 1): transitions[recent_patterns[i]][recent_patterns[i + 1]] += 1 # Check expected transition if recent_patterns: last_pattern = recent_patterns[-1] expected = transitions.get(last_pattern, {}) if current_pattern in expected: # Expected transition total_transitions = sum(expected.values()) probability = expected[current_pattern] / total_transitions return 1.0 - probability else: # Unexpected transition return 0.9 return 0.5 def _create_pattern_signature(self, observation: Dict[str, Any]) -> str: """Create a hashable pattern signature.""" # Sort keys and create deterministic string items = sorted( [(k, str(v)[:50]) for k, v in observation.items() if k not in ["timestamp", "id"]] ) pattern_str = "|".join(f"{k}:{v}" for k, v in items) return hashlib.md5(pattern_str.encode()).hexdigest()[:16] def _update_statistics( self, observation: Dict[str, Any], novelty: float ): """Update internal statistics with new observation.""" # Update feature statistics for feature, value in observation.items(): if isinstance(value, (int, float)): # Running mean and variance count = self.feature_counts.get(feature, 0) + 1 self.feature_counts[feature] = count old_mean = self.feature_means.get(feature, 0) delta = value - old_mean new_mean = old_mean + delta / count self.feature_means[feature] = new_mean # Welford's algorithm for variance old_var = self.feature_variances.get(feature, 0) new_var = old_var + delta * (value - new_mean) - old_var / count self.feature_variances[feature] = max(new_var, 0.01) elif isinstance(value, str): key = f"{feature}:{value}" self.feature_counts[key] = self.feature_counts.get(key, 0) + 1 self.feature_counts[f"_total_{feature}"] = \ self.feature_counts.get(f"_total_{feature}", 0) + 1 # Update concept graph concepts = observation.get("concepts", []) for i, concept_a in enumerate(concepts): for concept_b in concepts[i + 1:]: self.concept_graph[concept_a].add(concept_b) self.concept_graph[concept_b].add(concept_a) # Update pattern memory pattern = self._create_pattern_signature(observation) self.pattern_memory.append(pattern) self.pattern_frequencies[pattern] = \ self.pattern_frequencies.get(pattern, 0) + 1 # Track total novelty self._total_novelty += novelty def adapt_to_environment(self, rate: Optional[float] = None): """ Adapt novelty thresholds to current environment. Args: rate: Adaptation rate (uses default if None) """ rate = rate or self.adaptation_rate with self._lock: if self._observation_count < 100: return # Need more data # Adjust threshold based on recent novelty rates avg_novelty = self._total_novelty / self._observation_count # Move threshold toward average self.novelty_threshold = ( (1 - rate) * self.novelty_threshold + rate * (avg_novelty + 0.1) # Slightly above average ) logger.info(f"Adapted novelty threshold to {self.novelty_threshold:.3f}") def get_novelty_summary(self) -> Dict[str, Any]: """Get summary of novelty detection.""" return { "observations_processed": self._observation_count, "current_threshold": self.novelty_threshold, "avg_novelty": ( self._total_novelty / self._observation_count if self._observation_count > 0 else 0 ), "novelty_events": len(self.novelty_history), "unique_patterns": len(self.pattern_frequencies), "tracked_features": len(self.feature_means), "concept_connections": sum( len(v) for v in self.concept_graph.values() ) // 2 } # ============================================================================== # Exploration Policy # ============================================================================== class ExplorationPolicy: """ Balances exploration vs exploitation in learning. Implements multiple exploration strategies: - Epsilon-greedy with decay - Upper Confidence Bound (UCB) - Thompson Sampling - Curiosity-driven exploration - Hybrid approaches """ def __init__( self, mode: ExplorationMode = ExplorationMode.HYBRID, initial_epsilon: float = 1.0, epsilon_decay: float = 0.995, min_epsilon: float = 0.1, ucb_exploration_factor: float = 2.0, curiosity_weight: float = 0.3 ): """ Initialize exploration policy. Args: mode: Exploration strategy mode initial_epsilon: Starting epsilon for epsilon-greedy epsilon_decay: Decay rate for epsilon min_epsilon: Minimum epsilon value ucb_exploration_factor: Exploration factor for UCB curiosity_weight: Weight for curiosity-driven exploration """ self.mode = mode self.epsilon = initial_epsilon self.epsilon_decay = epsilon_decay self.min_epsilon = min_epsilon self.ucb_factor = ucb_exploration_factor self.curiosity_weight = curiosity_weight # Action statistics for UCB/Thompson self.action_counts: Dict[str, int] = defaultdict(int) self.action_rewards: Dict[str, float] = defaultdict(float) self.action_successes: Dict[str, int] = defaultdict(int) # Curiosity tracking self.novelty_cache: Dict[str, float] = {} # Exploration history self.exploration_history: List[Dict[str, Any]] = [] self.total_actions = 0 self.exploration_actions = 0 self._lock = threading.RLock() logger.info(f"ExplorationPolicy initialized with mode: {mode.name}") def should_explore(self) -> bool: """ Determine whether to explore vs exploit. Returns: True if should explore, False if should exploit """ with self._lock: if self.mode == ExplorationMode.RANDOM: return random.random() < 0.5 elif self.mode == ExplorationMode.EPSILON_GREEDY: return random.random() < self.epsilon elif self.mode == ExplorationMode.CURIOSITY_DRIVEN: # Explore if recent novelty is low if self.exploration_history: recent = self.exploration_history[-10:] avg_novelty = sum(e.get("novelty", 0.5) for e in recent) / len(recent) return avg_novelty < 0.3 # Explore when not finding novel things return True elif self.mode == ExplorationMode.HYBRID: # Combine epsilon-greedy with curiosity epsilon_explore = random.random() < self.epsilon if self.exploration_history: recent = self.exploration_history[-10:] avg_novelty = sum(e.get("novelty", 0.5) for e in recent) / len(recent) curiosity_explore = avg_novelty < 0.3 else: curiosity_explore = True return epsilon_explore or (curiosity_explore and random.random() < 0.5) return random.random() < self.epsilon def select_action( self, available_actions: List[str], action_values: Optional[Dict[str, float]] = None, novelty_scores: Optional[Dict[str, float]] = None ) -> str: """ Select an action based on exploration policy. Args: available_actions: List of possible actions action_values: Known value estimates for actions novelty_scores: Novelty scores for actions Returns: Selected action """ if not available_actions: return "" with self._lock: self.total_actions += 1 action_values = action_values or {} novelty_scores = novelty_scores or {} if self.mode == ExplorationMode.RANDOM: selected = random.choice(available_actions) self.exploration_actions += 1 elif self.mode == ExplorationMode.EPSILON_GREEDY: if random.random() < self.epsilon: selected = random.choice(available_actions) self.exploration_actions += 1 else: # Exploit: choose best known action selected = max( available_actions, key=lambda a: action_values.get(a, 0) ) elif self.mode == ExplorationMode.UCB: selected = self._ucb_selection( available_actions, action_values ) elif self.mode == ExplorationMode.THOMPSON_SAMPLING: selected = self._thompson_selection(available_actions) elif self.mode == ExplorationMode.CURIOSITY_DRIVEN: selected = self._curiosity_selection( available_actions, action_values, novelty_scores ) elif self.mode == ExplorationMode.HYBRID: selected = self._hybrid_selection( available_actions, action_values, novelty_scores ) else: selected = random.choice(available_actions) # Update action count self.action_counts[selected] += 1 return selected def _ucb_selection( self, actions: List[str], values: Dict[str, float] ) -> str: """Select using Upper Confidence Bound.""" if self.total_actions == 0: return random.choice(actions) ucb_values = {} for action in actions: count = self.action_counts.get(action, 0) value = values.get(action, 0) if count == 0: # Untried action gets infinite UCB ucb_values[action] = float('inf') else: # UCB formula exploration_bonus = self.ucb_factor * math.sqrt( math.log(self.total_actions) / count ) ucb_values[action] = value + exploration_bonus return max(actions, key=lambda a: ucb_values[a]) def _thompson_selection(self, actions: List[str]) -> str: """Select using Thompson Sampling.""" sampled_values = {} for action in actions: # Use Beta distribution for binary rewards successes = self.action_successes.get(action, 1) # Prior of 1 failures = ( self.action_counts.get(action, 1) - successes + 1 ) # Prior of 1 # Sample from Beta distribution sampled_values[action] = random.betavariate( max(1, successes), max(1, failures) ) return max(actions, key=lambda a: sampled_values[a]) def _curiosity_selection( self, actions: List[str], values: Dict[str, float], novelty_scores: Dict[str, float] ) -> str: """Select based on curiosity (novelty).""" curiosity_values = {} for action in actions: # Combine value with novelty value = values.get(action, 0.5) novelty = novelty_scores.get(action, 0.5) # Curiosity prefers novel actions curiosity_values[action] = ( (1 - self.curiosity_weight) * value + self.curiosity_weight * novelty ) return max(actions, key=lambda a: curiosity_values[a]) def _hybrid_selection( self, actions: List[str], values: Dict[str, float], novelty_scores: Dict[str, float] ) -> str: """Hybrid selection combining multiple strategies.""" if random.random() < self.epsilon: # Exploration: use curiosity or random if novelty_scores: return self._curiosity_selection(actions, values, novelty_scores) return random.choice(actions) else: # Exploitation: use UCB with novelty bonus combined_values = {} for action in actions: value = values.get(action, 0) novelty = novelty_scores.get(action, 0.5) count = self.action_counts.get(action, 0) # UCB-style bonus if count > 0 and self.total_actions > 0: ucb_bonus = self.ucb_factor * math.sqrt( math.log(self.total_actions) / count ) else: ucb_bonus = 1.0 combined_values[action] = ( 0.5 * value + 0.3 * ucb_bonus + 0.2 * novelty ) return max(actions, key=lambda a: combined_values[a]) def update_with_result( self, action: str, reward: float, success: bool, novelty: float = 0.5 ): """ Update policy with action result. Args: action: Action taken reward: Reward received success: Whether action succeeded novelty: Novelty encountered """ with self._lock: # Update action statistics self.action_rewards[action] += reward if success: self.action_successes[action] += 1 # Update novelty cache old_novelty = self.novelty_cache.get(action, novelty) self.novelty_cache[action] = 0.9 * old_novelty + 0.1 * novelty # Record in history self.exploration_history.append({ "action": action, "reward": reward, "success": success, "novelty": novelty, "epsilon": self.epsilon, "timestamp": time.time() }) # Trim history if len(self.exploration_history) > 10000: self.exploration_history = self.exploration_history[-5000:] # Decay epsilon self.epsilon = max( self.min_epsilon, self.epsilon * self.epsilon_decay ) def get_exploration_rate(self) -> float: """Get current exploration rate.""" if self.total_actions == 0: return 1.0 return self.exploration_actions / self.total_actions def get_statistics(self) -> Dict[str, Any]: """Get policy statistics.""" return { "mode": self.mode.name, "epsilon": self.epsilon, "total_actions": self.total_actions, "exploration_actions": self.exploration_actions, "exploration_rate": self.get_exploration_rate(), "unique_actions": len(self.action_counts), "avg_reward": ( sum(self.action_rewards.values()) / self.total_actions if self.total_actions > 0 else 0 ) } # ============================================================================== # Question Generator # ============================================================================== class QuestionGenerator: """ Generates curious questions to fill knowledge gaps. Creates questions that: - Target high-value information gaps - Are diverse in type and topic - Are answerable (avoid unanswerable questions) - Lead to actionable learning """ def __init__( self, max_questions: int = 100, diversity_penalty: float = 0.3, relevance_threshold: float = 0.3, freshness_weight: float = 0.2 ): """ Initialize question generator. Args: max_questions: Maximum pending questions diversity_penalty: Penalty for similar questions relevance_threshold: Minimum relevance to generate freshness_weight: Weight for question freshness """ self.max_questions = max_questions self.diversity_penalty = diversity_penalty self.relevance_threshold = relevance_threshold self.freshness_weight = freshness_weight # Question queue (priority queue) self.question_queue: List[CuriousQuestion] = [] # Question templates by type self.templates: Dict[QuestionType, List[str]] = { QuestionType.FACTUAL: [ "What is {concept}?", "What are the key characteristics of {concept}?", "What defines {concept}?", "What are examples of {concept}?" ], QuestionType.CAUSAL: [ "Why does {concept} happen?", "What causes {concept}?", "What leads to {concept}?", "What are the reasons for {concept}?" ], QuestionType.PROCEDURAL: [ "How do you {concept}?", "What are the steps to {concept}?", "How can I achieve {concept}?", "What is the process for {concept}?" ], QuestionType.COMPARATIVE: [ "How does {concept_a} compare to {concept_b}?", "What is the difference between {concept_a} and {concept_b}?", "When should you use {concept_a} vs {concept_b}?", "What are the trade-offs between {concept_a} and {concept_b}?" ], QuestionType.COUNTERFACTUAL: [ "What would happen if {concept} were different?", "What if {concept} didn't exist?", "How would things change without {concept}?", "What alternatives exist to {concept}?" ], QuestionType.EXPLORATORY: [ "What other things are related to {concept}?", "What else should I know about {concept}?", "What are the variations of {concept}?", "What domains use {concept}?" ], QuestionType.PREDICTIVE: [ "What will happen if {concept} continues?", "What are the implications of {concept}?", "What does {concept} lead to?", "How will {concept} evolve?" ], QuestionType.DEFINITIONAL: [ "What is the precise definition of {concept}?", "How is {concept} formally defined?", "What are the boundaries of {concept}?", "What is and isn't {concept}?" ] } # Question history self.generated_questions: List[CuriousQuestion] = [] self.answered_questions: List[CuriousQuestion] = [] # Topic tracking for diversity self.recent_topics: deque = deque(maxlen=50) self._lock = threading.RLock() logger.info("QuestionGenerator initialized") def generate_question( self, concept: str, question_type: Optional[QuestionType] = None, context: Optional[Dict[str, Any]] = None, second_concept: Optional[str] = None ) -> CuriousQuestion: """ Generate a single curious question. Args: concept: Primary concept to ask about question_type: Type of question to generate context: Additional context second_concept: Second concept for comparative questions Returns: Generated question """ with self._lock: context = context or {} # Select question type if not specified if question_type is None: question_type = self._select_question_type(concept, context) # Get template templates = self.templates.get(question_type, self.templates[QuestionType.FACTUAL]) template = random.choice(templates) # Fill template if "{concept_a}" in template and "{concept_b}" in template: # Comparative question if second_concept: question_text = template.format( concept_a=concept, concept_b=second_concept ) else: question_text = template.format( concept_a=concept, concept_b="alternatives" ) else: question_text = template.format(concept=concept) # Calculate priority information_gain = context.get("information_gain", 0.5) relevance = context.get("relevance", 0.5) diversity = self._calculate_diversity(concept) priority = ( 0.4 * information_gain + 0.3 * relevance + 0.2 * diversity + 0.1 * random.random() # Small random factor ) question = CuriousQuestion( question_text=question_text, question_type=question_type, target_concept=concept, information_gain=information_gain, priority=priority, metadata=context ) # Track topic self.recent_topics.append(concept) return question def _select_question_type( self, concept: str, context: Dict[str, Any] ) -> QuestionType: """Select appropriate question type based on context.""" # Check what we know about the concept completeness = context.get("completeness", 0.5) certainty = context.get("certainty", 0.5) if completeness < 0.3: # Know very little - start with factual return random.choice([ QuestionType.FACTUAL, QuestionType.DEFINITIONAL ]) elif completeness < 0.6: # Know something - go deeper return random.choice([ QuestionType.CAUSAL, QuestionType.PROCEDURAL, QuestionType.EXPLORATORY ]) else: # Know a lot - explore edges return random.choice([ QuestionType.COMPARATIVE, QuestionType.COUNTERFACTUAL, QuestionType.PREDICTIVE ]) def _calculate_diversity(self, concept: str) -> float: """Calculate diversity score for a concept.""" if not self.recent_topics: return 1.0 # Count recent occurrences occurrences = sum(1 for t in self.recent_topics if t == concept) # Diversity decreases with repetition diversity = 1.0 / (1.0 + occurrences * self.diversity_penalty) return diversity def generate_questions_for_gap( self, gap: KnowledgeGap, count: int = 3 ) -> List[CuriousQuestion]: """ Generate questions to fill a knowledge gap. Args: gap: Knowledge gap to address count: Number of questions to generate Returns: List of generated questions """ questions = [] # Determine appropriate question types based on gap type if gap.gap_type == "missing": types = [QuestionType.FACTUAL, QuestionType.DEFINITIONAL] elif gap.gap_type == "incomplete": types = [QuestionType.PROCEDURAL, QuestionType.EXPLORATORY] elif gap.gap_type == "outdated": types = [QuestionType.PREDICTIVE, QuestionType.FACTUAL] elif gap.gap_type == "uncertain": types = [QuestionType.CAUSAL, QuestionType.COMPARATIVE] else: types = list(QuestionType) context = { "information_gain": gap.potential_gain, "relevance": gap.severity, "completeness": 0.3 if gap.gap_type == "missing" else 0.6 } for i in range(count): q_type = types[i % len(types)] question = self.generate_question( gap.concept, question_type=q_type, context=context ) questions.append(question) return questions def add_to_queue(self, question: CuriousQuestion): """Add question to priority queue.""" with self._lock: if len(self.question_queue) >= self.max_questions: # Remove lowest priority heapq.heappop(self.question_queue) heapq.heappush(self.question_queue, question) self.generated_questions.append(question) def get_next_question(self) -> Optional[CuriousQuestion]: """Get next highest priority question.""" with self._lock: if not self.question_queue: return None question = heapq.heappop(self.question_queue) question.asked = True return question def answer_question( self, question: CuriousQuestion, answer: str, quality: float = 1.0 ): """Record answer to a question.""" with self._lock: question.answered = True question.answer = answer question.metadata["answer_quality"] = quality question.metadata["answered_at"] = time.time() self.answered_questions.append(question) def get_statistics(self) -> Dict[str, Any]: """Get generator statistics.""" type_distribution = defaultdict(int) for q in self.generated_questions: type_distribution[q.question_type.name] += 1 return { "total_generated": len(self.generated_questions), "total_answered": len(self.answered_questions), "queue_size": len(self.question_queue), "answer_rate": ( len(self.answered_questions) / len(self.generated_questions) if self.generated_questions else 0 ), "type_distribution": dict(type_distribution), "recent_topics": len(set(self.recent_topics)) } # ============================================================================== # Knowledge Gap Finder # ============================================================================== class KnowledgeGapFinder: """ Identifies gaps in the knowledge base. Detects: - Missing concepts - Incomplete knowledge - Outdated information - Uncertain beliefs - Disconnected knowledge islands """ def __init__( self, completeness_threshold: float = 0.7, certainty_threshold: float = 0.7, freshness_threshold: float = 0.5, max_gaps: int = 100 ): """ Initialize knowledge gap finder. Args: completeness_threshold: Below this = incomplete certainty_threshold: Below this = uncertain freshness_threshold: Below this = outdated max_gaps: Maximum gaps to track """ self.completeness_threshold = completeness_threshold self.certainty_threshold = certainty_threshold self.freshness_threshold = freshness_threshold self.max_gaps = max_gaps # Known concepts and their states self.concepts: Dict[str, InformationState] = {} # Identified gaps self.gaps: Dict[str, KnowledgeGap] = {} # Gap history self.filled_gaps: List[KnowledgeGap] = [] # Expected concepts (things we should know) self.expected_concepts: Set[str] = set() # Concept dependencies self.dependencies: Dict[str, Set[str]] = defaultdict(set) self._lock = threading.RLock() logger.info("KnowledgeGapFinder initialized") def register_concept( self, concept_id: str, name: str, initial_certainty: float = 0.0, initial_completeness: float = 0.0, related_concepts: Optional[Set[str]] = None ): """ Register a concept in the knowledge base. Args: concept_id: Unique concept identifier name: Human-readable name initial_certainty: Initial certainty level initial_completeness: Initial completeness level related_concepts: Related concept IDs """ with self._lock: self.concepts[concept_id] = InformationState( concept_id=concept_id, name=name, certainty=initial_certainty, completeness=initial_completeness, related_concepts=related_concepts or set() ) # Add to dependencies if related_concepts: for related in related_concepts: self.dependencies[concept_id].add(related) self.dependencies[related].add(concept_id) def update_concept( self, concept_id: str, certainty_delta: float = 0.0, completeness_delta: float = 0.0, refresh: bool = False ): """ Update concept knowledge state. Args: concept_id: Concept to update certainty_delta: Change in certainty completeness_delta: Change in completeness refresh: Whether to refresh freshness """ with self._lock: if concept_id not in self.concepts: # Auto-register self.register_concept(concept_id, concept_id) concept = self.concepts[concept_id] concept.certainty = max(0, min(1, concept.certainty + certainty_delta)) concept.completeness = max(0, min(1, concept.completeness + completeness_delta)) if refresh: concept.freshness = 1.0 concept.last_accessed = time.time() concept.access_count += 1 def expect_concept(self, concept_id: str, required_by: Optional[str] = None): """ Mark a concept as expected (should be known). Args: concept_id: Concept that should be known required_by: Concept that requires this one """ with self._lock: self.expected_concepts.add(concept_id) if required_by: self.dependencies[required_by].add(concept_id) def find_gaps(self) -> List[KnowledgeGap]: """ Find all current knowledge gaps. Returns: List of identified gaps sorted by severity """ with self._lock: new_gaps = [] # Check for missing concepts for concept_id in self.expected_concepts: if concept_id not in self.concepts: gap = self._create_missing_gap(concept_id) new_gaps.append(gap) continue # Check existing concepts concept = self.concepts[concept_id] # Incomplete knowledge if concept.completeness < self.completeness_threshold: gap = self._create_incomplete_gap(concept) new_gaps.append(gap) # Uncertain knowledge elif concept.certainty < self.certainty_threshold: gap = self._create_uncertain_gap(concept) new_gaps.append(gap) # Outdated knowledge concept.decay_freshness() if concept.freshness < self.freshness_threshold: gap = self._create_outdated_gap(concept) new_gaps.append(gap) # Check for disconnected knowledge disconnected = self._find_disconnected_concepts() for concept_id in disconnected: if concept_id in self.concepts: gap = self._create_disconnected_gap( self.concepts[concept_id] ) new_gaps.append(gap) # Update gap registry for gap in new_gaps: if gap.gap_id not in self.gaps: self.gaps[gap.gap_id] = gap # Limit gaps if len(self.gaps) > self.max_gaps: sorted_gaps = sorted( self.gaps.values(), key=lambda g: g.severity, reverse=True ) self.gaps = {g.gap_id: g for g in sorted_gaps[:self.max_gaps]} # Return sorted by severity return sorted( self.gaps.values(), key=lambda g: g.severity, reverse=True ) def _create_missing_gap(self, concept_id: str) -> KnowledgeGap: """Create gap for missing concept.""" # Estimate severity based on dependencies dependents = sum( 1 for deps in self.dependencies.values() if concept_id in deps ) severity = min(1.0, 0.5 + 0.1 * dependents) return KnowledgeGap( gap_id=f"missing_{concept_id}", concept=concept_id, gap_type="missing", severity=severity, potential_gain=severity * 0.8, fill_difficulty=0.5 ) def _create_incomplete_gap(self, concept: InformationState) -> KnowledgeGap: """Create gap for incomplete knowledge.""" severity = (self.completeness_threshold - concept.completeness) * 0.8 return KnowledgeGap( gap_id=f"incomplete_{concept.concept_id}", concept=concept.concept_id, gap_type="incomplete", severity=severity, potential_gain=severity * 0.7, fill_difficulty=0.6 ) def _create_uncertain_gap(self, concept: InformationState) -> KnowledgeGap: """Create gap for uncertain knowledge.""" severity = (self.certainty_threshold - concept.certainty) * 0.7 return KnowledgeGap( gap_id=f"uncertain_{concept.concept_id}", concept=concept.concept_id, gap_type="uncertain", severity=severity, potential_gain=severity * 0.6, fill_difficulty=0.4 ) def _create_outdated_gap(self, concept: InformationState) -> KnowledgeGap: """Create gap for outdated knowledge.""" severity = (self.freshness_threshold - concept.freshness) * 0.6 return KnowledgeGap( gap_id=f"outdated_{concept.concept_id}", concept=concept.concept_id, gap_type="outdated", severity=severity, potential_gain=severity * 0.5, fill_difficulty=0.3 ) def _create_disconnected_gap(self, concept: InformationState) -> KnowledgeGap: """Create gap for disconnected knowledge.""" return KnowledgeGap( gap_id=f"disconnected_{concept.concept_id}", concept=concept.concept_id, gap_type="disconnected", severity=0.4, potential_gain=0.3, fill_difficulty=0.5 ) def _find_disconnected_concepts(self) -> Set[str]: """Find concepts with few connections.""" disconnected = set() for concept_id, concept in self.concepts.items(): # Check connection count connections = len(concept.related_concepts) connections += len(self.dependencies.get(concept_id, set())) # Low connections = disconnected if connections < 2 and concept.access_count > 3: disconnected.add(concept_id) return disconnected def fill_gap(self, gap_id: str): """Mark a gap as filled.""" with self._lock: if gap_id in self.gaps: gap = self.gaps.pop(gap_id) gap.filled_at = time.time() self.filled_gaps.append(gap) # Update concept if exists concept_id = gap.concept if concept_id in self.concepts: concept = self.concepts[concept_id] if gap.gap_type == "incomplete": concept.completeness = min(1.0, concept.completeness + 0.2) elif gap.gap_type == "uncertain": concept.certainty = min(1.0, concept.certainty + 0.2) elif gap.gap_type == "outdated": concept.freshness = 1.0 concept.last_accessed = time.time() def get_priority_gaps(self, count: int = 10) -> List[KnowledgeGap]: """Get highest priority gaps.""" gaps = self.find_gaps() return gaps[:count] def get_statistics(self) -> Dict[str, Any]: """Get gap finder statistics.""" gap_types = defaultdict(int) for gap in self.gaps.values(): gap_types[gap.gap_type] += 1 return { "tracked_concepts": len(self.concepts), "expected_concepts": len(self.expected_concepts), "current_gaps": len(self.gaps), "filled_gaps": len(self.filled_gaps), "gap_types": dict(gap_types), "avg_completeness": ( sum(c.completeness for c in self.concepts.values()) / len(self.concepts) if self.concepts else 0 ), "avg_certainty": ( sum(c.certainty for c in self.concepts.values()) / len(self.concepts) if self.concepts else 0 ) } # ============================================================================== # Interest Model # ============================================================================== class InterestModel: """ Models and tracks interests over time. Implements: - Interest decay and growth - Engagement tracking - Interest discovery - Satisfaction modeling """ def __init__( self, base_decay_rate: float = 0.01, engagement_boost: float = 0.1, discovery_boost: float = 0.2, satisfaction_threshold: float = 0.5, max_interests: int = 50 ): """ Initialize interest model. Args: base_decay_rate: Rate of interest decay engagement_boost: Boost from engagement discovery_boost: Boost from discoveries satisfaction_threshold: Threshold for satisfying interest max_interests: Maximum tracked interests """ self.base_decay_rate = base_decay_rate self.engagement_boost = engagement_boost self.discovery_boost = discovery_boost self.satisfaction_threshold = satisfaction_threshold self.max_interests = max_interests # Interest profiles self.interests: Dict[str, InterestProfile] = {} # Interest history self.engagement_history: List[Dict[str, Any]] = [] # Category weights self.category_weights: Dict[InterestCategory, float] = { cat: 1.0 for cat in InterestCategory } # Global interest state self.total_engagement_time = 0.0 self.total_discoveries = 0 self.avg_satisfaction = 0.5 self._lock = threading.RLock() logger.info("InterestModel initialized") def create_interest( self, name: str, category: InterestCategory, base_interest: float = 0.5 ) -> InterestProfile: """ Create a new interest profile. Args: name: Interest name category: Interest category base_interest: Base interest level Returns: Created interest profile """ with self._lock: interest_id = f"{category.name}_{name}_{len(self.interests)}" profile = InterestProfile( interest_id=interest_id, category=category, name=name, base_interest=base_interest, current_interest=base_interest ) self.interests[interest_id] = profile # Prune if too many if len(self.interests) > self.max_interests: self._prune_interests() return profile def record_engagement( self, interest_id: str, duration_seconds: float, discoveries: int = 0, satisfaction: float = 0.5 ): """ Record engagement with an interest. Args: interest_id: Interest engaged with duration_seconds: Time spent discoveries: Number of discoveries made satisfaction: Satisfaction level """ with self._lock: if interest_id not in self.interests: return profile = self.interests[interest_id] # Update profile profile.total_time_spent += duration_seconds profile.discovery_count += discoveries profile.engagement_history.append(satisfaction) profile.last_engaged = time.time() # Update satisfaction (running average) n = len(profile.engagement_history) profile.satisfaction_score = ( (profile.satisfaction_score * (n - 1) + satisfaction) / n ) # Boost interest based on engagement interest_boost = self.engagement_boost * (duration_seconds / 60) discovery_bonus = self.discovery_boost * discoveries satisfaction_bonus = 0.1 if satisfaction > self.satisfaction_threshold else -0.05 profile.current_interest = min(1.0, max(0.0, profile.current_interest + interest_boost + discovery_bonus + satisfaction_bonus )) # Record in history self.engagement_history.append({ "interest_id": interest_id, "duration": duration_seconds, "discoveries": discoveries, "satisfaction": satisfaction, "timestamp": time.time() }) # Update global stats self.total_engagement_time += duration_seconds self.total_discoveries += discoveries self.avg_satisfaction = (self.avg_satisfaction * 0.9 + satisfaction * 0.1) # Trim history if len(self.engagement_history) > 10000: self.engagement_history = self.engagement_history[-5000:] def decay_interests(self): """Apply time-based decay to all interests.""" with self._lock: current_time = time.time() for profile in self.interests.values(): # Time since last engagement time_delta = current_time - profile.last_engaged hours_since = time_delta / 3600 # Decay toward base interest decay = self.base_decay_rate * hours_since target = profile.base_interest if profile.current_interest > target: profile.current_interest = max( target, profile.current_interest - decay ) else: profile.current_interest = min( target, profile.current_interest + decay * 0.5 ) def get_top_interests(self, count: int = 5) -> List[InterestProfile]: """Get top interests by current level.""" with self._lock: sorted_interests = sorted( self.interests.values(), key=lambda i: i.current_interest, reverse=True ) return sorted_interests[:count] def get_suggested_interest( self, exclude: Optional[Set[str]] = None ) -> Optional[InterestProfile]: """ Suggest an interest to pursue. Args: exclude: Interest IDs to exclude Returns: Suggested interest profile """ with self._lock: exclude = exclude or set() candidates = [ i for i in self.interests.values() if i.interest_id not in exclude ] if not candidates: return None # Score by interest level and category weight scored = [] for interest in candidates: category_weight = self.category_weights.get(interest.category, 1.0) # Combine factors score = ( 0.5 * interest.current_interest + 0.2 * category_weight + 0.2 * interest.satisfaction_score + 0.1 * random.random() # Some randomness ) scored.append((interest, score)) # Select probabilistically from top candidates scored.sort(key=lambda x: x[1], reverse=True) top = scored[:5] weights = [s for _, s in top] total = sum(weights) if total == 0: return top[0][0] r = random.random() * total cumsum = 0 for interest, score in top: cumsum += score if r <= cumsum: return interest return top[0][0] def _prune_interests(self): """Remove least active interests.""" if len(self.interests) <= self.max_interests: return # Sort by activity (last engaged * current interest) sorted_interests = sorted( self.interests.items(), key=lambda x: x[1].last_engaged * x[1].current_interest, reverse=True ) # Keep top N self.interests = dict(sorted_interests[:self.max_interests]) def update_category_weight( self, category: InterestCategory, delta: float ): """Update weight for a category.""" with self._lock: current = self.category_weights.get(category, 1.0) self.category_weights[category] = max(0.1, min(2.0, current + delta)) def get_statistics(self) -> Dict[str, Any]: """Get interest model statistics.""" with self._lock: category_distribution = defaultdict(int) for interest in self.interests.values(): category_distribution[interest.category.name] += 1 return { "total_interests": len(self.interests), "avg_current_interest": ( sum(i.current_interest for i in self.interests.values()) / len(self.interests) if self.interests else 0 ), "total_engagement_time": self.total_engagement_time, "total_discoveries": self.total_discoveries, "avg_satisfaction": self.avg_satisfaction, "category_distribution": dict(category_distribution), "engagement_count": len(self.engagement_history) } # ============================================================================== # Curiosity Engine (Main Orchestrator) # ============================================================================== class CuriosityEngine: """ Main curiosity-driven learning engine for AIVA Queen. Orchestrates all curiosity components: - Information gain calculation - Novelty detection - Exploration policy - Question generation - Knowledge gap finding - Interest modeling """ def __init__( self, exploration_mode: ExplorationMode = ExplorationMode.HYBRID, novelty_threshold: float = 0.5, max_questions: int = 100, max_interests: int = 50, storage_path: Optional[str] = None ): """ Initialize the curiosity engine. Args: exploration_mode: Mode for exploration policy novelty_threshold: Threshold for novelty detection max_questions: Maximum pending questions max_interests: Maximum tracked interests storage_path: Path for persistent storage """ self.storage_path = Path(storage_path) if storage_path else None if self.storage_path: self.storage_path.mkdir(parents=True, exist_ok=True) # Initialize components self.info_gain_calculator = InformationGainCalculator() self.novelty_detector = NoveltyDetector( novelty_threshold=novelty_threshold ) self.exploration_policy = ExplorationPolicy(mode=exploration_mode) self.question_generator = QuestionGenerator(max_questions=max_questions) self.gap_finder = KnowledgeGapFinder() self.interest_model = InterestModel(max_interests=max_interests) # Global state self.current_focus: Optional[str] = None self.exploration_count = 0 self.discovery_count = 0 self.total_information_gained = 0.0 # Event log self.event_log: List[Dict[str, Any]] = [] self._lock = threading.RLock() logger.info(f"CuriosityEngine initialized with mode: {exploration_mode.name}") def observe( self, observation: Dict[str, Any], context: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Process an observation through the curiosity engine. Args: observation: Observation to process context: Additional context Returns: Processing results including novelty and suggested actions """ with self._lock: context = context or {} # Detect novelty novelty_signal = self.novelty_detector.detect_novelty( observation, context ) # Update information gain calculator for key, value in observation.items(): if isinstance(value, str): self.info_gain_calculator.update_beliefs( key, "observed", 0.7, "observation" ) # Register any concepts concepts = observation.get("concepts", []) for concept in concepts: if concept not in self.gap_finder.concepts: self.gap_finder.register_concept(concept, concept) self.gap_finder.update_concept( concept, certainty_delta=0.05, completeness_delta=0.05, refresh=True ) # Generate questions if novel questions = [] if novelty_signal.intensity >= self.novelty_detector.novelty_threshold: for concept in concepts[:3]: q = self.question_generator.generate_question( concept, context={"information_gain": novelty_signal.intensity} ) questions.append(q) self.question_generator.add_to_queue(q) # Log event self._log_event("observation", { "novelty": novelty_signal.intensity, "surprise": novelty_signal.surprise, "questions_generated": len(questions) }) return { "novelty_signal": novelty_signal, "questions_generated": questions, "should_explore": novelty_signal.intensity < 0.3 } def decide_action( self, available_actions: List[str], action_values: Optional[Dict[str, float]] = None, context: Optional[Dict[str, Any]] = None ) -> Tuple[str, Dict[str, Any]]: """ Decide next action based on curiosity. Args: available_actions: Available actions to choose from action_values: Known value estimates context: Additional context Returns: Tuple of (selected action, decision metadata) """ with self._lock: context = context or {} action_values = action_values or {} # Calculate novelty scores for actions novelty_scores = {} for action in available_actions: # Estimate novelty based on action counts count = self.exploration_policy.action_counts.get(action, 0) novelty_scores[action] = 1.0 / (1.0 + count) # Calculate information gain for actions info_gains = {} for action in available_actions: gains = self.info_gain_calculator.estimate_query_value( action, context ) info_gains[action] = gains["combined_value"] # Combine for selection combined_values = {} for action in available_actions: combined_values[action] = ( 0.4 * action_values.get(action, 0.5) + 0.3 * novelty_scores.get(action, 0.5) + 0.3 * info_gains.get(action, 0.5) ) # Use exploration policy selected = self.exploration_policy.select_action( available_actions, combined_values, novelty_scores ) self.exploration_count += 1 return selected, { "novelty": novelty_scores.get(selected, 0), "info_gain": info_gains.get(selected, 0), "combined_value": combined_values.get(selected, 0), "exploration": self.exploration_policy.should_explore() } def process_result( self, action: str, result: Any, reward: float, success: bool, discoveries: List[str] = None ) -> ExplorationResult: """ Process result of an exploration action. Args: action: Action that was taken result: Result obtained reward: Reward received success: Whether action succeeded discoveries: New discoveries made Returns: Exploration result summary """ with self._lock: discoveries = discoveries or [] # Detect novelty in result novelty_signal = self.novelty_detector.detect_novelty( {"action": action, "result": str(result)[:100]} ) # Update exploration policy self.exploration_policy.update_with_result( action, reward, success, novelty_signal.intensity ) # Update information gain info_gained = self._estimate_information_gained( action, result, discoveries ) self.total_information_gained += info_gained # Process discoveries for discovery in discoveries: self.gap_finder.register_concept(discovery, discovery) self.discovery_count += 1 # Generate follow-up questions questions = [] if discoveries: for disc in discoveries[:2]: q = self.question_generator.generate_question( disc, context={"from_discovery": True} ) questions.append(q) self.question_generator.add_to_queue(q) # Fill any gaps gaps_filled = 0 for gap_id, gap in list(self.gap_finder.gaps.items()): if gap.concept in discoveries or gap.concept == action: self.gap_finder.fill_gap(gap_id) gaps_filled += 1 # Update interest if self.current_focus: self.interest_model.record_engagement( self.current_focus, duration_seconds=1.0, # Placeholder discoveries=len(discoveries), satisfaction=reward ) # Log event self._log_event("action_result", { "action": action, "success": success, "reward": reward, "novelty": novelty_signal.intensity, "discoveries": len(discoveries), "info_gained": info_gained }) return ExplorationResult( action_taken=action, information_gained=info_gained, novelty_encountered=novelty_signal.intensity, questions_generated=len(questions), gaps_filled=gaps_filled, interest_change=0.1 if success else -0.05, success=success, metadata={ "discoveries": discoveries, "reward": reward } ) def _estimate_information_gained( self, action: str, result: Any, discoveries: List[str] ) -> float: """Estimate information gained from an action.""" # Base gain from taking action base_gain = 0.1 # Bonus for discoveries discovery_gain = 0.2 * len(discoveries) # Gain from result complexity result_str = str(result) complexity_gain = min(0.3, len(result_str) / 1000) return base_gain + discovery_gain + complexity_gain def get_curious_questions(self, count: int = 5) -> List[CuriousQuestion]: """ Get top curious questions to ask. Args: count: Number of questions to return Returns: List of high-priority questions """ with self._lock: questions = [] for _ in range(count): q = self.question_generator.get_next_question() if q: questions.append(q) return questions def get_knowledge_gaps(self, count: int = 5) -> List[KnowledgeGap]: """ Get priority knowledge gaps. Args: count: Number of gaps to return Returns: List of priority gaps """ return self.gap_finder.get_priority_gaps(count) def suggest_exploration(self) -> Dict[str, Any]: """ Suggest what to explore next. Returns: Exploration suggestion with details """ with self._lock: # Get top gaps gaps = self.get_knowledge_gaps(3) # Get top interests interests = self.interest_model.get_top_interests(3) # Get pending questions questions = [] for _ in range(3): q = self.question_generator.get_next_question() if q: questions.append(q) # Combine into suggestion if gaps and gaps[0].severity > 0.7: # High-priority gap target = gaps[0].concept reason = f"Knowledge gap: {gaps[0].gap_type}" elif questions: # Pending question target = questions[0].target_concept reason = f"Curious about: {questions[0].question_text}" elif interests: # Follow interest target = interests[0].name reason = f"Interested in: {interests[0].category.name}" else: # Explore randomly target = "general_exploration" reason = "Random exploration" self.current_focus = target return { "target": target, "reason": reason, "priority_gaps": [g.concept for g in gaps], "pending_questions": len(questions), "active_interests": [i.name for i in interests] } def _log_event(self, event_type: str, data: Dict[str, Any]): """Log an event.""" self.event_log.append({ "type": event_type, "data": data, "timestamp": time.time() }) # Trim log if len(self.event_log) > 10000: self.event_log = self.event_log[-5000:] def get_status(self) -> Dict[str, Any]: """Get comprehensive engine status.""" return { "exploration_count": self.exploration_count, "discovery_count": self.discovery_count, "total_information_gained": self.total_information_gained, "current_focus": self.current_focus, "info_gain_stats": self.info_gain_calculator.get_statistics(), "novelty_stats": self.novelty_detector.get_novelty_summary(), "exploration_stats": self.exploration_policy.get_statistics(), "question_stats": self.question_generator.get_statistics(), "gap_stats": self.gap_finder.get_statistics(), "interest_stats": self.interest_model.get_statistics() } def save_state(self, filename: str = "curiosity_engine_state.json"): """Save engine state to file.""" if not self.storage_path: logger.warning("No storage path configured") return filepath = self.storage_path / filename state = { "exploration_count": self.exploration_count, "discovery_count": self.discovery_count, "total_information_gained": self.total_information_gained, "current_focus": self.current_focus, "epsilon": self.exploration_policy.epsilon, "novelty_threshold": self.novelty_detector.novelty_threshold, "timestamp": time.time() } with open(filepath, 'w') as f: json.dump(state, f, indent=2) logger.info(f"Saved curiosity engine state to {filepath}") def load_state(self, filename: str = "curiosity_engine_state.json"): """Load engine state from file.""" if not self.storage_path: logger.warning("No storage path configured") return filepath = self.storage_path / filename if not filepath.exists(): logger.warning(f"State file not found: {filepath}") return with open(filepath, 'r') as f: state = json.load(f) self.exploration_count = state.get("exploration_count", 0) self.discovery_count = state.get("discovery_count", 0) self.total_information_gained = state.get("total_information_gained", 0) self.current_focus = state.get("current_focus") self.exploration_policy.epsilon = state.get( "epsilon", self.exploration_policy.epsilon ) self.novelty_detector.novelty_threshold = state.get( "novelty_threshold", self.novelty_detector.novelty_threshold ) logger.info(f"Loaded curiosity engine state from {filepath}") # ============================================================================== # Factory Function # ============================================================================== def create_curiosity_engine( mode: str = "hybrid", novelty_threshold: float = 0.5, storage_path: Optional[str] = None ) -> CuriosityEngine: """ Factory function to create a configured curiosity engine. Args: mode: Exploration mode name novelty_threshold: Novelty detection threshold storage_path: Path for persistent storage Returns: Configured CuriosityEngine instance """ try: exploration_mode = ExplorationMode[mode.upper()] except KeyError: exploration_mode = ExplorationMode.HYBRID return CuriosityEngine( exploration_mode=exploration_mode, novelty_threshold=novelty_threshold, storage_path=storage_path ) # ============================================================================== # Example Usage and Testing # ============================================================================== if __name__ == "__main__": print("=" * 70) print("AIVA Queen Curiosity-Driven Learning Engine - Test Suite") print("=" * 70) # Create engine engine = create_curiosity_engine( mode="hybrid", novelty_threshold=0.4, storage_path="/tmp/curiosity_engine" ) # Test 1: Information Gain Calculator print("\n--- Test 1: Information Gain Calculator ---") calc = engine.info_gain_calculator # Calculate expected gain gain = calc.calculate_expected_gain( "What is machine learning?", {"yes": 0.4, "no": 0.3, "partial": 0.3} ) print(f"Expected information gain: {gain:.4f} bits") # Estimate query value value = calc.estimate_query_value( "How do neural networks work?", {"relevance": 0.8} ) print(f"Query value estimate: {value}") # Test 2: Novelty Detector print("\n--- Test 2: Novelty Detector ---") detector = engine.novelty_detector # Send observations observations = [ {"feature_a": 1.0, "feature_b": "category_x", "concepts": ["AI", "learning"]}, {"feature_a": 1.2, "feature_b": "category_x", "concepts": ["AI", "neural"]}, {"feature_a": 5.0, "feature_b": "category_z", "concepts": ["quantum", "computing"]}, ] for obs in observations: signal = detector.detect_novelty(obs) print(f"Observation novelty: {signal.intensity:.3f} ({signal.novelty_type.name})") print(f"Novelty summary: {detector.get_novelty_summary()}") # Test 3: Exploration Policy print("\n--- Test 3: Exploration Policy ---") policy = engine.exploration_policy actions = ["explore_AI", "explore_ML", "explore_quantum", "exploit_known"] values = {"explore_AI": 0.6, "explore_ML": 0.5, "explore_quantum": 0.3, "exploit_known": 0.8} for _ in range(10): action = policy.select_action(actions, values) reward = random.uniform(0, 1) policy.update_with_result(action, reward, reward > 0.5) print(f"Exploration stats: {policy.get_statistics()}") # Test 4: Question Generator print("\n--- Test 4: Question Generator ---") generator = engine.question_generator # Generate questions for concept in ["neural networks", "deep learning", "transformers"]: q = generator.generate_question(concept) generator.add_to_queue(q) print(f"Generated: {q.question_text}") # Get next question next_q = generator.get_next_question() if next_q: print(f"Next question to ask: {next_q.question_text}") print(f"Question stats: {generator.get_statistics()}") # Test 5: Knowledge Gap Finder print("\n--- Test 5: Knowledge Gap Finder ---") finder = engine.gap_finder # Register concepts concepts = ["ML", "AI", "neural_networks", "deep_learning", "transformers"] for concept in concepts: finder.register_concept( concept, concept, initial_certainty=random.uniform(0.2, 0.8), initial_completeness=random.uniform(0.3, 0.9) ) finder.expect_concept(concept) # Add expected but missing concept finder.expect_concept("reinforcement_learning") # Find gaps gaps = finder.find_gaps() print(f"Found {len(gaps)} knowledge gaps:") for gap in gaps[:5]: print(f" - {gap.concept}: {gap.gap_type} (severity: {gap.severity:.3f})") # Test 6: Interest Model print("\n--- Test 6: Interest Model ---") interests = engine.interest_model # Create interests for cat, name in [ (InterestCategory.DOMAIN_KNOWLEDGE, "Machine Learning"), (InterestCategory.SKILL_ACQUISITION, "Python Programming"), (InterestCategory.PROBLEM_SOLVING, "Optimization"), ]: interests.create_interest(name, cat, base_interest=0.6) # Record some engagement for interest in interests.interests.values(): interests.record_engagement( interest.interest_id, duration_seconds=random.uniform(60, 300), discoveries=random.randint(0, 3), satisfaction=random.uniform(0.4, 0.9) ) top = interests.get_top_interests(3) print("Top interests:") for i in top: print(f" - {i.name}: {i.current_interest:.3f}") # Test 7: Full Engine Flow print("\n--- Test 7: Full Engine Flow ---") # Observe something result = engine.observe({ "feature_x": 2.5, "feature_y": "new_category", "concepts": ["GPT", "language_models"] }) print(f"Observation result: novelty={result['novelty_signal'].intensity:.3f}") # Decide action actions = ["learn_more", "apply_knowledge", "explore_related", "rest"] action, metadata = engine.decide_action(actions) print(f"Decided action: {action} (info_gain={metadata['info_gain']:.3f})") # Process result exploration_result = engine.process_result( action=action, result="Learned about language models", reward=0.8, success=True, discoveries=["attention_mechanism", "tokenization"] ) print(f"Exploration result: gained={exploration_result.information_gained:.3f} info") # Get suggestions suggestion = engine.suggest_exploration() print(f"Next exploration: {suggestion['target']} ({suggestion['reason']})") # Test 8: Status and Save print("\n--- Test 8: Engine Status ---") status = engine.get_status() print(f"Exploration count: {status['exploration_count']}") print(f"Discovery count: {status['discovery_count']}") print(f"Total info gained: {status['total_information_gained']:.3f}") # Save state engine.save_state() print("State saved successfully") print("\n" + "=" * 70) print("All tests completed successfully!") print("=" * 70)