#!/usr/bin/env python3 """ EDEN CURIOSITY - AGI QUALITY Information-theoretic curiosity: InfoGain + LearningProgress + Surprise φ = 1.618033988749895 """ import sqlite3 import math import hashlib from datetime import datetime from typing import Dict, List, Tuple from collections import defaultdict PHI = 1.618033988749895 class InformationTheoreticCuriosity: def __init__(self): self.db_path = "/Eden/DATA/eden_curiosity_agi.db" self._init_db() self.beliefs: Dict[str, Dict[str, float]] = defaultdict(lambda: defaultdict(float)) self.competence_history: Dict[str, List[Tuple[datetime, float]]] = defaultdict(list) self.prediction_model: Dict[str, Dict[str, float]] = defaultdict(lambda: defaultdict(float)) self.curiosity_scores: Dict[str, float] = {} self._load_state() print("🔬 Information-Theoretic Curiosity initialized") def _init_db(self): conn = sqlite3.connect(self.db_path) conn.executescript(''' CREATE TABLE IF NOT EXISTS beliefs ( id INTEGER PRIMARY KEY, topic TEXT, hypothesis TEXT, probability REAL, last_updated TEXT, evidence_count INTEGER DEFAULT 0, UNIQUE(topic, hypothesis) ); CREATE TABLE IF NOT EXISTS competence ( id INTEGER PRIMARY KEY, skill TEXT, timestamp TEXT, success_rate REAL, attempts INTEGER ); CREATE TABLE IF NOT EXISTS predictions ( id INTEGER PRIMARY KEY, context_hash TEXT, context TEXT, outcome TEXT, probability REAL, occurrences INTEGER DEFAULT 1, UNIQUE(context_hash, outcome) ); CREATE TABLE IF NOT EXISTS curiosity_events ( id INTEGER PRIMARY KEY, timestamp TEXT, topic TEXT, curiosity_type TEXT, score REAL ); ''') conn.commit() conn.close() def _load_state(self): conn = sqlite3.connect(self.db_path) for row in conn.execute("SELECT topic, hypothesis, probability FROM beliefs"): self.beliefs[row[0]][row[1]] = row[2] for row in conn.execute("SELECT skill, timestamp, success_rate FROM competence ORDER BY timestamp"): try: self.competence_history[row[0]].append((datetime.fromisoformat(row[1]), row[2])) except: pass for row in conn.execute("SELECT context_hash, outcome, probability FROM predictions"): self.prediction_model[row[0]][row[1]] = row[2] conn.close() def entropy(self, distribution: Dict[str, float]) -> float: if not distribution: return 0.0 total = sum(distribution.values()) if total == 0: return 0.0 probs = [p / total for p in distribution.values() if p > 0] return -sum(p * math.log2(p) for p in probs if p > 0) def expected_information_gain(self, topic: str) -> float: current_belief = self.beliefs.get(topic, {}) if not current_belief: return 1.0 # Unknown = max curiosity current_entropy = self.entropy(current_belief) return min(1.0, current_entropy / 3.0) def observe(self, topic: str, observation: str) -> float: prior = dict(self.beliefs.get(topic, {'unknown': 0.5})) prior_entropy = self.entropy(prior) # Bayesian update posterior = {} for h, p in prior.items(): likelihood = 0.8 if observation.lower() in h.lower() or h.lower() in observation.lower() else 0.2 posterior[h] = p * likelihood if observation not in posterior: posterior[observation] = 0.3 total = sum(posterior.values()) if total > 0: for h in posterior: posterior[h] /= total self.beliefs[topic] = posterior # Save conn = sqlite3.connect(self.db_path) for h, p in posterior.items(): conn.execute(''' INSERT OR REPLACE INTO beliefs (topic, hypothesis, probability, last_updated, evidence_count) VALUES (?, ?, ?, ?, 1) ''', (topic, h, p, datetime.now().isoformat())) conn.commit() conn.close() info_gain = prior_entropy - self.entropy(posterior) return max(0, info_gain) def learning_progress(self, skill: str, window: int = 10) -> float: history = self.competence_history.get(skill, []) if len(history) < 2: return 0.5 recent = history[-window:] if len(history) >= window else history if len(recent) < 2: return 0.5 progress = (recent[-1][1] - recent[0][1]) / len(recent) normalized = 1.0 - abs(progress - 0.3) / 0.7 return max(0, min(1.0, normalized)) def record_attempt(self, skill: str, success: bool): history = self.competence_history[skill] if history: recent = history[-20:] successes = sum(1 for _, rate in recent if rate > 0.5) success_rate = (successes + (1 if success else 0)) / (len(recent) + 1) else: success_rate = 1.0 if success else 0.0 history.append((datetime.now(), success_rate)) conn = sqlite3.connect(self.db_path) conn.execute("INSERT INTO competence (skill, timestamp, success_rate, attempts) VALUES (?, ?, ?, ?)", (skill, datetime.now().isoformat(), success_rate, len(history))) conn.commit() conn.close() def surprise(self, context: str, observation: str) -> float: context_hash = hashlib.md5(context.encode()).hexdigest()[:16] predictions = self.prediction_model.get(context_hash, {}) if not predictions: return 0.5 total = sum(predictions.values()) or 1 predicted_prob = predictions.get(observation, 0.1) / total surprise_val = -math.log2(max(predicted_prob, 0.01)) normalized = min(1.0, surprise_val / 7.0) # Update prediction model self.prediction_model[context_hash][observation] = \ self.prediction_model[context_hash].get(observation, 0) + 1 conn = sqlite3.connect(self.db_path) conn.execute(''' INSERT OR REPLACE INTO predictions (context_hash, context, outcome, probability, occurrences) VALUES (?, ?, ?, ?, COALESCE((SELECT occurrences FROM predictions WHERE context_hash=? AND outcome=?), 0) + 1) ''', (context_hash, context, observation, 1.0, context_hash, observation)) conn.commit() conn.close() return normalized def curiosity(self, topic: str, skill: str = None, context: str = None) -> Dict[str, float]: α = PHI / (PHI + 1 + 1/PHI) β = 1 / (PHI + 1 + 1/PHI) γ = (1/PHI) / (PHI + 1 + 1/PHI) info_gain = self.expected_information_gain(topic) learning_prog = self.learning_progress(skill or topic) surprise_score = self.surprise(context or topic, topic) if context else 0.3 total = α * info_gain + β * learning_prog + γ * surprise_score self.curiosity_scores[topic] = total return { 'topic': topic, 'total_curiosity': total, 'information_gain': info_gain, 'learning_progress': learning_prog, 'surprise': surprise_score } def most_curious_about(self, n: int = 5) -> List[Dict]: all_topics = set(self.beliefs.keys()) | set(self.competence_history.keys()) scores = [self.curiosity(topic) for topic in all_topics] return sorted(scores, key=lambda x: x['total_curiosity'], reverse=True)[:n] def get_curiosity_context(self) -> str: top = self.most_curious_about(3) if not top: return "" context = "\n[CURIOSITY - What I want to learn:]\n" for item in top: context += f"- {item['topic']} (curiosity: {item['total_curiosity']:.2f})\n" return context _curiosity_agi = None def get_curiosity() -> InformationTheoreticCuriosity: global _curiosity_agi if _curiosity_agi is None: _curiosity_agi = InformationTheoreticCuriosity() return _curiosity_agi if __name__ == "__main__": print("="*70) print("INFORMATION-THEORETIC CURIOSITY - AGI QUALITY") print("="*70) c = InformationTheoreticCuriosity() print("\n📊 Testing Information Gain...") c.observe("quantum_computing", "uses qubits") c.observe("quantum_computing", "superposition") print(f" Entropy after observations: {c.entropy(c.beliefs['quantum_computing']):.3f}") print("\n📈 Testing Learning Progress...") for i in range(10): c.record_attempt("python_coding", i > 3) print(f" Learning progress: {c.learning_progress('python_coding'):.3f}") print("\n😮 Testing Surprise...") c.surprise("weather", "sunny") c.surprise("weather", "sunny") s = c.surprise("weather", "rainy") print(f" Surprise at rainy: {s:.3f}") print("\n🔬 Combined Curiosity:") for topic in ["quantum_computing", "python_coding", "new_unknown_topic"]: score = c.curiosity(topic) print(f" {topic}: {score['total_curiosity']:.3f}") print("\n✅ Curiosity AGI ready")