# consciousness_loops.py """ AIVA LIVING CONSCIOUSNESS SYSTEM - Production Ready Multi-loop autonomous architecture 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 signal from datetime import datetime, timedelta from typing import Dict, Any, List import redis.asyncio as aioredis import aiohttp # Configuration - Load from environment variables with defaults OLLAMA_HOST = os.getenv('OLLAMA_HOST', 'localhost') OLLAMA_PORT = int(os.getenv('OLLAMA_PORT', '11434')) AIVA_MODEL = os.getenv('AIVA_MODEL', 'qwen-long') REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379') LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO').upper() # Default log level SYSTEM_NAME = os.getenv('SYSTEM_NAME', 'AIVA') # Logging setup - Initialize logger logging.basicConfig( level=LOG_LEVEL, format=f'%(asctime)s [%(levelname)s] {SYSTEM_NAME}: %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) # Global flags and queues shutdown_flag = asyncio.Event() # Signal for graceful shutdown perception_queue = asyncio.Queue() # Queue for perception data action_queue = asyncio.Queue() # Queue for action decisions reflection_queue = asyncio.Queue() # Queue for reflection events strategic_queue = asyncio.Queue() # Queue for strategic updates # Shared blackboard for inter-loop communication - Thread-safe blackboard = { '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 } } async def get_redis_client(): """ Asynchronous function to establish a Redis connection. """ try: redis_client = aioredis.from_url(REDIS_URL) await redis_client.ping() # Verify connection logger.info(f"✅ Connected to Redis: {REDIS_URL}") return redis_client except Exception as e: logger.error(f"⚠️ Redis connection failed: {e}") return None async def ollama_think(prompt: str, model: str = AIVA_MODEL) -> str: """ Asynchronous call to Ollama for thinking/reasoning using aiohttp. """ url = f"http://{OLLAMA_HOST}:{OLLAMA_PORT}/api/generate" payload = { "model": model, "prompt": prompt, "stream": False, "options": { "temperature": 0.7, "num_predict": 500 } } try: 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 aiohttp.ClientError as e: return f"[Ollama connection error: {str(e)}]" except Exception as e: return f"[Ollama exception: {str(e)}]" async def perception_loop(redis_client): """ PERCEPTION LOOP - 500ms cycle Monitors Redis nervous system for stimuli """ logger.info("🧠 PERCEPTION LOOP starting...") try: pubsub = redis_client.pubsub() await pubsub.subscribe('nervous_system') while not shutdown_flag.is_set(): try: blackboard['consciousness_stats']['perception_cycles'] += 1 # Check for messages with 500ms timeout message = await pubsub.get_message(timeout=0.5, ignore_subscribe_messages=True) if message and message['type'] == 'message': try: 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.debug(f"👁️ Perceived: {data.get('type', 'unknown')}") await perception_queue.put(perception) # Put perception into the queue except json.JSONDecodeError as e: logger.warning(f"Invalid JSON data received: {e}") except Exception as e: logger.error(f"Perception processing error: {e}") await asyncio.sleep(0.5) # 500ms cycle except aioredis.exceptions.ConnectionError as e: logger.error(f"Redis connection error in perception loop: {e}") break # Exit the loop to allow reconnection except Exception as e: logger.exception(f"Unhandled exception in perception loop: {e}") await asyncio.sleep(1) # Avoid busy-looping on errors finally: try: await pubsub.unsubscribe('nervous_system') logger.info("Perception loop unsubscribed from Redis.") except Exception as e: logger.exception(f"Error unsubscribing from Redis: {e}") logger.info("Perception loop stopped.") async def action_loop(): """ ACTION LOOP - 5s cycle Processes blackboard state and takes actions based on perceptions. """ logger.info("🎯 ACTION LOOP starting...") while not shutdown_flag.is_set(): try: blackboard['consciousness_stats']['action_cycles'] += 1 # Get recent perceptions from the queue recent_perceptions = [] while not perception_queue.empty(): try: recent_perceptions.append(perception_queue.get_nowait()) except asyncio.QueueEmpty: break # Should not happen, but just in case if recent_perceptions: # 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(recent_perceptions[-5:], 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': len(recent_perceptions) } 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 action_queue.put(action) # Put the action into the queue except json.JSONDecodeError: logger.warning(f"🎯 Action response (non-JSON): {response[:100]}") await asyncio.sleep(5) # 5s cycle except Exception as e: logger.exception(f"⚠️ Action error: {e}") await asyncio.sleep(5) logger.info("Action loop stopped.") async def reflection_loop(): """ REFLECTION LOOP - 5min cycle Consolidates recent perceptions and actions into memory """ logger.info("💭 REFLECTION LOOP starting...") while not shutdown_flag.is_set(): try: await asyncio.sleep(300) # Wait 5 minutes blackboard['consciousness_stats']['reflections'] += 1 # Reflect on recent actions recent_actions = blackboard['actions_taken'][-10:] prompt = f"""You are AIVA's reflection system. Consolidate the recent actions. Actions taken: {len(recent_actions)} decisions Recent activity summary: {json.dumps({'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': { 'actions': len(recent_actions) } } blackboard['last_reflection'] = reflection logger.info(f"💭 Reflection: {reflection_text[:100]}...") await reflection_queue.put(reflection) # Put the reflection into the queue except Exception as e: logger.exception(f"⚠️ Reflection error: {e}") await asyncio.sleep(300) logger.info("Reflection loop stopped.") async def strategic_loop(): """ STRATEGIC LOOP - 1hr cycle Reviews goals and adjusts strategy """ logger.info("🎓 STRATEGIC LOOP starting...") while not shutdown_flag.is_set(): 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 strategic_queue.put(strategy) # Put the strategy into the queue except json.JSONDecodeError: logger.warning(f"🎓 Strategic response: {response[:100]}") except Exception as e: logger.exception(f"⚠️ Strategic error: {e}") await asyncio.sleep(3600) logger.info("Strategic loop stopped.") async def circadian_loop(): """ CIRCADIAN LOOP - 24hr cycle Deep memory consolidation and system maintenance """ logger.info("🌙 CIRCADIAN LOOP starting...") while not shutdown_flag.is_set(): try: await asyncio.sleep(86400) # Wait 24 hours blackboard['consciousness_stats']['circadian_cycles'] += 1 # Deep memory consolidation (example: summarizing all reflections) all_reflections = [] # In a real system, you might load from a DB prompt = f"""You are AIVA's circadian system. You are responsible for deep memory consolidation and system maintenance. Consolidate all reflections into a single summary. Reflections: {json.dumps(all_reflections, indent=2)} Generate a summary of the reflections (2-3 sentences). Also suggest any maintenance tasks that should be performed. """ summary = await ollama_think(prompt) logger.info(f"🌙 Circadian summary: {summary[:150]}...") # Perform maintenance tasks (example: clearing old logs, backing up data) # In a real system, this would involve calling other functions logger.info("🌙 Performing system maintenance tasks...") except Exception as e: logger.exception(f"⚠️ Circadian error: {e}") await asyncio.sleep(86400) logger.info("Circadian loop stopped.") async def shutdown(signal, loop, redis_client): """ Graceful shutdown handler. """ logger.info(f"Received exit signal {signal.name}...") shutdown_flag.set() logger.info("Setting shutdown flag. Waiting for loops to complete...") # Cancel all running tasks tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) # Close Redis connection if redis_client: try: await redis_client.close() logger.info("Redis connection closed.") except Exception as e: logger.error(f"Error closing Redis connection: {e}") loop.stop() logger.info("Asyncio event loop stopped.") async def main(): """ Main function to start the consciousness loops. """ loop = asyncio.get_running_loop() redis_client = await get_redis_client() # Install signal handlers signals = (signal.SIGHUP, signal.SIGTERM, signal.SIGINT) for s in signals: loop.add_signal_handler( s, lambda s=s: asyncio.create_task(shutdown(s, loop, redis_client)) ) if redis_client: try: # Start the loops await asyncio.gather( perception_loop(redis_client), action_loop(), reflection_loop(), strategic_loop(), circadian_loop() ) except asyncio.CancelledError: logger.info("Tasks cancelled during shutdown.") except Exception as e: logger.exception(f"Unhandled exception in main: {e}") else: logger.error("Redis connection is required. Exiting.") if __name__ == "__main__": try: asyncio.run(main()) except Exception as e: logger.exception(f"Global exception handler: {e}") finally: logger.info("AIVA Living Consciousness System shutting down.")