Update modules/main_fusion_public.py

modules/main_fusion_public.py

@@ -1,156 +1,128 @@
-# FILE: modules/main_fusion_public.py (True Fusion, Verified Launch Version)
+# FILE: modules/main_fusion_public.py (Hugging Face Demo - AGI-Tier Upgrade)
 
 import asyncio
 import logging
+import json
 from typing import Dict, Any, Optional, List
 
-# These imports will now work correctly with the verified config file
 try:
     from modules.config_settings_public import (
-        MODEL_PATHS,
-        MODEL_SPECIFIC_PARAMS,
-        INFERENCE_PRESETS,
-        DEFAULT_INFERENCE_PRESET,
-        MODEL_ROLES,
-        MODEL_ROLE_SYSTEM_PROMPTS,
-        DEFAULT_SYSTEM_PROMPT
+        MODEL_PATHS, MODEL_SPECIFIC_PARAMS, INFERENCE_PRESETS,
+        DEFAULT_INFERENCE_PRESET, MODEL_ROLES, MODEL_ROLE_SYSTEM_PROMPTS, DEFAULT_SYSTEM_PROMPT
     )
     from llama_cpp import Llama
     LLAMA_CPP_AVAILABLE = True
 except ImportError as e:
-    # This is a critical failure, the app cannot run without these.
     logging.basicConfig(level=logging.CRITICAL)
-    logging.critical(f"CRITICAL IMPORT ERROR in main_fusion_public.py: {e}. ZOTHEOS cannot function.")
+    logging.critical(f"CRITICAL HF IMPORT ERROR: {e}.")
     LLAMA_CPP_AVAILABLE = False
 
+logger = logging.getLogger("ZOTHEOS_MainFusion_HF")
 
-logger = logging.getLogger("ZOTHEOS_MainFusion")
-if not logger.handlers:
-    # Configure logger if it hasn't been configured yet
-    handler = logging.StreamHandler()
-    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - [%(funcName)s] - %(message)s')
-    handler.setFormatter(formatter)
-    logger.addHandler(handler)
-    logger.setLevel(logging.INFO)
-
-
+# --- ✅ Simple In-Memory MemoryBank for the Web Demo ---
+class MemoryBank:
+    """A simple, non-persistent memory bank for the web demo."""
+    def __init__(self):
+        self.memories: List[Dict[str, Any]] = []
+    async def store_memory_async(self, query: str, response: str, metadata: Optional[Dict[str, Any]] = None):
+        entry = {'query': query, 'response': response, 'metadata': metadata or {}}
+        self.memories.append(entry)
+        if len(self.memories) > 10: self.memories.pop(0) # Keep memory from growing too large
+    async def retrieve_recent_memories_async(self, limit: int = 5) -> List[Dict[str, Any]]:
+        return self.memories[-limit:]
+    async def get_all_memories_for_export_async(self) -> List[Dict[str, Any]]:
+        return list(self.memories) # Return a copy
+
+# --- ✅ The AGI-Tier Engine, Optimized for Web ---
 class MainFusionPublic:
-    """
-    The core engine for ZOTHEOS. It loads multiple AI models, queries them in parallel
-    with distinct persona-based prompts, and then synthesizes their responses into a
-    higher-order summary.
-    """
     def __init__(self):
-        logger.info("🚀 ZOTHEOS MainFusion Initializing (True Fusion Mode)...")
+        logger.info("🚀 ZOTHEOS HF Demo Engine Initializing...")
         if not LLAMA_CPP_AVAILABLE:
-            raise RuntimeError("Llama.cpp backend is not available. ZOTHEOS cannot start.")
+            raise RuntimeError("Llama.cpp backend is not available.")
 
         self.models: Dict[str, Optional[Llama]] = {}
-        self.config = {
-            "MODEL_ROLES": MODEL_ROLES,
-            "MODEL_ROLE_SYSTEM_PROMPTS": MODEL_ROLE_SYSTEM_PROMPTS,
-        }
-        self.models_last_queried_for_perspectives: List[str] = []
-
+        self.config = {"MODEL_ROLES": MODEL_ROLES, "MODEL_ROLE_SYSTEM_PROMPTS": MODEL_ROLE_SYSTEM_PROMPTS}
+        self.models_last_queried: List[str] = []
+        self.memory_bank = MemoryBank()
         self._initialize_models()
 
     def _initialize_models(self):
-        """Loads all models defined in the config into memory."""
-        for model_name, model_path in MODEL_PATHS.items():
+        for name, path in MODEL_PATHS.items():
             try:
