# 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 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') # Logging Configuration logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] LOOP: %(name)s - %(message)s', handlers=[logging.StreamHandler()] ) # 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") # Shared blackboard for inter-loop communication class Blackboard: def __init__(self): self.perceptions: List[Dict[str, Any]] = [] self.actions_taken: List[Dict[str, Any]] = [] self.current_goals: List[str] = ['Understand Genesis', 'Process patent knowledge', 'Coordinate MVP'] self.strategic_state: str = 'initializing' self.last_reflection: Dict[str, Any] = {} self.consciousness_stats: Dict[str, int] = { 'perception_cycles': 0, 'action_cycles': 0, 'reflections': 0, 'strategic_reviews': 0, 'circadian_cycles': 0 } self.lock = asyncio.Lock() # Add lock for thread safety async def update(self, key: str, value: Any): async with self.lock: setattr(self, key, value) async def get(self, key: str) -> Any: async with self.lock: return getattr(self, key) # Global Blackboard instance blackboard = Blackboard() # Event Queue for inter-loop communication event_queue: asyncio.Queue = asyncio.Queue() # Redis connection async def create_redis_connection(): try: redis_client = aioredis.from_url(REDIS_URL) await redis_client.ping() perception_logger.info(f"✅ Connected to Redis: {REDIS_URL}") return redis_client except Exception as e: perception_logger.error(f"⚠️ Redis connection failed: {e}") return None async def ollama_think(prompt: str, model: str = AIVA_MODEL) -> str: """ Call 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 Exception as e: return f"[Ollama exception: {str(e)}]" async def perception_loop(redis_client: aioredis.Redis): """ PERCEPTION LOOP - 500ms cycle Monitors Redis nervous system for stimuli """ perception_logger.info("🧠 PERCEPTION LOOP starting...") while True: try: stats = await blackboard.get('consciousness_stats') stats['perception_cycles'] += 1 await blackboard.update('consciousness_stats', stats) if redis_client: try: pubsub = redis_client.pubsub() await pubsub.subscribe('genesis: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': try: data = json.loads(message['data'].decode('utf-8')) perception = { 'timestamp': datetime.now().isoformat(), 'type': 'redis_event', 'data': data } perceptions = await blackboard.get('perceptions') perceptions.append(perception) if len(perceptions) > 100: perceptions = perceptions[-100:] await blackboard.update('perceptions', perceptions) perception_logger.info(f"👁️ Perceived: {data.get('type', 'unknown')}") await event_queue.put(('perception', perception)) # Put perception event on the queue except json.JSONDecodeError as e: perception_logger.warning(f"Non-JSON message received: {message['data'][:100]} - {e}") except Exception as e: perception_logger.error(f"Error processing message: {e}") except asyncio.TimeoutError: pass # No message received within timeout finally: await pubsub.unsubscribe('genesis:nervous_system') await pubsub.close() except Exception as e: perception_logger.error(f"Redis PubSub error: {e}") await asyncio.sleep(0.5) # 500ms cycle except Exception as e: perception_logger.exception(f"⚠️ Perception loop error: {e}") await asyncio.sleep(1) async def action_loop(): """ ACTION LOOP - 5s cycle Processes blackboard state and takes actions """ action_logger.info("🎯 ACTION LOOP starting...") while True: try: stats = await blackboard.get('consciousness_stats') stats['action_cycles'] += 1 await blackboard.update('consciousness_stats', stats) # Check if there are new perceptions to act on perceptions = await blackboard.get('perceptions') recent_perceptions = [p for p in perceptions if p['timestamp'] > (datetime.now() - timedelta(seconds=10)).isoformat()] if recent_perceptions: # Decide on action using Ollama goals = await blackboard.get('current_goals') state = await blackboard.get('strategic_state') 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: {goals} Strategic state: {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) } actions_taken = await blackboard.get('actions_taken') actions_taken.append(action) if len(actions_taken) > 50: actions_taken = actions_taken[-50:] await blackboard.update('actions_taken', actions_taken) action_logger.info(f"🎯 Action: {action_decision.get('action', 'unknown')}") await event_queue.put(('action', action)) # Put action event on the queue except json.JSONDecodeError: action_logger.warning(f"🎯 Action response (non-JSON): {response[:100]}") except Exception as e: action_logger.exception(f"Error processing action decision: {e}") await asyncio.sleep(5) # 5s cycle except Exception as e: action_logger.exception(f"⚠️ Action loop error: {e}") await asyncio.sleep(5) async def reflection_loop(): """ REFLECTION LOOP - 5min cycle Consolidates recent perceptions and actions into memory """ reflection_logger.info("💭 REFLECTION LOOP starting...") while True: try: await asyncio.sleep(300) # Wait 5 minutes stats = await blackboard.get('consciousness_stats') stats['reflections'] += 1 await blackboard.update('consciousness_stats', stats) # Reflect on last 5 minutes perceptions = await blackboard.get('perceptions') actions_taken = await blackboard.get('actions_taken') recent_perceptions = perceptions[-20:] recent_actions = 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) } } await blackboard.update('last_reflection', reflection) reflection_logger.info(f"💭 Reflection: {reflection_text[:100]}...") await event_queue.put(('reflection', reflection)) # Put reflection event on the queue except Exception as e: reflection_logger.exception(f"⚠️ Reflection loop error: {e}") await asyncio.sleep(300) async def strategic_loop(): """ STRATEGIC LOOP - 1hr cycle Reviews goals and adjusts strategy """ strategic_logger.info("🎓 STRATEGIC LOOP starting...") while True: try: await asyncio.sleep(3600) # Wait 1 hour stats = await blackboard.get('consciousness_stats') stats['strategic_reviews'] += 1 await blackboard.update('consciousness_stats', stats) # Strategic review goals = await blackboard.get('current_goals') state = await blackboard.get('strategic_state') last_reflection = await blackboard.get('last_reflection') consciousness_stats = await blackboard.get('consciousness_stats') prompt = f"""You are AIVA's strategic mind. Review the current state and adjust strategy. Current goals: {goals} Strategic state: {state} Last hour statistics: - Perceptions: {consciousness_stats['perception_cycles']} cycles - Actions: {consciousness_stats['action_cycles']} cycles - Reflections: {consciousness_stats['reflections']} Last reflection: {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.update('current_goals', strategy['goal_adjustments']) if 'new_strategic_state' in strategy: await blackboard.update('strategic_state', strategy['new_strategic_state']) strategic_logger.info(f"🎓 Strategic update: {strategy.get('assessment', 'Updated')[:100]}") await event_queue.put(('strategy', strategy)) # Put strategy event on the queue except json.JSONDecodeError: strategic_logger.warning(f"🎓 Strategic response: {response[:100]}") except Exception as e: strategic_logger.exception(f"Error processing strategy response: {e}") except Exception as e: strategic_logger.exception(f"⚠️ Strategic loop error: {e}") await asyncio.sleep(3600) async def circadian_loop(): """ CIRCADIAN LOOP - 24hr cycle Deep memory consolidation and system maintenance """ circadian_logger.info("🌙 CIRCADIAN LOOP starting...") while True: try: await asyncio.sleep(86400) # Wait 24 hours stats = await blackboard.get('consciousness_stats') stats['circadian_cycles'] += 1 await blackboard.update('consciousness_stats', stats) # Deep consolidation (placeholder - expand this) circadian_logger.info("🌙 Performing deep memory consolidation...") # System maintenance (placeholder - add tasks here) circadian_logger.info("🌙 Performing system maintenance...") await event_queue.put(('circadian', 'maintenance')) # Put circadian event on the queue except Exception as e: circadian_logger.exception(f"⚠️ Circadian loop error: {e}") await asyncio.sleep(86400) async def event_handler_task(): """ Handles events published on the event queue """ while True: try: event_type, event_data = await event_queue.get() if event_type == 'perception': perception_logger.debug(f"Handling perception event: {event_data.get('type')}") # Handle perception event (e.g. trigger actions) elif event_type == 'action': action_logger.debug(f"Handling action event: {event_data.get('decision').get('action')}") # Handle action event (e.g. log, trigger feedback) elif event_type == 'reflection': reflection_logger.debug(f"Handling reflection event: {event_data.get('reflection')}") # Handle reflection event (e.g. update long-term memory) elif event_type == 'strategy': strategic_logger.debug(f"Handling strategy event: {event_data.get('assessment')}") # Handle strategy event (e.g. adjust goals) elif event_type == 'circadian': circadian_logger.debug(f"Handling circadian event: {event_data}") # Handle circadian event (e.g. perform backups) else: logging.warning(f"Unknown event type: {event_type}") event_queue.task_done() except Exception as e: logging.error(f"Error handling event: {e}") async def main(): """ Main function to start all loops """ redis_client = await create_redis_connection() # Create and start the tasks tasks = [ asyncio.create_task(perception_loop(redis_client), name="Perception"), asyncio.create_task(action_loop(), name="Action"), asyncio.create_task(reflection_loop(), name="Reflection"), asyncio.create_task(strategic_loop(), name="Strategic"), asyncio.create_task(circadian_loop(), name="Circadian"), asyncio.create_task(event_handler_task(), name="EventHandler") # Start the event handler task ] try: await asyncio.gather(*tasks) except (KeyboardInterrupt, asyncio.CancelledError): logging.info("Shutting down...") finally: # Cancel all tasks for task in tasks: task.cancel() if redis_client: await redis_client.close() await asyncio.gather(*tasks, return_exceptions=True) # Await all tasks to finish logging.info("Shutdown complete.") if __name__ == "__main__": asyncio.run(main())