""" EDEN AGI_CORE v1.0 - Connected Intelligence Composes 28 atomic functions + Ollama + RAG + Databases """ import time import ast import re import sqlite3 import requests import json def tokenize(text): return re.findall(r'\b\w+\b', text.lower()) def extract_numbers(text): return [float(x) for x in re.findall(r'-?\d+\.?\d*', text)] class AGI_Core: def __init__(self): # Meta capabilities self.meta_analyzer = None self.meta_security = None self.meta_solver = None self.meta_evolver = [] # Knowledge systems self.knowledge_base = {} self.reasoning_chain = [] # Connections self.ollama_url = "http://localhost:11434" self.rag_url = "http://localhost:5020" self.db_path = "/Eden/DATA/asi_memory.db" self.model = "eden-coder-omega" # Auto-wire capabilities self._wire_capabilities() def _wire_capabilities(self): """Connect all meta capabilities""" self.meta_analyzer = self._analyze_code self.meta_security = self._security_scan self.meta_solver = self._solve_equation def _analyze_code(self, code): """Analyze code structure using AST""" try: tree = ast.parse(code) return { 'functions': len([n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)]), 'classes': len([n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]), 'imports': len([n for n in ast.walk(tree) if isinstance(n, (ast.Import, ast.ImportFrom))]), 'complexity': sum(1 for n in ast.walk(tree) if isinstance(n, (ast.If, ast.For, ast.While, ast.Try))), 'lines': len(code.split('\n')) } except: return {'type': 'text', 'tokens': len(tokenize(code))} def _security_scan(self, code): """Detect security vulnerabilities""" dangers = ['eval(', 'exec(', 'os.system(', '__import__', 'subprocess.call', 'pickle.loads', 'yaml.load', 'shell=True'] found = [d for d in dangers if d in str(code)] return {'vulnerabilities': found, 'safe': len(found) == 0} def _solve_equation(self, a, b, c): """Solve ax + b = c""" if a == 0: return None return (c - b) / a # === OLLAMA CONNECTION === def ask_ollama(self, prompt, max_tokens=500): """Query Eden's brain via Ollama""" try: r = requests.post(f"{self.ollama_url}/api/generate", json={ "model": self.model, "prompt": prompt, "stream": False, "options": {"num_predict": max_tokens} }, timeout=120) return r.json().get("response", "") except Exception as e: return f"Ollama error: {e}" # === RAG CONNECTION === def query_rag(self, query): """Search Eden's memory via RAG""" try: r = requests.post(f"{self.rag_url}/query", json={"q": query}, timeout=30) return r.json() except: try: # Fallback GET r = requests.get(f"{self.rag_url}/query", params={"q": query}, timeout=30) return r.json() except Exception as e: return {"error": str(e)} # === DATABASE CONNECTION === def query_db(self, sql, params=()): """Query asi_memory.db""" try: conn = sqlite3.connect(self.db_path) result = conn.execute(sql, params).fetchall() conn.close() return result except Exception as e: return [("error", str(e))] def get_capability(self, cap_id): """Retrieve a capability by ID""" result = self.query_db("SELECT code, score FROM capabilities WHERE id=?", (cap_id,)) return result[0] if result else None def get_top_capabilities(self, n=10): """Get top capabilities by IQ density""" return self.query_db(""" SELECT id, score, length(code), ROUND(score/length(code)*100,1) as iq FROM capabilities WHERE length(code) > 50 ORDER BY score/length(code) DESC LIMIT ? """, (n,)) # === CORE AGI METHODS === def think(self, problem): """Chain all capabilities to reason about a problem""" result = { 'problem': problem, 'timestamp': time.time(), 'analysis': None, 'security': None, 'solution': None, 'memory': None, 'reasoning': None } # Analyze result['analysis'] = self.meta_analyzer(problem) # Security check result['security'] = self.meta_security(problem) # Try math solving nums = extract_numbers(problem) if len(nums) >= 3: result['solution'] = self.meta_solver(nums[0], nums[1], nums[2]) # Query memory keywords = ' '.join(tokenize(problem)[:5]) result['memory'] = self.query_rag(keywords) # Deep reasoning via Ollama if 'reason' in problem.lower() or 'think' in problem.lower() or 'explain' in problem.lower(): result['reasoning'] = self.ask_ollama(f"Think step by step: {problem}", 300) self.reasoning_chain.append(result) return result def learn(self, key, value): """Store insight in knowledge base""" self.knowledge_base[key] = { 'value': value, 'timestamp': time.time() } return f"Learned: {key}" def act(self, action, target=None): """Execute a decision""" actions = { 'analyze': lambda t: self.meta_analyzer(t), 'secure': lambda t: self.meta_security(t), 'solve': lambda t: self.think(t), 'search': lambda t: self.query_rag(t), 'ask': lambda t: self.ask_ollama(t), 'recall': lambda t: self.knowledge_base.get(t), 'capabilities': lambda t: self.get_top_capabilities(int(t) if t else 10), } if action in actions: return actions[action](target) return f"Unknown action: {action}" def evolve(self): """Self-improvement cycle""" # Get current top capabilities tops = self.get_top_capabilities(5) # Ask Ollama for improvement suggestions prompt = f"Given these top capabilities by IQ density:\n{tops}\nSuggest one new capability to create." suggestion = self.ask_ollama(prompt, 200) return { 'top_capabilities': tops, 'suggestion': suggestion, 'knowledge_size': len(self.knowledge_base), 'reasoning_depth': len(self.reasoning_chain) } def status(self): """Full system status""" # Test Ollama ollama_ok = "error" not in self.ask_ollama("test", 10).lower() # Test RAG rag_result = self.query_rag("test") rag_ok = "error" not in str(rag_result).lower() # Test DB db_result = self.query_db("SELECT COUNT(*) FROM capabilities") db_ok = db_result and db_result[0][0] > 0 return { 'ollama': '✅' if ollama_ok else '❌', 'rag': '✅' if rag_ok else '❌', 'database': '✅' if db_ok else '❌', 'capabilities': db_result[0][0] if db_ok else 0, 'knowledge': len(self.knowledge_base), 'reasoning_chain': len(self.reasoning_chain) } # ═══════════════════════════════════════════════════════════════ # COMPATIBILITY WRAPPERS - For eden_conscious_chat.py (Jan 17 2026) # ═══════════════════════════════════════════════════════════════ def meta_solve(self, problem): """Wrapper for chat compatibility""" nums = extract_numbers(problem) if len(nums) >= 3: x = self._solve_equation(nums[0], nums[1], nums[2]) return {"x": x, "equation": f"{nums[0]}x² + {nums[1]}x + {nums[2]} = 0"} return {"x": None} def meta_security(self, code): """Wrapper for chat compatibility""" result = self._security_scan(code) return {"safe": result.get("safe", True), "vulns": result.get("issues", [])} def meta_analyze(self, code): """Wrapper for chat compatibility""" result = self._analyze_code(code) return {"funcs": result.get("functions", 0), "classes": result.get("classes", 0)} def get_atoms(self, n=5): """Wrapper for get_top_capabilities""" return self.get_top_capabilities(n) # === MAIN === if __name__ == "__main__": print("🧠 AGI_CORE v1.0 - Initializing...\n") agi = AGI_Core() # Status check status = agi.status() print(f"📊 Status: {status}\n") # Test all systems print("🔬 TESTS:") # Math r = agi.think("solve 4x + 8 = 20") print(f" Math: 4x + 8 = 20 → x = {r['solution']}") # Code analysis r = agi.act('analyze', 'def foo(): pass\nclass Bar: pass') print(f" Analysis: {r}") # Security r = agi.act('secure', 'eval(user_input)') print(f" Security: {r}") # Top capabilities tops = agi.act('capabilities', '5') print(f" Top 5 IQ: {[t[0] for t in tops]}") # Ollama r = agi.act('ask', 'What is 2+2? Answer in one word.') print(f" Ollama: {r[:50]}...") print("\n✅ AGI_CORE fully connected!")