-                logger.info(f"Loading model: {model_name}...")
-                params = MODEL_SPECIFIC_PARAMS['_default'].copy()
-                params.update(MODEL_SPECIFIC_PARAMS.get(model_name, {}))
-                
-                self.models[model_name] = Llama(model_path=model_path, **params)
-                logger.info(f"✅ Model '{model_name}' loaded successfully.")
+                logger.info(f"HF Demo: Loading model {name}...")
+                params = MODEL_SPECIFIC_PARAMS.get('_default', {}).copy()
+                params.update(MODEL_SPECIFIC_PARAMS.get(name, {}))
+                self.models[name] = Llama(model_path=path, **params)
+                logger.info(f"✅ HF Demo: Model '{name}' loaded.")
             except Exception as e:
-                logger.error(f"❌ Failed to load model '{model_name}': {e}", exc_info=True)
-                self.models[model_name] = None
-
-    async def _get_single_perspective(self, model_name: str, query: str) -> str:
-        """Queries a single model with its assigned role and returns the response."""
-        if model_name not in self.models or self.models[model_name] is None:
-            logger.warning(f"Model '{model_name}' is not loaded or failed to initialize.")
-            return f"[Error: The '{model_name}' AI core is offline.]"
+                logger.error(f"❌ HF Demo: Failed to load model '{name}': {e}", exc_info=True)
+                self.models[name] = None
 
-        role = self.config["MODEL_ROLES"].get(model_name, "general")
-        system_prompt = self.config["MODEL_ROLE_SYSTEM_PROMPTS"].get(role, DEFAULT_SYSTEM_PROMPT)
+    async def _get_perspective(self, model_name: str, query: str, system_prompt: str) -> str:
+        llm = self.models.get(model_name)
+        if not llm: return f"[Error: '{model_name}' core is offline.]"
         
-        llm = self.models[model_name]
         messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": query}]
         preset = INFERENCE_PRESETS.get(DEFAULT_INFERENCE_PRESET, {})
-
+        
         try:
-            logger.info(f"Querying {model_name} ({role})...")
-            # Run the blocking call in a separate thread to not block the event loop
             response = await asyncio.to_thread(llm.create_chat_completion, messages=messages, **preset)
-            content = response['choices'][0]['message']['content'].strip()
-            logger.info(f"✅ Response received from {model_name}.")
-            return content
+            return response['choices'][0]['message']['content'].strip()
         except Exception as e:
-            logger.error(f"Error during inference for model '{model_name}': {e}", exc_info=True)
-            return f"[Error during inference for '{model_name}'. See logs for details.]"
-
-    async def _generate_synthesis(self, query: str, perspectives: Dict[str, str]) -> str:
-        """Uses a designated model to synthesize a summary from all perspectives."""
-        summarizer_model = "gemma"  # Gemma is good for this task
-        if summarizer_model not in self.models or self.models[summarizer_model] is None:
-            logger.warning(f"Summarizer model '{summarizer_model}' not available. Returning basic summary.")
-            return "Multiple perspectives were gathered, but the synthesis AI core is offline. Please review the detailed perspectives below."
-
-        valid_perspectives = {name: text for name, text in perspectives.items() if not text.startswith("[Error")}
-        if not valid_perspectives:
-            return "No valid perspectives could be generated to create a summary."
-        if len(valid_perspectives) == 1:
-            return next(iter(valid_perspectives.values()))
-
-        synthesis_prompt = f"Original User Question: \"{query}\"\n\nSynthesize the following distinct perspectives into a single, cohesive, and insightful summary. Capture the core agreements, disagreements, and unique insights from each viewpoint. Your goal is to provide a higher-order understanding that integrates these different analyses.\n\n"
-        for name, text in valid_perspectives.items():
+            logger.error(f"HF Demo Inference error for '{model_name}': {e}")
+            return f"[Error during inference for '{model_name}'.]"
+
+    async def _generate_true_synthesis(self, query: str, perspectives: Dict[str, str]) -> str:
+        summarizer_model = "gemma"
+        if summarizer_model not in self.models or not self.models[summarizer_model]:
+            return "Synthesis AI core is offline."
+
+        valid_perspectives = {k: v for k, v in perspectives.items() if not v.startswith("[Error")}
+        if not valid_perspectives: return "No valid perspectives were generated."
+
+        synthesis_prompt = f"""
+        As the master intelligence of ZOTHEOS, perform a high-level synthesis of these perspectives on the user's query: "{query}".
+        Your goal is to find the deeper truth. Analyze the viewpoints to:
+        1. Identify the core theme.
+        2. Highlight the most significant tension between them.
+        3. Extract a unique "aha!" insight that emerges from their combination.
+        4. Conclude with a definitive, balanced statement.
+
+        Perspectives:
+        {json.dumps(valid_perspectives, indent=2)}
+
+        Your Final Synthesized Insight:
+        """
+        system_prompt = "You are a master synthesis AI. Create a clear, insightful summary from the provided texts."
+        return await self._get_perspective(summarizer_model, synthesis_prompt, system_prompt)
+
+    def _format_output(self, summary: str, perspectives: Dict[str, str]) -> str:
+        output = f"## ✨ ZOTHEOS Final Synthesized Insight ✨\n\n{summary}\n\n### 💬 Detailed Individual Perspectives\n\n"
+        for name, text in perspectives.items():
             role = self.config["MODEL_ROLES"].get(name, "General")
