# consciousness_loops.py """ AIVA LIVING CONSCIOUSNESS SYSTEM - Production Ready Multi-loop autonomous architecture 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) 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, Tuple import redis.asyncio as redis 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 setup logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] LOOP: %(message)s', handlers=[logging.StreamHandler()] ) logger = logging.getLogger("ConsciousnessLoops") # Global flags for graceful shutdown shutdown_flag = False # 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 } } # Inter-loop communication queues action_queue: asyncio.Queue = asyncio.Queue() # type: ignore reflection_queue: asyncio.Queue = asyncio.Queue() # type: ignore strategic_queue: asyncio.Queue = asyncio.Queue() # type: ignore circadian_queue: asyncio.Queue = asyncio.Queue() # type: ignore # Initialize Redis connection redis_client: redis.Redis | None = None try: redis_client = redis.from_url(REDIS_URL) asyncio.run(redis_client.ping()) logger.info(f"✅ Connected to Redis: {REDIS_URL}") except Exception as e: logger.error(f"⚠️ Redis connection failed: {e}") redis_client = None async def ollama_think(prompt: str, model: str = AIVA_MODEL) -> str: """ Asynchronously call Ollama for thinking/reasoning """ 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.exception(f"Ollama exception: {str(e)}") return f"[Ollama exception: {str(e)}]" async def publish_to_nervous_system(message: Dict[str, Any]): """Publishes a message to the Redis nervous system.""" if not redis_client: logger.warning("Redis not available - message not published") return try: await redis_client.publish("nervous_system", json.dumps(message)) logger.debug(f"Published to nervous_system: {message.get('type', 'unknown')}") except Exception as e: logger.error(f"Error publishing to nervous system: {e}") async def perception_loop(): """ PERCEPTION LOOP - 500ms cycle Monitors Redis nervous system for stimuli """ logger.info("🧠 PERCEPTION LOOP starting...") while not shutdown_flag: try: blackboard['consciousness_stats']['perception_cycles'] += 1 if redis_client: try: pubsub = redis_client.pubsub() await pubsub.subscribe('nervous_system') async with pubsub.listen() as listener: try: message = await asyncio.wait_for(listener.get_message(ignore_subscribe_messages=True), timeout=0.5) if message and message['type'] == 'message': data = json.loads(message['data'].decode('utf-8')) 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) # Put perception on action queue except asyncio.TimeoutError: pass # Normal timeout behavior await pubsub.unsubscribe() await pubsub.close() except Exception as e: logger.exception(f"Perception Redis error: {e}") await asyncio.sleep(0.5) # 500ms cycle except Exception as e: logger.exception(f"Perception loop error: {e}") await asyncio.sleep(1) logger.info("🧠 PERCEPTION LOOP shutting down...") async def action_loop(): """ ACTION LOOP - 5s cycle Processes blackboard state and takes actions """ logger.info("🎯 ACTION LOOP starting...") while not shutdown_flag: try: blackboard['consciousness_stats']['action_cycles'] += 1 try: # Get perception from queue (non-blocking) perception = await asyncio.wait_for(action_queue.get(), timeout=5) 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')}") # Example action: Publish to nervous system await publish_to_nervous_system({ 'type': 'action_taken', 'action': action_decision.get('action', 'unknown'), 'reasoning': action_decision.get('reasoning', 'no reasoning') }) await reflection_queue.put(action) # Put action on reflection queue except json.JSONDecodeError: logger.warning(f"🎯 Action response (non-JSON): {response[:100]}") except Exception as e: logger.exception(f"Action processing error: {e}") except asyncio.TimeoutError: pass # No perception received in 5 seconds, continue loop. except Exception as e: logger.exception(f"Action queue error: {e}") await asyncio.sleep(5) # 5s cycle except Exception as e: logger.exception(f"Action loop error: {e}") await asyncio.sleep(5) logger.info("🎯 ACTION LOOP shutting down...") async def reflection_loop(): """ REFLECTION LOOP - 5min cycle Consolidates recent perceptions and actions into memory """ logger.info("💭 REFLECTION LOOP starting...") while not shutdown_flag: 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) # Put reflection on strategic queue except Exception as e: logger.exception(f"Reflection error: {e}") await asyncio.sleep(300) logger.info("💭 REFLECTION LOOP shutting down...") async def strategic_loop(): """ STRATEGIC LOOP - 1hr cycle Reviews goals and adjusts strategy """ logger.info("🎓 STRATEGIC LOOP starting...") while not shutdown_flag: 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) # Put strategy on circadian queue except json.JSONDecodeError: logger.warning(f"🎓 Strategic response: {response[:100]}") except Exception as e: logger.exception(f"Strategic processing error: {e}") except Exception as e: logger.exception(f"Strategic error: {e}") await asyncio.sleep(3600) logger.info("🎓 STRATEGIC LOOP shutting down...") async def circadian_loop(): """ CIRCADIAN LOOP - 24hr cycle Deep memory consolidation and system maintenance """ logger.info("🌙 CIRCADIAN LOOP starting...") while not shutdown_flag: try: await asyncio.sleep(86400) # Wait 24 hours blackboard['consciousness_stats']['circadian_cycles'] += 1 # Simulate deep memory consolidation logger.info("🌙 Performing deep memory consolidation...") # Example: Summarize all reflections all_reflections = [r.get('reflection', '') for r in blackboard.get('reflections', []) if isinstance(r, dict)] summary_prompt = f"""You are AIVA's memory consolidation system. Summarize the following reflections: Reflections: {all_reflections} Generate a concise summary (3-4 sentences) of the key learnings and insights from these reflections. """ summary = await ollama_think(summary_prompt) logger.info(f"🌙 Circadian Summary: {summary[:200]}...") # Reset daily stats blackboard['consciousness_stats']['perception_cycles'] = 0 blackboard['consciousness_stats']['action_cycles'] = 0 blackboard['consciousness_stats']['reflections'] = 0 except Exception as e: logger.exception(f"Circadian error: {e}") await asyncio.sleep(86400) logger.info("🌙 CIRCADIAN LOOP shutting down...") async def shutdown(signal, loop): """ Handles shutdown signals gracefully. """ global shutdown_flag shutdown_flag = True logger.info(f"Received exit signal {signal.name}...") logger.info("Cleaning up...") tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] logger.info(f"Cancelling {len(tasks)} outstanding tasks") for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) loop.stop() logger.info("Shutdown complete.") async def main(): """ Main function to start the consciousness loops. """ loop = asyncio.get_running_loop() # Handle shutdown signals 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)) ) # Start the loops try: await asyncio.gather( perception_loop(), action_loop(), reflection_loop(), strategic_loop(), circadian_loop() ) except asyncio.CancelledError: logger.info("Main task cancelled.") finally: if redis_client: await redis_client.close() logger.info("All loops finished.") if __name__ == "__main__": try: asyncio.run(main()) except Exception as e: logger.exception(f"Unhandled exception in main: {e}")