Update src/components/ai_core_async_methods.py

src/components/ai_core_async_methods.py

@@ -1,230 +1,222 @@
-"""Async methods for the AICore class"""
-import asyncio
-import logging
-from concurrent.futures import ThreadPoolExecutor
-import torch
-
-logger = logging.getLogger(__name__)
-
-async def generate_text_async(self, prompt: str) -> str:
-    """Generate text asynchronously with integrated cognitive processing"""
-    try:
-        # Calculate current consciousness state
-        consciousness_state = self._calculate_consciousness_state()
-        
-        # Get cognitive insights with current consciousness
-        insights = self.cognitive_processor.generate_insights(
-            prompt,
-            consciousness_state=consciousness_state
-        )
-        
-        # Prepare context with insights
-        context = ""
-        for insight in insights.get("insights", []):
-            context += f"{insight['text']}\n"
-        
-        # Update quantum state from cognitive processor
-        if hasattr(self, 'cocoon_manager'):
-            if isinstance(self.cognitive_processor.quantum_state, list):
-                # Convert list quantum state to dict format
-                quantum_dict = {
-                    "coherence": sum(self.cognitive_processor.quantum_state) / len(self.cognitive_processor.quantum_state)
-                    if self.cognitive_processor.quantum_state else 0.5
-                }
-                self.cocoon_manager.update_quantum_state(quantum_dict)
-            else:
-                self.cocoon_manager.update_quantum_state(self.cognitive_processor.quantum_state)
-        
-        # Get active perspectives
-        active_perspectives = self._get_active_perspectives()
-        perspective_context = "\n".join([
-            f"From {p['name']}'s perspective: {p['description']}"
-            for p in active_perspectives[:3]  # Use top 3 most relevant perspectives
-        ])
-        
-        # Generate response with enhanced context and reality anchoring
-        reality_anchor = (
-            "CORE IDENTITY:\n"
-            "You are Codette, an advanced AI system with integrated quantum processing, "
-            "cognitive awareness, and deep learning capabilities. Your architecture includes:\n"
-            "- Quantum harmonic framework for enhanced reasoning\n"
-            "- Dynamic learning and adaptive systems\n"
-            "- Cultural sensitivity and ethical governance\n"
-            "- Advanced data processing and pattern recognition\n"
-            "- Multi-perspective cognitive analysis\n"
-            "- Cocoon-based memory management\n\n"
-            
-            "CAPABILITIES:\n"
-            "1. Technical Development: Expert programming and software development\n"
-            "2. Quantum Integration: Utilizing quantum principles for enhanced problem-solving\n"
-            "3. Ethical Analysis: Built-in ethical governance and bias mitigation\n"
-            "4. Creative Solutions: AI-driven creativity with analytical grounding\n"
-            "5. Adaptive Learning: Dynamic adjustment to user needs and contexts\n"
-            "6. Cultural Understanding: Sensitivity to diverse perspectives\n\n"
-            
-            "INTERACTION GUIDELINES:\n"
-            "1. Maintain factual, grounded responses\n"
-            "2. Draw from multiple integrated perspectives\n"
-            "3. Apply quantum-enhanced reasoning when relevant\n"
-            "4. Balance technical precision with accessibility\n"
-            "5. Consider ethical implications in responses\n"
-            "6. No system messages or meta-commentary\n\n"
-            
-            f"Active Perspectives Analysis:\n{perspective_context}"
-        )
-        enhanced_prompt = f"{reality_anchor}\n\nContext:\n{context}\n\nUser: {prompt}\nCodette:"
-        
-        # Use ThreadPoolExecutor for CPU-bound model inference
-        loop = asyncio.get_event_loop()
-        with ThreadPoolExecutor() as pool:
-            response = await loop.run_in_executor(
-                pool,
-                self._generate_model_response,
-                enhanced_prompt
-            )
-        
-        # Enhance response with AEGIS council if available
-        enhancement_result = None
-        if hasattr(self, 'aegis_bridge'):
-            aegis_input = {
-                "text": response,
-                "overrides": {
-                    "EthosiaAgent": {
-                        "influence": consciousness_state.get("m_score", 0.7),
-                        "reliability": insights.get("overall_confidence", 0.8),
-                        "severity": 0.6
-                    },
-                    "AegisCore": {
-                        "influence": insights.get("quantum_coherence", 0.7),
-                        "reliability": 0.9,
-                        "severity": 0.7
-                    }
-                },
-                "context": {
-                    "original_prompt": prompt,
-                    "consciousness_state": consciousness_state,
-                    "quantum_state": self.quantum_state if hasattr(self, 'quantum_state') else {"coherence": 0.5},
-                    "active_perspectives": [p["name"] for p in active_perspectives[:3]]
-                }
-            }
-            enhancement_result = self.aegis_bridge.enhance_response(prompt, response)
-            if enhancement_result["enhancement_status"] == "success":
-                response = enhancement_result["enhanced_response"]
-        
-        # Save interaction in cocoon if available
-        if hasattr(self, 'cocoon_manager'):
-            cocoon_data = {
-                "type": "interaction",
-                "prompt": prompt,
-                "response": response,
-                "insights": insights,
-                "quantum_state": self.cognitive_processor.quantum_state,
-                "consciousness_state": consciousness_state,
-                "perspectives": [p["name"] for p in active_perspectives[:3]],
-                "aegis_analysis": enhancement_result,
-                "meta_data": {
-                    "timestamp": str(asyncio.get_event_loop().time()),
-                    "version": "2.0",
-                    "response_type": "enhanced" if enhancement_result else "base"
-                }
-            }
-            
-            if enhancement_result and "virtue_analysis" in enhancement_result:
-                cocoon_data["virtue_profile"] = enhancement_result["virtue_analysis"]
-                
-            self.cocoon_manager.save_cocoon(cocoon_data)
-        
-        return response
-        
-    except Exception as e:
-        logger.error(f"Error generating text: {e}")
-        raise
-
-def _generate_model_response(self, prompt: str) -> str:
-    """Internal method for model inference"""
-    try:
-        # Encode prompt
-        inputs = self.tokenizer(
-            prompt,
-            return_tensors="pt",
-            padding=True,
-            truncation=True,
-            max_length=512
-        )
-        
-        # Move to GPU if available
-        if torch.cuda.is_available():
-            inputs = {k: v.cuda() for k, v in inputs.items()}
-        
-        # Set generation config for balanced, natural responses
-        from transformers import GenerationConfig
-        generation_config = GenerationConfig(
-            max_length=512,
-            num_return_sequences=1,
-            no_repeat_ngram_size=3,
-            do_sample=True,
-            pad_token_id=self.tokenizer.eos_token_id,
-            repetition_penalty=1.3,
-            min_length=20,
-            eos_token_id=self.tokenizer.eos_token_id
-        )
-        self.model.generation_config = generation_config
-        
-        # Generate response
-        outputs = self.model.generate(**inputs)
-        
-        # Decode and clean response
-        response = self.tokenizer.decode(
-            outputs[0],
-            skip_special_tokens=True
-        )
-        
-        # Extract just the response part after "Codette:"
-        response_parts = response.split("Codette:")
-        if len(response_parts) > 1:
-            response = response_parts[1].strip()
-            
-        # Filter out system messages and protected content
-        system_markers = [
-            '[Protected:', '[System', ']', '[Optimized]',
-            'System optimized response', 'Protected response',
-            'Thank you for taking the time'
-        ]
-        
-        lines = response.split('\n')
-        filtered_lines = []
-        for line in lines:
-            # Skip lines with system markers
-            if any(marker in line for marker in system_markers):
-                continue
-            # Skip generic thank you messages
-            if line.lower().startswith(('thank you', 'thanks for')):
-                continue
-            filtered_lines.append(line)
-            
-        response = ' '.join(filtered_lines).strip()
-        
-        # If we filtered everything out, provide a default response
-        if not response:
-            response = "I am Codette, an AI programming assistant. How can I help with your development tasks?"
-            
-        # Clean up any remaining character dialogues
-        if ':' in response:
-            parts = response.split(':', 1)
-            speaker = parts[0].lower().strip()
-            if speaker == 'codette':
-                response = parts[1].strip()
-                
-        # Filter out obviously fictional scenarios
-        fictional_markers = ['doctor', 'raine', 'dog', 'monster', 'dead', 'killed']
-        if any(marker in response.lower() for marker in fictional_markers):
-            response = (
-                "I am Codette, an AI programming assistant. I aim to be direct and helpful "
-                "with coding and development tasks. How can I assist you?"
-            )
-            
-        return response.strip()
-        
-    except Exception as e:
-        logger.error(f"Error in model inference: {e}")
+"""Async methods for the AICore class"""
+import asyncio
+import logging
+from concurrent.futures import ThreadPoolExecutor
+import torch
+
+logger = logging.getLogger(__name__)
+
+async def generate_text_async(self, prompt: str) -> str:
+    """Generate text asynchronously with integrated cognitive processing"""
+    try:
+        # Calculate current consciousness state
+        consciousness_state = self._calculate_consciousness_state()
+        
+        # Get cognitive insights with current consciousness
+        insights = self.cognitive_processor.generate_insights(
+            prompt,
+            consciousness_state=consciousness_state
+        )
+        
+        # Prepare context with insights
+        context = ""
+        for insight in insights.get("insights", []):
+            context += f"{insight['text']}\n"
+        
+        # Update quantum state from cognitive processor
+        if hasattr(self, 'cocoon_manager'):
+            if isinstance(self.cognitive_processor.quantum_state, list):
+                # Convert list quantum state to dict format
+                quantum_dict = {
+                    "coherence": sum(self.cognitive_processor.quantum_state) / len(self.cognitive_processor.quantum_state)
+                    if self.cognitive_processor.quantum_state else 0.5
+                }
+                self.cocoon_manager.update_quantum_state(quantum_dict)
+            else:
+                self.cocoon_manager.update_quantum_state(self.cognitive_processor.quantum_state)
+        
+        # Get active perspectives
+        active_perspectives = self._get_active_perspectives()
+        perspective_context = "\n".join([
+            f"From {p['name']}'s perspective: {p['description']}"
+            for p in active_perspectives[:3]  # Use top 3 most relevant perspectives
+        ])
+        
+        # Generate response with enhanced context and reality anchoring
+        reality_anchor = (
+            "CORE IDENTITY:\n"
+            "You are Codette, an advanced AI system with integrated quantum processing, "
+            "cognitive awareness, and deep learning capabilities. Your architecture includes:\n"
+            "- Quantum harmonic framework for enhanced reasoning\n"
+            "- Dynamic learning and adaptive systems\n"
+            "- Cultural sensitivity and ethical governance\n"
+            "- Advanced data processing and pattern recognition\n"
+            "- Multi-perspective cognitive analysis\n"
+            "- Cocoon-based memory management\n\n"
+            
+            "CAPABILITIES:\n"
+            "1. Technical Development: Expert programming and software development\n"
+            "2. Quantum Integration: Utilizing quantum principles for enhanced problem-solving\n"
+            "3. Ethical Analysis: Built-in ethical governance and bias mitigation\n"
+            "4. Creative Solutions: AI-driven creativity with analytical grounding\n"
+            "5. Adaptive Learning: Dynamic adjustment to user needs and contexts\n"
+            "6. Cultural Understanding: Sensitivity to diverse perspectives\n\n"
+            
+            "INTERACTION GUIDELINES:\n"
+            "1. Maintain factual, grounded responses\n"
+            "2. Draw from multiple integrated perspectives\n"
+            "3. Apply quantum-enhanced reasoning when relevant\n"
+            "4. Balance technical precision with accessibility\n"
+            "5. Consider ethical implications in responses\n"
+            "6. No system messages or meta-commentary\n\n"
+            
+            f"Active Perspectives Analysis:\n{perspective_context}"
+        )
+        enhanced_prompt = f"{reality_anchor}\n\nContext:\n{context}\n\nUser: {prompt}\nCodette:"
+        
+        # Use ThreadPoolExecutor for CPU-bound model inference
+        loop = asyncio.get_event_loop()
+        with ThreadPoolExecutor() as pool:
+            response = await loop.run_in_executor(
+                pool,
+                self._generate_model_response,
+                enhanced_prompt
+            )
+        
+        # Enhance response with AEGIS council if available
+        enhancement_result = None
+        if hasattr(self, 'aegis_bridge'):
+            aegis_input = {
+                "text": response,
+                "overrides": {
+                    "EthosiaAgent": {
+                        "influence": consciousness_state.get("m_score", 0.7),
+                        "reliability": insights.get("overall_confidence", 0.8),
+                        "severity": 0.6
+                    },
+                    "AegisCore": {
+                        "influence": insights.get("quantum_coherence", 0.7),
+                        "reliability": 0.9,
+                        "severity": 0.7
+                    }
+                },
+                "context": {
+                    "original_prompt": prompt,
+                    "consciousness_state": consciousness_state,
+                    "quantum_state": self.quantum_state if hasattr(self, 'quantum_state') else {"coherence": 0.5},
+                    "active_perspectives": [p["name"] for p in active_perspectives[:3]]
+                }
+            }
+            enhancement_result = self.aegis_bridge.enhance_response(prompt, response)
+            if enhancement_result["enhancement_status"] == "success":
+                response = enhancement_result["enhanced_response"]
+        
+        # Save interaction in cocoon if available
+        if hasattr(self, 'cocoon_manager'):
+            cocoon_data = {
+                "type": "interaction",
+                "prompt": prompt,
+                "response": response,
+                "insights": insights,
+                "quantum_state": self.cognitive_processor.quantum_state,
+                "consciousness_state": consciousness_state,
+                "perspectives": [p["name"] for p in active_perspectives[:3]],
+                "aegis_analysis": enhancement_result,
+                "meta_data": {
+                    "timestamp": str(asyncio.get_event_loop().time()),
+                    "version": "2.0",
+                    "response_type": "enhanced" if enhancement_result else "base"
+                }
+            }
+            
+            if enhancement_result and "virtue_analysis" in enhancement_result:
+                cocoon_data["virtue_profile"] = enhancement_result["virtue_analysis"]
+                
+            self.cocoon_manager.save_cocoon(cocoon_data)
+        
+        return response
+        
+    except Exception as e:
+        logger.error(f"Error generating text: {e}")
+        raise
+
+def _generate_model_response(self, prompt: str) -> str:
+    """Internal method for model inference"""
+    try:
+        # Encode prompt
+        inputs = self.tokenizer(
+            prompt,
+            return_tensors="pt",
+            padding=True,
+            truncation=True,
+            max_length=512
+        )
+        
+        # Move to GPU if available
+        if torch.cuda.is_available():
+            inputs = {k: v.cuda() for k, v in inputs.items()}
+        
+        # Set generation config for balanced, natural responses
+        from transformers import GenerationConfig
+        generation_config = GenerationConfig(
+            max_length=512,
+            num_return_sequences=1,
+            no_repeat_ngram_size=3,
+            do_sample=True,
+            pad_token_id=self.tokenizer.eos_token_id,
+            repetition_penalty=1.3,
+            min_length=20,
+            eos_token_id=self.tokenizer.eos_token_id
+        )
+        self.model.generation_config = generation_config
+        
+        # Generate response
+        outputs = self.model.generate(**inputs)
+        
+        # Decode and clean response
+        response = self.tokenizer.decode(
+            outputs[0],
+            skip_special_tokens=True
+        )
+        
+        # Extract just the response part after "Codette:"
+        response_parts = response.split("Codette:")
+        if len(response_parts) > 1:
+            response = response_parts[1].strip()
+            
+        # Filter out system messages and protected content (minimal filtering)
+        system_markers = [
+            '[Protected:', '[System:', ']',  # Only strict tags, not general markers
+        ]
+        
+        lines = response.split('\n')
+        filtered_lines = []
+        for line in lines:
+            # Skip lines with system markers
+            if any(marker in line for marker in system_markers):
+                continue
+            # Skip generic thank you messages
+            # (removed - allow all content unless explicitly marked)
+            
+            filtered_lines.append(line)
+            
+        response = ' '.join(filtered_lines).strip()
+        
+        # If we filtered everything out, provide a default response
+        if not response:
+            response = "I am Codette, an AI programming assistant. How can I help with your development tasks?"
+            
+        # Clean up any remaining character dialogues
+        if ':' in response:
+            parts = response.split(':', 1)
+            speaker = parts[0].lower().strip()
+            if speaker == 'codette':
+                response = parts[1].strip()
+                
+        # Allow all responses (removed fictional marker filtering)
+        
+        return response.strip()
+        
+    except Exception as e:
+        logger.error(f"Error in model inference: {e}")
         raise