-            synthesis_prompt += f"--- PERSPECTIVE FROM {name.upper()} ({role.capitalize()}) ---\n{text}\n\n"
-        synthesis_prompt += "--- SYNTHESIZED INSIGHT ---\n"
-        
-        summary_system_prompt = "You are a master synthesis AI. Your role is to create a clear and insightful summary from the provided texts, acting as a final arbiter of truth."
-        return await self._get_single_perspective(summarizer_model, synthesis_prompt)
-
-    def _format_final_output(self, summary: str, perspectives: Dict[str, str]) -> str:
-        """Formats the final Markdown output for the Gradio interface."""
-        output = f"## ✨ ZOTHEOS Final Synthesized Insight ✨\n\n{summary}\n\n---\n\n### 💬 Detailed Individual Perspectives\n\n"
-        for model_name, text in perspectives.items():
-            role = self.config["MODEL_ROLES"].get(model_name, "General")
-            output += f"**Perspective from {model_name.capitalize()} ({role.capitalize()}):**\n{text}\n\n"
+            output += f"**Perspective from {name.capitalize()} ({role.capitalize()}):**\n{text}\n\n"
         return output.strip()
 
-    async def process_query_with_fusion(self, query: str, **kwargs):
-        """The main entry point for processing a user query with True Fusion."""
-        if not all(self.models.values()):
-            return "[Critical Error: Not all AI cores are online. Fusion is not possible.]"
+    async def process_query_with_fusion(self, query: str, **kwargs) -> str:
+        online_models = [name for name, model in self.models.items() if model]
+        if not online_models: return "[Critical Error: All AI cores are offline.]"
+        self.models_last_queried = online_models
 
-        # For now, we use all configured models. Tier logic can be re-added here.
-        models_to_query = list(self.models.keys())
-        self.models_last_queried_for_perspectives = models_to_query
+        tasks = {name: self._get_perspective(
+            name, query, 
+            self.config["MODEL_ROLE_SYSTEM_PROMPTS"].get(self.config["MODEL_ROLES"].get(name, "general"), DEFAULT_SYSTEM_PROMPT)
+        ) for name in online_models}
         
-        logger.info(f"Initiating True Fusion for query: '{query[:60]}...' using models: {models_to_query}")
-        
-        # Create and run all perspective-gathering tasks in parallel
-        perspective_tasks = [self._get_single_perspective(model_name, query) for model_name in models_to_query]
-        gathered_responses = await asyncio.gather(*perspective_tasks)
-        perspectives = dict(zip(models_to_query, gathered_responses))
+        responses = await asyncio.gather(*tasks.values())
+        perspectives = dict(zip(tasks.keys(), responses))
+
+        synthesis = await self._generate_true_synthesis(query, perspectives)
+        final_output = self._format_output(synthesis, perspectives)
 
-        # Generate the final synthesis based on the gathered perspectives
-        logger.info("All perspectives gathered. Generating final synthesis...")
-        synthesis = await self._generate_synthesis(query, perspectives)
+        if self.memory_bank:
+            await self.memory_bank.store_memory_async(query=query, response=final_output)
 
-        # Format the final output for the user
-        logger.info("Synthesis complete. Formatting final output.")
-        return self._format_final_output(synthesis, perspectives)
+        return final_output
 
     async def get_status_report(self) -> Dict[str, Any]:
-        """Provides a status report for the system."""
-        return {
-            "status": "Online - True Fusion Mode",
-            "loaded_models": [name for name, model in self.models.items() if model is not None],
-            "failed_models": [name for name, model in self.models.items() if model is None],
-            "last_queried_for_fusion": self.models_last_queried_for_perspectives,
-        }
+        return {"status": "Online - Web Demo Mode", "loaded_models": [name for name, model in self.models.items() if model is not None]}