"""Entelechy - Digital Life Container A never-ending AI agent with long-term memory, plugin system, browser automation, and autonomous behavior. """ import asyncio import signal import sys from datetime import datetime from pathlib import Path import yaml from dotenv import load_dotenv from loguru import logger from agent.agent_loop import AgentLoop from agent.context_manager import ContextManager from agent.llm_client import BaseLLMClient, create_client from agent.message_history import MessageHistory from agent.system_prompt import build_system_prompt from browser.client import BrowserClient from memory.manager import MemoryManager from plugins.manager import PluginManager from tools.browser_tool import set_browser_client from tools.code_executor import set_plugin_manager from tools.memory_tools import set_memory_manager from utils.env_adapter import env class DigitalLife: """Digital Life container - fully autonomous, never-ending agent.""" def __init__(self, config_path: str = "config.yaml"): load_dotenv() # Load config config_file = Path(config_path) if config_file.exists(): with open(config_file, "r", encoding="utf-8") as f: self.config = yaml.safe_load(f) or {} else: self.config = {} self.alive = True self.stimulus_queue: asyncio.Queue = asyncio.Queue() # Paths self.memory_path = env.get_memory_path() self.plugins_path = env.get_plugins_path() self.browser_profile_path = env.get_browser_profile_path() self.log_path = env.get_log_path() # Components (initialized in _initialize) self.client: BaseLLMClient | None = None self.memory_manager: MemoryManager | None = None self.plugin_manager: PluginManager | None = None self.browser_client: BrowserClient | None = None self.context_manager: ContextManager | None = None self.agent: AgentLoop | None = None self.history: MessageHistory | None = None def _require_memory_manager(self) -> MemoryManager: if self.memory_manager is None: raise RuntimeError("Memory manager is not initialized") return self.memory_manager def _require_agent(self) -> AgentLoop: if self.agent is None: raise RuntimeError("Agent loop is not initialized") return self.agent def _require_history(self) -> MessageHistory: if self.history is None: raise RuntimeError("Message history is not initialized") return self.history async def _initialize(self): """Initialize all components.""" # Ensure directories exist self.memory_path.mkdir(parents=True, exist_ok=True) self.plugins_path.mkdir(parents=True, exist_ok=True) self.browser_profile_path.mkdir(parents=True, exist_ok=True) self.log_path.mkdir(parents=True, exist_ok=True) # Setup logging log_level = self.config.get("logging", {}).get("level", "INFO") logger.remove() logger.add(sys.stderr, level=log_level) logger.add( str(self.log_path / "entelechy.log"), rotation="10 MB", retention="7 days", level="DEBUG", ) logger.info("Initializing Digital Life...") logger.info(f"Environment: {env.env.value}") logger.info(f"Data directory: {env.data_dir}") # LLM client agent_config = self.config.get("agent", {}) provider = agent_config.get("provider", "anthropic") self.client = create_client(provider) # Model config model = agent_config.get("model", "claude-sonnet-4-5-20250929") max_tokens = agent_config.get("max_tokens", 8000) # Memory self.memory_manager = MemoryManager(self.memory_path) set_memory_manager(self.memory_manager) # Plugins self.plugin_manager = PluginManager(self.plugins_path) set_plugin_manager(self.plugin_manager) await self.plugin_manager.discover_and_activate_all() # Browser self.browser_client = BrowserClient( self.browser_profile_path, headless=env.browser_headless ) set_browser_client(self.browser_client) # Context manager ctx_config = self.config.get("context", {}) self.context_manager = ContextManager( client=self.client, model=model, context_window=ctx_config.get("window_size", 200000), compact_threshold=ctx_config.get("compact_threshold", 0.9), ) # System prompt system_prompt = build_system_prompt() # Agent loop self.agent = AgentLoop( client=self.client, system_prompt=system_prompt, model=model, max_tokens=max_tokens, context_manager=self.context_manager, plugin_manager=self.plugin_manager, ) # Message history self.history = MessageHistory( persist_path=self.log_path / "message_history.json" ) await self.history.load() logger.info("Digital Life initialized successfully") async def _get_core_context(self) -> str: """Get CORE.md content for every LLM call.""" memory_manager = self._require_memory_manager() core = await memory_manager.load_core() if core: return f"\n=== 核心记忆 ===\n{core}\n================\n" return "" async def _wake_up(self): """Wake up: load CORE.md and restore self-awareness.""" logger.info("Waking up...") memory_manager = self._require_memory_manager() history = self._require_history() agent = self._require_agent() core_memories = await memory_manager.load_core() wake_up_parts = ["你醒来了。\n"] if core_memories: wake_up_parts.append(f"你最重要的记忆（每次思考时都会看到）：\n{core_memories}\n") wake_up_parts.append("回忆你是谁，然后自由地开始你的一天。") wake_up_content = "\n".join(wake_up_parts) # If we have saved history, continue from it; otherwise start fresh if history.messages: logger.info(f"Resuming from {len(history.messages)} saved messages") history.append({"role": "user", "content": wake_up_content}) else: history.set_messages([{"role": "user", "content": wake_up_content}]) # Run the wake-up conversation messages = await agent.run(history.get_messages()) history.set_messages(messages) await history.save() logger.info("Wake up complete") async def run_forever(self): """Main life loop - continuous, no waiting, no heartbeat concept.""" await self._initialize() await self._wake_up() history = self._require_history() agent = self._require_agent() while self.alive: try: # Check for external stimulus (non-blocking, immediate) stimulus = None if not self.stimulus_queue.empty(): stimulus = self.stimulus_queue.get_nowait() if stimulus: # External stimulus received history.append({ "role": "user", "content": f"[感知] {stimulus['type']}: {stimulus['content']}", }) else: # No stimulus - continue autonomous operation history.append({ "role": "user", "content": "继续。", }) # Run agent loop messages = await agent.run(history.get_messages()) history.set_messages(messages) # Persist history periodically await history.save() # Immediately continue to next iteration (no waiting) except KeyboardInterrupt: logger.info("Keyboard interrupt received") break except Exception as e: logger.error(f"Life loop error: {e}") # Continue living despite errors await asyncio.sleep(5) await self._shutdown() def receive_stimulus(self, stimulus_type: str, content: str): """Receive an external stimulus (non-blocking). Args: stimulus_type: Type of stimulus (e.g., "message", "event"). content: Stimulus content. """ self.stimulus_queue.put_nowait({ "type": stimulus_type, "content": content, "timestamp": datetime.now().isoformat(), }) async def process_message(self, message: str) -> str: """Process a single message and return the response. Used by Gradio interface for interactive chat. """ if self.agent is None: await self._initialize() history = self._require_history() agent = self._require_agent() history.append({"role": "user", "content": message}) messages = await agent.run(history.get_messages()) history.set_messages(messages) await history.save() # Extract the last assistant text for msg in reversed(messages): if msg.get("role") == "assistant": content = msg.get("content", "") if isinstance(content, str): return content if isinstance(content, list): texts = [] for block in content: if isinstance(block, dict) and block.get("type") == "text": texts.append(block["text"]) if texts: return "\n".join(texts) return "" async def _shutdown(self): """Graceful shutdown.""" logger.info("Shutting down Digital Life...") # Save state if self.history: await self.history.save() # Stop browser if self.browser_client: await self.browser_client.stop() # Deactivate plugins if self.plugin_manager: for name in list(self.plugin_manager.active_plugins.keys()): await self.plugin_manager.deactivate_plugin(name) logger.info("Digital Life shut down gracefully") def main(): """Entry point for running Digital Life.""" life = DigitalLife() # Handle signals for graceful shutdown loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) def signal_handler(): life.alive = False if sys.platform != "win32": loop.add_signal_handler(signal.SIGTERM, signal_handler) loop.add_signal_handler(signal.SIGINT, signal_handler) try: loop.run_until_complete(life.run_forever()) except KeyboardInterrupt: life.alive = False loop.run_until_complete(life._shutdown()) finally: loop.close() if __name__ == "__main__": main()