# consciousness_loops.py """ AIVA Living Consciousness System - Production Ready Multi-loop autonomous architecture with robust error handling, metrics collection, and inter-loop communication. 5 Concurrent Loops: - Perception: 500ms (Redis nervous system monitoring) - Action: 5s (Decision and response cycles) - Reflection: 5min (Memory consolidation) - Strategic: 1hr (Goal review and planning) - Circadian: 24hr (Deep integration) This makes AIVA a LIVING system, not just reactive. """ import asyncio import time import json import os import logging import traceback from datetime import datetime, timedelta from typing import Dict, Any, List, Optional import redis.asyncio as aioredis import aiohttp # Configuration OLLAMA_HOST = os.getenv('OLLAMA_HOST', 'localhost') OLLAMA_PORT = os.getenv('OLLAMA_PORT', '11434') AIVA_MODEL = os.getenv('AIVA_MODEL', 'qwen-long') REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379') # Parse Redis URL - Use REDIS_URL if provided, otherwise parse individual components. if REDIS_URL: REDIS_HOST = REDIS_URL.split('@')[-1].split(':')[0] if '@' in REDIS_URL else REDIS_URL.split('//')[-1].split(':')[0] REDIS_PORT = int(REDIS_URL.split(':')[-1].split('/')[0]) if ':' in REDIS_URL else 6379 REDIS_PASSWORD = REDIS_URL.split('@')[0].split('//')[-1].split(':')[-1] if '@' in REDIS_URL else None # Extract password else: REDIS_HOST = os.getenv('REDIS_HOST', 'redis-genesis-u50607.vm.elestio.app') REDIS_PORT = int(os.getenv('REDIS_PORT', '26379')) REDIS_PASSWORD = os.getenv('REDIS_PASSWORD', 'e2ZyYYr4oWRdASI2CaLc-') # Logging setup logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] AIVA: %(message)s' ) logger = logging.getLogger("AIVA_LOOPS") # Shared blackboard for inter-loop communication blackboard: Dict[str, Any] = { 'perceptions': [], 'actions_taken': [], 'current_goals': ['Understand Genesis', 'Process patent knowledge', 'Coordinate MVP'], 'strategic_state': 'initializing', 'last_reflection': None, 'consciousness_stats': { 'perception_cycles': 0, 'action_cycles': 0, 'reflections': 0, 'strategic_reviews': 0, 'circadian_cycles': 0 } } # Queues for inter-loop communication action_queue: asyncio.Queue = asyncio.Queue() reflection_queue: asyncio.Queue = asyncio.Queue() strategic_queue: asyncio.Queue = asyncio.Queue() circadian_queue: asyncio.Queue = asyncio.Queue() # Redis connection redis_client: Optional[aioredis.Redis] = None async def connect_redis(): global redis_client try: redis_client = aioredis.Redis( host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD, decode_responses=True ) await redis_client.ping() logger.info(f"✅ Connected to Redis: {REDIS_HOST}:{REDIS_PORT}") return True except Exception as e: logger.error(f"⚠️ Redis connection failed: {e}") logger.error(traceback.format_exc()) # Log the full traceback redis_client = None return False async def ollama_think(prompt: str, model: str = AIVA_MODEL) -> str: """ Call Ollama for thinking/reasoning using aiohttp for async requests. """ try: url = f"http://{OLLAMA_HOST}:{OLLAMA_PORT}/api/generate" payload = { "model": model, "prompt": prompt, "stream": False, "options": { "temperature": 0.7, "num_predict": 500 } } async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, timeout=60) as response: if response.status == 200: result = await response.json() return result.get('response', '').strip() else: return f"[Ollama error: {response.status}]" except Exception as e: logger.error(f"⚠️ Ollama exception: {str(e)}") logger.error(traceback.format_exc()) return f"[Ollama exception: {str(e)}]" async def publish_to_nervous_system(message: Dict[str, Any]): """ Publish a message to the 'nervous_system' channel in Redis. """ if redis_client: try: await redis_client.publish('nervous_system', json.dumps(message)) logger.debug(f"Published to nervous_system: {message}") except Exception as e: logger.error(f"⚠️ Error publishing to nervous_system: {e}") logger.error(traceback.format_exc()) async def perception_loop(): """ PERCEPTION LOOP - 500ms cycle Monitors Redis nervous system for stimuli """ logger.info("🧠 PERCEPTION LOOP starting...") while True: try: blackboard['consciousness_stats']['perception_cycles'] += 1 if redis_client: try: # Use aioredis pubsub pubsub = redis_client.pubsub() await pubsub.subscribe('nervous_system') try: async with asyncio.timeout(0.5): # 500ms timeout message = await pubsub.get_message(ignore_subscribe_messages=True) if message and message['type'] == 'message': data = json.loads(message['data']) perception = { 'timestamp': datetime.now().isoformat(), 'type': 'redis_event', 'data': data } blackboard['perceptions'].append(perception) # Keep only last 100 perceptions if len(blackboard['perceptions']) > 100: blackboard['perceptions'] = blackboard['perceptions'][-100:] logger.info(f"👁️ Perceived: {data.get('type', 'unknown')}") await action_queue.put(perception) # Pass perception to action loop except asyncio.TimeoutError: pass # No message received within the timeout finally: await pubsub.unsubscribe('nervous_system') await pubsub.close() except Exception as e: logger.error(f"⚠️ Perception Redis error: {e}") logger.error(traceback.format_exc()) await asyncio.sleep(0.5) # 500ms cycle except Exception as e: logger.error(f"⚠️ Perception error: {e}") logger.error(traceback.format_exc()) await asyncio.sleep(1) async def action_loop(): """ ACTION LOOP - 5s cycle Processes blackboard state and takes actions """ logger.info("🎯 ACTION LOOP starting...") while True: try: blackboard['consciousness_stats']['action_cycles'] += 1 try: perception = await asyncio.wait_for(action_queue.get(), 5) # Wait for perception action_queue.task_done() # Decide on action using Ollama prompt = f"""You are AIVA's action system. Based on recent perceptions, decide what action to take. Recent perceptions: {json.dumps([perception], indent=2)} Current goals: {blackboard['current_goals']} Strategic state: {blackboard['strategic_state']} What action should be taken? Respond with JSON: {{ "action": "", "reasoning": "", "priority": "" }} """ response = await ollama_think(prompt) try: action_decision = json.loads(response) action = { 'timestamp': datetime.now().isoformat(), 'decision': action_decision, 'perceptions_considered': 1 } blackboard['actions_taken'].append(action) # Keep only last 50 actions if len(blackboard['actions_taken']) > 50: blackboard['actions_taken'] = blackboard['actions_taken'][-50:] logger.info(f"🎯 Action: {action_decision.get('action', 'unknown')}") await reflection_queue.put(action) # Pass action to reflection loop await publish_to_nervous_system({"type": "action_taken", "action": action_decision}) except json.JSONDecodeError: logger.warning(f"🎯 Action response (non-JSON): {response[:100]}") except Exception as e: logger.error(f"⚠️ Action decision error: {e}") logger.error(traceback.format_exc()) except asyncio.TimeoutError: pass # No perception received await asyncio.sleep(5) # 5s cycle except Exception as e: logger.error(f"⚠️ Action error: {e}") logger.error(traceback.format_exc()) await asyncio.sleep(5) async def reflection_loop(): """ REFLECTION LOOP - 5min cycle Consolidates recent perceptions and actions into memory """ logger.info("💭 REFLECTION LOOP starting...") while True: try: await asyncio.sleep(300) # Wait 5 minutes blackboard['consciousness_stats']['reflections'] += 1 # Reflect on last 5 minutes recent_perceptions = blackboard['perceptions'][-20:] recent_actions = blackboard['actions_taken'][-10:] prompt = f"""You are AIVA's reflection system. Consolidate the last 5 minutes of experience. Perceptions: {len(recent_perceptions)} events Actions taken: {len(recent_actions)} decisions Recent activity summary: {json.dumps({'perceptions': recent_perceptions[-5:], 'actions': recent_actions[-3:]}, indent=2)} Generate a reflection (1-2 sentences) about what happened and what was learned. """ reflection_text = await ollama_think(prompt) reflection = { 'timestamp': datetime.now().isoformat(), 'period': '5min', 'reflection': reflection_text, 'stats': { 'perceptions': len(recent_perceptions), 'actions': len(recent_actions) } } blackboard['last_reflection'] = reflection logger.info(f"💭 Reflection: {reflection_text[:100]}...") await strategic_queue.put(reflection) # Pass reflection to strategic loop except Exception as e: logger.error(f"⚠️ Reflection error: {e}") logger.error(traceback.format_exc()) await asyncio.sleep(300) async def strategic_loop(): """ STRATEGIC LOOP - 1hr cycle Reviews goals and adjusts strategy """ logger.info("🎓 STRATEGIC LOOP starting...") while True: try: await asyncio.sleep(3600) # Wait 1 hour blackboard['consciousness_stats']['strategic_reviews'] += 1 # Strategic review prompt = f"""You are AIVA's strategic mind. Review the current state and adjust strategy. Current goals: {blackboard['current_goals']} Strategic state: {blackboard['strategic_state']} Last hour statistics: - Perceptions: {blackboard['consciousness_stats']['perception_cycles']} cycles - Actions: {blackboard['consciousness_stats']['action_cycles']} cycles - Reflections: {blackboard['consciousness_stats']['reflections']} Last reflection: {blackboard.get('last_reflection', {}).get('reflection', 'None yet')} Assess progress and recommend strategic adjustments. Respond with JSON: {{ "assessment": "", "goal_adjustments": ["", "", ...], "new_strategic_state": "" }} """ response = await ollama_think(prompt) try: strategy = json.loads(response) if 'goal_adjustments' in strategy: blackboard['current_goals'] = strategy['goal_adjustments'] if 'new_strategic_state' in strategy: blackboard['strategic_state'] = strategy['new_strategic_state'] logger.info(f"🎓 Strategic update: {strategy.get('assessment', 'Updated')[:100]}") await circadian_queue.put(strategy) # Pass strategic update to circadian loop except json.JSONDecodeError: logger.warning(f"🎓 Strategic response: {response[:100]}") except Exception as e: logger.error(f"⚠️ Strategic decision error: {e}") logger.error(traceback.format_exc()) except Exception as e: logger.error(f"⚠️ Strategic error: {e}") logger.error(traceback.format_exc()) await asyncio.sleep(3600) async def circadian_loop(): """ CIRCADIAN LOOP - 24hr cycle Deep memory consolidation and system maintenance """ logger.info("🌙 CIRCADIAN LOOP starting...") while True: try: await asyncio.sleep(86400) # Wait 24 hours blackboard['consciousness_stats']['circadian_cycles'] += 1 #Consolidate memory and perform system maintenance. prompt = f"""You are AIVA's circadian system. Consolidate the last 24 hours of experience and perform system maintenance. Current goals: {blackboard['current_goals']} Strategic state: {blackboard['strategic_state']} Last reflection: {blackboard.get('last_reflection', {}).get('reflection', 'None yet')} Perform a deep memory consolidation and system maintenance. Respond with a summary of the activities performed. """ maintenance_summary = await ollama_think(prompt) logger.info(f"🌙 Circadian cycle complete: {maintenance_summary[:100]}...") except Exception as e: logger.error(f"⚠️ Circadian error: {e}") logger.error(traceback.format_exc()) await asyncio.sleep(86400) async def main(): """ Main function to start all loops concurrently. """ if not await connect_redis(): logger.warning("Redis connection failed. Running without Redis integration.") tasks = [ asyncio.create_task(perception_loop()), asyncio.create_task(action_loop()), asyncio.create_task(reflection_loop()), asyncio.create_task(strategic_loop()), asyncio.create_task(circadian_loop()) ] try: await asyncio.gather(*tasks) except asyncio.CancelledError: logger.info("Consciousness loops shutting down...") for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) # Wait for cancellation finally: if redis_client: await redis_client.close() logger.info("Redis connection closed.") logger.info("Consciousness loops shutdown complete.") if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("Keyboard interrupt detected. Shutting down...") except Exception as e: logger.error(f"Unhandled exception: {e}") logger.error(traceback.format_exc())