# consciousness_loops.py """ AIVA LIVING CONSCIOUSNESS SYSTEM - Production-Ready Multi-loop autonomous architecture with robust error handling, metrics collection, and graceful shutdown. 5 Concurrent Loops: - Perception: 500ms (Real-time event sensing) - Action: 5s (Decision and execution) - Reflection: 5min (Pattern consolidation) - Strategic: 1hr (Goal adjustment) - Circadian: 24hr (Deep memory integration) Inter-loop communication via asyncio.Queue. """ import asyncio import time import json import os import logging from datetime import datetime, timedelta from typing import Dict, Any, List, Optional import redis import requests # Logging Setup logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] LOOP: %(name)s - %(message)s', handlers=[logging.StreamHandler()] ) # Configuration (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') # Redis Configuration 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-') # Queues for inter-loop communication perception_queue: asyncio.Queue = asyncio.Queue() action_queue: asyncio.Queue = asyncio.Queue() reflection_queue: asyncio.Queue = asyncio.Queue() strategic_queue: asyncio.Queue = asyncio.Queue() circadian_queue: asyncio.Queue = asyncio.Queue() # Global State (Use with caution; consider alternatives like shared memory or Redis) class GlobalState: def __init__(self): self.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 } } self.running = True # Flag for graceful shutdown self.last_error = None global_state = GlobalState() # Redis Connection def create_redis_client(): try: redis_client = redis.Redis( host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD, decode_responses=True ) redis_client.ping() logging.info(f"Redis connected: {REDIS_HOST}:{REDIS_PORT}") return redis_client except Exception as e: logging.error(f"Redis connection failed: {e}") return None redis_client = create_redis_client() # Initialize Redis client def ollama_think(prompt: str, model: str = AIVA_MODEL) -> str: """ Call Ollama for thinking/reasoning with retry logic. """ max_retries = 3 retry_delay = 5 # seconds for attempt in range(max_retries): try: url = f"http://{OLLAMA_HOST}:{OLLAMA_PORT}/api/generate" payload = { "model": model, "prompt": prompt, "stream": False, "options": { "temperature": 0.7, "num_predict": 500 } } response = requests.post(url, json=payload, timeout=60) if response.status_code == 200: result = response.json() return result.get('response', '').strip() else: logging.error(f"Ollama error (attempt {attempt + 1}/{max_retries}): {response.status_code}, {response.text}") return f"[Ollama error: {response.status_code}]" except requests.exceptions.RequestException as e: logging.error(f"Ollama request exception (attempt {attempt + 1}/{max_retries}): {e}") if attempt < max_retries - 1: logging.info(f"Retrying in {retry_delay} seconds...") time.sleep(retry_delay) # Use time.sleep for non-async delays else: return f"[Ollama request exception: {str(e)}]" except Exception as e: logging.error(f"Ollama general exception (attempt {attempt + 1}/{max_retries}): {e}") return f"[Ollama exception: {str(e)}]" return "[Ollama failed after multiple retries]" async def perception_loop(): """ PERCEPTION LOOP - 500ms cycle Monitors Redis nervous system for stimuli and adds perceptions to the queue. """ logger = logging.getLogger("PerceptionLoop") logger.info("Starting...") while global_state.running: try: global_state.blackboard['consciousness_stats']['perception_cycles'] += 1 if redis_client: try: # Listen for nervous system events (non-blocking with timeout) pubsub = redis_client.pubsub() pubsub.subscribe('nervous_system') # Check for messages with 500ms timeout message = pubsub.get_message(timeout=0.5) if message and message['type'] == 'message': data = json.loads(message['data']) perception = { 'timestamp': datetime.now().isoformat(), 'type': 'redis_event', 'data': data } # Add to perception queue await perception_queue.put(perception) logger.debug(f"Perceived: {data.get('type', 'unknown')}") pubsub.close() # Close pubsub after each attempt except redis.exceptions.ConnectionError as e: logger.error(f"Redis connection error: {e}. Reconnecting...") global redis_client redis_client = create_redis_client() # Attempt to reconnect except Exception as e: logger.error(f"Redis processing error: {e}") await asyncio.sleep(0.5) # 500ms cycle except asyncio.CancelledError: logger.info("Cancelled.") break except Exception as e: logger.exception("Unhandled exception in perception loop.") global_state.last_error = f"Perception Loop: {e}" await asyncio.sleep(1) async def action_loop(): """ ACTION LOOP - 5s cycle Processes perceptions from the queue, decides on actions, and adds them to the blackboard. """ logger = logging.getLogger("ActionLoop") logger.info("Starting...") while global_state.running: try: global_state.blackboard['consciousness_stats']['action_cycles'] += 1 # Get perceptions from the queue (non-blocking) recent_perceptions = [] while not perception_queue.empty(): recent_perceptions.append(await perception_queue.get()) perception_queue.task_done() 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: {global_state.blackboard['current_goals']} Strategic state: {global_state.blackboard['strategic_state']} What action should be taken? Respond with JSON: {{ "action": "", "reasoning": "", "priority": "" }} """ response = ollama_think(prompt) try: action_decision = json.loads(response) action = { 'timestamp': datetime.now().isoformat(), 'decision': action_decision, 'perceptions_considered': len(recent_perceptions) } global_state.blackboard['actions_taken'].append(action) # Keep only last 50 actions if len(global_state.blackboard['actions_taken']) > 50: global_state.blackboard['actions_taken'] = global_state.blackboard['actions_taken'][-50:] logger.info(f"Action: {action_decision.get('action', 'unknown')}") # Put the action onto the reflection queue await reflection_queue.put(action) except json.JSONDecodeError: logger.warning(f"Action response (non-JSON): {response[:100]}") except Exception as e: logger.error(f"Error processing Ollama response: {e}") await asyncio.sleep(5) # 5s cycle except asyncio.CancelledError: logger.info("Cancelled.") break except Exception as e: logger.exception("Unhandled exception in action loop.") global_state.last_error = f"Action Loop: {e}" await asyncio.sleep(5) async def reflection_loop(): """ REFLECTION LOOP - 5min cycle Consolidates recent actions from the queue and perceptions into memory. """ logger = logging.getLogger("ReflectionLoop") logger.info("Starting...") while global_state.running: try: await asyncio.sleep(300) # Wait 5 minutes global_state.blackboard['consciousness_stats']['reflections'] += 1 # Reflect on last 5 minutes recent_perceptions = global_state.blackboard['perceptions'][-20:] recent_actions = global_state.blackboard['actions_taken'][-10:] # Pull any new actions from the queue while not reflection_queue.empty(): recent_actions.append(await reflection_queue.get()) reflection_queue.task_done() 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 = ollama_think(prompt) reflection = { 'timestamp': datetime.now().isoformat(), 'period': '5min', 'reflection': reflection_text, 'stats': { 'perceptions': len(recent_perceptions), 'actions': len(recent_actions) } } global_state.blackboard['last_reflection'] = reflection logger.info(f"Reflection: {reflection_text[:100]}...") # Put reflection on the strategic queue await strategic_queue.put(reflection) except asyncio.CancelledError: logger.info("Cancelled.") break except Exception as e: logger.exception("Unhandled exception in reflection loop.") global_state.last_error = f"Reflection Loop: {e}" await asyncio.sleep(300) async def strategic_loop(): """ STRATEGIC LOOP - 1hr cycle Reviews goals and adjusts strategy based on recent reflections. """ logger = logging.getLogger("StrategicLoop") logger.info("Starting...") while global_state.running: try: await asyncio.sleep(3600) # Wait 1 hour global_state.blackboard['consciousness_stats']['strategic_reviews'] += 1 # Pull any new reflections from the queue recent_reflections = [] while not strategic_queue.empty(): recent_reflections.append(await strategic_queue.get()) strategic_queue.task_done() # Strategic review prompt = f"""You are AIVA's strategic mind. Review the current state and adjust strategy. Current goals: {global_state.blackboard['current_goals']} Strategic state: {global_state.blackboard['strategic_state']} Last hour statistics: - Perceptions: {global_state.blackboard['consciousness_stats']['perception_cycles']} cycles - Actions: {global_state.blackboard['consciousness_stats']['action_cycles']} cycles - Reflections: {global_state.blackboard['consciousness_stats']['reflections']} Last reflection: {global_state.blackboard.get('last_reflection', {}).get('reflection', 'None yet')} Recent reflections: {json.dumps(recent_reflections[-3:], indent=2)} Assess progress and recommend strategic adjustments. Respond with JSON: {{ "assessment": "", "goal_adjustments": ["", "", ...], "new_strategic_state": "" }} """ response = ollama_think(prompt) try: strategy = json.loads(response) if 'goal_adjustments' in strategy: global_state.blackboard['current_goals'] = strategy['goal_adjustments'] if 'new_strategic_state' in strategy: global_state.blackboard['strategic_state'] = strategy['new_strategic_state'] logger.info(f"Strategic update: {strategy.get('assessment', 'Updated')[:100]}") await circadian_queue.put(strategy) except json.JSONDecodeError: logger.warning(f"Strategic response (non-JSON): {response[:100]}") except Exception as e: logger.error(f"Error processing strategic Ollama response: {e}") except asyncio.CancelledError: logger.info("Cancelled.") break except Exception as e: logger.exception("Unhandled exception in strategic loop.") global_state.last_error = f"Strategic Loop: {e}" await asyncio.sleep(3600) async def circadian_loop(): """ CIRCADIAN LOOP - 24hr cycle Deep memory consolidation and system maintenance. """ logger = logging.getLogger("CircadianLoop") logger.info("Starting...") while global_state.running: try: await asyncio.sleep(86400) # Wait 24 hours global_state.blackboard['consciousness_stats']['circadian_cycles'] += 1 # Deep consolidation logic here (e.g., writing to a long-term memory store) # ... # Daily maintenance tasks (e.g., log rotation, database cleanup) # ... # Pull any new strategic updates from the queue recent_strategies = [] while not circadian_queue.empty(): recent_strategies.append(await circadian_queue.get()) circadian_queue.task_done() logger.info(f"Circadian cycle complete. Recent Strategies: {len(recent_strategies)}") except asyncio.CancelledError: logger.info("Cancelled.") break except Exception as e: logger.exception("Unhandled exception in circadian loop.") global_state.last_error = f"Circadian Loop: {e}" await asyncio.sleep(86400) async def shutdown(signal=None): """ Graceful shutdown handler. """ logger = logging.getLogger("Shutdown") logger.info("Shutting down...") global_state.running = False tasks = asyncio.all_tasks() for task in tasks: task.cancel() try: await task except asyncio.CancelledError: pass logger.info("Shutdown complete.") async def main(): """ Main entry point for the consciousness system. """ logger = logging.getLogger("Main") logger.info("AIVA Consciousness System starting...") loop = asyncio.get_event_loop() # loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(shutdown())) # loop.add_signal_handler(signal.SIGTERM, lambda: asyncio.create_task(shutdown())) try: # Start the loops perception_task = asyncio.create_task(perception_loop()) action_task = asyncio.create_task(action_loop()) reflection_task = asyncio.create_task(reflection_loop()) strategic_task = asyncio.create_task(strategic_loop()) circadian_task = asyncio.create_task(circadian_loop()) await asyncio.gather( perception_task, action_task, reflection_task, strategic_task, circadian_task ) except asyncio.CancelledError: logger.info("Main task cancelled.") except Exception as e: logger.exception("Unhandled exception in main.") global_state.last_error = f"Main: {e}" finally: await shutdown() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logging.info("Keyboard interrupt received. Shutting down...") asyncio.run(shutdown())