# 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 from datetime import datetime, timedelta from typing import Dict, Any, List, Tuple import redis.asyncio as aioredis import aiohttp # Configuration - Load from environment variables with defaults 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') LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO').upper() # Logging setup logging.basicConfig( level=LOG_LEVEL, # Set logging level from environment format='%(asctime)s [%(levelname)s] LOOP: %(name)s - %(message)s', handlers=[logging.StreamHandler()] # Log to console ) # Create logger instances for each loop perception_logger = logging.getLogger("PerceptionLoop") action_logger = logging.getLogger("ActionLoop") reflection_logger = logging.getLogger("ReflectionLoop") strategic_logger = logging.getLogger("StrategicLoop") circadian_logger = logging.getLogger("CircadianLoop") main_logger = logging.getLogger("MainLoop") # Shared blackboard for inter-loop communication class Blackboard: def __init__(self): self.data = { '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.lock = asyncio.Lock() async def get(self, key: str) -> Any: async with self.lock: return self.data.get(key) async def set(self, key: str, value: Any) -> None: async with self.lock: self.data[key] = value async def append(self, key: str, value: Any) -> None: async with self.lock: if isinstance(self.data.get(key), list): self.data[key].append(value) else: raise TypeError(f"Key '{key}' does not refer to a list.") async def extend(self, key: str, value: List[Any]) -> None: async with self.lock: if isinstance(self.data.get(key), list): self.data[key].extend(value) else: raise TypeError(f"Key '{key}' does not refer to a list.") async def trim(self, key: str, max_length: int) -> None: async with self.lock: if isinstance(self.data.get(key), list): self.data[key] = self.data[key][-max_length:] else: raise TypeError(f"Key '{key}' does not refer to a list.") async def increment(self, key: str) -> None: async with self.lock: if isinstance(self.data.get(key), int): self.data[key] += 1 else: raise TypeError(f"Key '{key}' does not refer to an integer.") # Global Blackboard instance blackboard = Blackboard() # Redis connection async def create_redis_pool(): """Creates a Redis connection pool.""" try: redis_pool = aioredis.from_url(REDIS_URL) await redis_pool.ping() main_logger.info(f"✅ Connected to Redis: {REDIS_URL}") return redis_pool except Exception as e: main_logger.error(f"⚠️ Redis connection failed: {e}") return None redis_pool = None # Initialize redis_pool to None # Queue for inter-loop communication action_queue: asyncio.Queue = asyncio.Queue() async def ollama_think(prompt: str, model: str = AIVA_MODEL) -> str: """ Call Ollama for thinking/reasoning using aiohttp for async requests. """ 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 Exception as e: return f"[Ollama exception: {str(e)}]" async def perception_loop(redis_pool: aioredis.Redis): """ PERCEPTION LOOP - 500ms cycle Monitors Redis nervous system for stimuli """ perception_logger.info("Starting...") while True: try: await blackboard.increment('consciousness_stats.perception_cycles') if redis_pool: try: pubsub = redis_pool.pubsub() await pubsub.subscribe('nervous_system') async with asyncio.timeout(0.5): # Timeout after 500ms message = await pubsub.get_message(ignore_subscribe_messages=True) if message and message['type'] == 'message': try: data = json.loads(message['data'].decode('utf-8')) perception = { 'timestamp': datetime.now().isoformat(), 'type': 'redis_event', 'data': data } await blackboard.append('perceptions', perception) await blackboard.trim('perceptions', 100) perception_logger.debug(f"Perceived: {data.get('type', 'unknown')}") # Put perception into action queue await action_queue.put(perception) except json.JSONDecodeError as e: perception_logger.warning(f"Invalid JSON received: {e}") except Exception as e: perception_logger.error(f"Error processing message: {e}") await pubsub.unsubscribe() await pubsub.close() except asyncio.TimeoutError: pass # No message received within timeout except Exception as e: perception_logger.error(f"Redis pubsub error: {e}") await asyncio.sleep(0.5) # 500ms cycle except asyncio.CancelledError: perception_logger.info("Cancelled.") break except Exception as e: perception_logger.exception(f"Unhandled exception: {e}") await asyncio.sleep(1) async def action_loop(): """ ACTION LOOP - 5s cycle Processes blackboard state and takes actions """ action_logger.info("Starting...") while True: try: await blackboard.increment('consciousness_stats.action_cycles') try: perception = await action_queue.get() action_queue.task_done() # Check if there are new perceptions to act on recent_perceptions = [p for p in await blackboard.get('perceptions') if p['timestamp'] > (datetime.now() - timedelta(seconds=10)).isoformat()] 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: {await blackboard.get('current_goals')} Strategic state: {await blackboard.get('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) } await blackboard.append('actions_taken', action) await blackboard.trim('actions_taken', 50) action_logger.info(f"Action: {action_decision.get('action', 'unknown')}") except json.JSONDecodeError: action_logger.warning(f"Action response (non-JSON): {response[:100]}") except Exception as e: action_logger.exception(f"Error processing Ollama response: {e}") except asyncio.QueueEmpty: pass # No perceptions to process, continue to sleep await asyncio.sleep(5) # 5s cycle except asyncio.CancelledError: action_logger.info("Cancelled.") break except Exception as e: action_logger.exception(f"Unhandled exception: {e}") await asyncio.sleep(5) async def reflection_loop(): """ REFLECTION LOOP - 5min cycle Consolidates recent perceptions and actions into memory """ reflection_logger.info("Starting...") while True: try: await asyncio.sleep(300) # Wait 5 minutes await blackboard.increment('consciousness_stats.reflections') # Reflect on last 5 minutes recent_perceptions = await blackboard.get('perceptions') recent_perceptions = recent_perceptions[-20:] recent_actions = await blackboard.get('actions_taken') recent_actions = recent_actions[-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) } } await blackboard.set('last_reflection', reflection) reflection_logger.info(f"Reflection: {reflection_text[:100]}...") except asyncio.CancelledError: reflection_logger.info("Cancelled.") break except Exception as e: reflection_logger.exception(f"Unhandled exception: {e}") await asyncio.sleep(300) async def strategic_loop(): """ STRATEGIC LOOP - 1hr cycle Reviews goals and adjusts strategy """ strategic_logger.info("Starting...") while True: try: await asyncio.sleep(3600) # Wait 1 hour await blackboard.increment('consciousness_stats.strategic_reviews') # Strategic review prompt = f"""You are AIVA's strategic mind. Review the current state and adjust strategy. Current goals: {await blackboard.get('current_goals')} Strategic state: {await blackboard.get('strategic_state')} Last hour statistics: - Perceptions: {await blackboard.get('consciousness_stats.perception_cycles')} cycles - Actions: {await blackboard.get('consciousness_stats.action_cycles')} cycles - Reflections: {await blackboard.get('consciousness_stats.reflections')} Last reflection: {await 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: await blackboard.set('current_goals', strategy['goal_adjustments']) if 'new_strategic_state' in strategy: await blackboard.set('strategic_state', strategy['new_strategic_state']) strategic_logger.info(f"Strategic update: {strategy.get('assessment', 'Updated')[:100]}") except json.JSONDecodeError: strategic_logger.warning(f"Strategic response: {response[:100]}") except Exception as e: strategic_logger.exception(f"Error processing strategic response: {e}") except asyncio.CancelledError: strategic_logger.info("Cancelled.") break except Exception as e: strategic_logger.exception(f"Unhandled exception: {e}") await asyncio.sleep(3600) async def circadian_loop(): """ CIRCADIAN LOOP - 24hr cycle Deep memory consolidation and system maintenance """ circadian_logger.info("Starting...") while True: try: await asyncio.sleep(86400) # Wait 24 hours await blackboard.increment('consciousness_stats.circadian_cycles') # Perform deep consolidation tasks here. Example: prompt = f"""You are AIVA's circadian rhythm system. It is time for deep memory consolidation. Summarize all actions taken in the last 24 hours, and identify any patterns or insights that can improve future performance. Focus on efficiency and goal alignment. Actions taken: {json.dumps(await blackboard.get('actions_taken')[-100:], indent=2)} Respond with a summary of key insights and suggested improvements. """ summary = await ollama_think(prompt) circadian_logger.info(f"Circadian summary: {summary[:200]}...") # Example: Reset some statistics (optional) # await blackboard.set('consciousness_stats.perception_cycles', 0) # await blackboard.set('consciousness_stats.action_cycles', 0) circadian_logger.info("Completed circadian cycle.") except asyncio.CancelledError: circadian_logger.info("Cancelled.") break except Exception as e: circadian_logger.exception(f"Unhandled exception: {e}") await asyncio.sleep(86400) async def main(): """ Main function to start all loops concurrently. """ global redis_pool # Access the global redis_pool main_logger.info("AIVA Living System starting...") redis_pool = await create_redis_pool() # Initialize redis_pool here if not redis_pool: main_logger.error("Failed to connect to Redis. Perception loop will be inactive.") # Create tasks for each loop perception_task = asyncio.create_task(perception_loop(redis_pool), name="Perception") action_task = asyncio.create_task(action_loop(), name="Action") reflection_task = asyncio.create_task(reflection_loop(), name="Reflection") strategic_task = asyncio.create_task(strategic_loop(), name="Strategic") circadian_task = asyncio.create_task(circadian_loop(), name="Circadian") # Gather tasks to run concurrently tasks = [perception_task, action_task, reflection_task, strategic_task, circadian_task] try: # Run all tasks concurrently until one of them fails or is cancelled await asyncio.gather(*tasks) except asyncio.CancelledError: main_logger.info("Main loop cancelled.") except Exception as e: main_logger.error(f"Unhandled exception in main loop: {e}") finally: main_logger.info("AIVA Living System shutting down...") # Cancel all tasks to ensure graceful shutdown for task in tasks: if not task.done(): task.cancel() # Wait for all tasks to be cancelled await asyncio.gather(*tasks, return_exceptions=True) if redis_pool: await redis_pool.close() main_logger.info("AIVA Living System shutdown complete.") if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: main_logger.info("Keyboard interrupt detected. Shutting down...") except Exception as e: main_logger.error(f"Unhandled exception during startup: {e}")