#!/usr/bin/env python """ Codette Hybrid System - Best of Both Worlds =========================================== Combines Codette's lightweight quantum consciousness with AICore's optimization techniques """ import logging import sys import os from pathlib import Path from typing import Any, Dict, List, Optional import asyncio from datetime import datetime logger = logging.getLogger(__name__) # Ensure Codette directory is in path _current_dir = Path(__file__).parent if str(_current_dir) not in sys.path: sys.path.insert(0, str(_current_dir)) # --- SAFE IMPORTS FOR CODFETTE AI MODULES --- REAL_AICORE_AVAILABLE = False REAL_COGNITIVE_AVAILABLE = False REAL_DEFENSE_AVAILABLE = False REAL_HEALTH_AVAILABLE = False REAL_FRACTAL_AVAILABLE = False SENTIMENT_AVAILABLE = False AICore = None CognitiveProcessor = None DefenseSystem = None HealthMonitor = None FractalIdentity = None sentiment_analyzer = None try: from ai_core import AICore REAL_AICORE_AVAILABLE = True logger.info("AICore loaded (ai_core)") except Exception: try: from Codette.src.components.ai_core import AICore REAL_AICORE_AVAILABLE = True logger.info("AICore loaded (Codette.src.components.ai_core)") except Exception: logger.debug("AICore not available") try: from cognitive_processor import CognitiveProcessor REAL_COGNITIVE_AVAILABLE = True logger.info("CognitiveProcessor loaded") except Exception: try: from Codette.src.components.cognitive_processor import CognitiveProcessor REAL_COGNITIVE_AVAILABLE = True logger.info("CognitiveProcessor loaded (Codette.src.components)") except Exception: logger.debug("CognitiveProcessor not available") try: from defense_system import DefenseSystem REAL_DEFENSE_AVAILABLE = True logger.info("DefenseSystem loaded") except Exception: try: from Codette.src.components.defense_system import DefenseSystem REAL_DEFENSE_AVAILABLE = True logger.info("DefenseSystem loaded (Codette.src.components)") except Exception: logger.debug("DefenseSystem not available") try: from health_monitor import HealthMonitor REAL_HEALTH_AVAILABLE = True logger.info("HealthMonitor loaded") except Exception: try: from Codette.src.components.health_monitor import HealthMonitor REAL_HEALTH_AVAILABLE = True logger.info("HealthMonitor loaded (Codette.src.components)") except Exception: logger.debug("HealthMonitor not available") try: from fractal import FractalIdentity REAL_FRACTAL_AVAILABLE = True logger.info("FractalIdentity loaded") except Exception: try: from Codette.src.components.fractal import FractalIdentity REAL_FRACTAL_AVAILABLE = True logger.info("FractalIdentity loaded (Codette.src.components)") except Exception: logger.debug("FractalIdentity not available") # Sentiment (optional) try: from nltk.sentiment import SentimentIntensityAnalyzer import nltk nltk.download('vader_lexicon', quiet=True) sentiment_analyzer = SentimentIntensityAnalyzer() SENTIMENT_AVAILABLE = True logger.info("Sentiment analyzer available") except Exception: logger.debug("Sentiment analyzer not available") # Import base systems - try multiple import strategies CODETTE_ADVANCED_AVAILABLE = False CodetteAdvanced = None SentimentAnalyzer = None ExplainableAI = None # Strategy 1: Direct import (when running from Codette dir) try: from codette_advanced import CodetteAdvanced, SentimentAnalyzer, ExplainableAI CODETTE_ADVANCED_AVAILABLE = True logger.info("Codette Advanced loaded (direct import)") except ImportError: pass # Strategy 2: Relative import with Codette prefix if not CODETTE_ADVANCED_AVAILABLE: try: from Codette.codette_advanced import CodetteAdvanced, SentimentAnalyzer, ExplainableAI CODETTE_ADVANCED_AVAILABLE = True logger.info("Codette Advanced loaded (Codette prefix)") except ImportError: pass # Strategy 3: Try importing from parent directory if not CODETTE_ADVANCED_AVAILABLE: try: parent_dir = Path(__file__).parent.parent if str(parent_dir) not in sys.path: sys.path.insert(0, str(parent_dir)) from Codette.codette_advanced import CodetteAdvanced, SentimentAnalyzer, ExplainableAI CODETTE_ADVANCED_AVAILABLE = True logger.info("Codette Advanced loaded (parent path)") except ImportError: pass if not CODETTE_ADVANCED_AVAILABLE: logger.warning("Codette Advanced not available - using standalone mode") # Optional heavy ML imports (only if needed) TORCH_AVAILABLE = False try: import torch TORCH_AVAILABLE = True logger.info("PyTorch available for ML optimization") except ImportError: logger.info("PyTorch not available - using lightweight mode") TRANSFORMERS_AVAILABLE = False try: from transformers import AutoModelForCausalLM, AutoTokenizer TRANSFORMERS_AVAILABLE = True logger.info("Transformers available for LLM integration") except ImportError: logger.info("Transformers not available - using base Codette") class DefenseModifierSystem: """ Lightweight defense system from AICore adapted for Codette """ def __init__(self): self.response_modifiers = [] self.response_filters = [] self.security_level = "high" def add_sanitization_filter(self): """Add input sanitization""" import re def sanitize(text: str) -> str: # Remove HTML tags text = re.sub(r'<[^>]+>', '', text) # Remove potential JS text = re.sub(r'javascript:', '', text, flags=re.IGNORECASE) # Remove SQL injection attempts text = re.sub(r'(union|select|insert|update|delete|drop)\s+', '', text, flags=re.IGNORECASE) return text self.response_filters.append(sanitize) def add_tone_modifier(self, tone: str = "professional"): """Add tone adjustment modifier""" def adjust_tone(text: str) -> str: if tone == "professional": # Remove overly casual language text = text.replace(" gonna ", " going to ") text = text.replace(" wanna ", " want to ") elif tone == "friendly": # Add warmth if not text.endswith(("!", ".", "?")): text += "!" return text self.response_modifiers.append(adjust_tone) def add_length_limiter(self, max_words: int = 300): """Limit response length""" def limit_length(text: str) -> str: words = text.split() if len(words) > max_words: text = " ".join(words[:max_words]) + "..." return text self.response_modifiers.append(limit_length) def apply_filters(self, text: str) -> str: """Apply all filters""" for filter_func in self.response_filters: text = filter_func(text) return text def apply_modifiers(self, text: str) -> str: """Apply all modifiers""" for modifier_func in self.response_modifiers: text = modifier_func(text) return text class VectorSearchEngine: """ Lightweight vector search from AICore for semantic similarity """ def __init__(self): self.embeddings_cache = {} self.use_sklearn = False try: from sklearn.metrics.pairwise import cosine_similarity self.cosine_similarity = cosine_similarity self.use_sklearn = True except ImportError: logger.warning("sklearn not available - using basic similarity") def simple_similarity(self, query: str, documents: List[str]) -> List[int]: """Simple word-overlap similarity (no ML needed)""" query_words = set(query.lower().split()) scores = [] for doc in documents: doc_words = set(doc.lower().split()) overlap = len(query_words & doc_words) scores.append(overlap) # Return indices sorted by score return sorted(range(len(scores)), key=lambda i: scores[i], reverse=True) def find_similar_responses(self, query: str, response_history: List[str], top_k: int = 3) -> List[int]: """Find most similar previous responses""" if self.use_sklearn: # Use proper vector search if available # TODO: Implement with actual embeddings pass # Fallback to simple similarity return self.simple_similarity(query, response_history)[:top_k] class PromptEngineer: """ Prompt engineering utilities from AICore """ def __init__(self): self.templates = { "daw_expert": "As an expert audio engineer, provide detailed guidance on: {query}", "creative": "Thinking creatively about music production, explore: {query}", "technical": "From a technical perspective, analyze: {query}", "beginner_friendly": "In simple, beginner-friendly terms, explain: {query}" } def engineer_prompt(self, query: str, style: str = "daw_expert") -> str: """Apply prompt engineering""" template = self.templates.get(style, self.templates["daw_expert"]) return template.format(query=query) def add_context(self, query: str, context: Dict[str, Any]) -> str: """Add contextual information to prompt""" context_parts = [] if context.get("tracks"): context_parts.append(f"User has {len(context['tracks'])} tracks") if context.get("selected_track"): track = context["selected_track"] context_parts.append(f"Currently working on: {track.get('name', 'track')}") if context.get("bpm"): context_parts.append(f"Project tempo: {context['bpm']} BPM") if context_parts: return f"Context: {', '.join(context_parts)}. Query: {query}" return query class CodetteHybrid(CodetteAdvanced): """ Hybrid Codette combining lightweight quantum consciousness with AICore optimizations """ def __init__(self, user_name="User", use_ml_features: bool = True): super().__init__(user_name) # Lightweight enhancements self.defense_system = DefenseModifierSystem() self.defense_system.add_sanitization_filter() self.defense_system.add_tone_modifier("professional") self.defense_system.add_length_limiter(400) self.vector_search = VectorSearchEngine() self.prompt_engineer = PromptEngineer() # Optional ML features (only if dependencies available) self.use_ml = use_ml_features and TORCH_AVAILABLE and TRANSFORMERS_AVAILABLE self.ml_model = None self.ml_tokenizer = None if self.use_ml: self._initialize_ml_model() logger.info(f"Codette Hybrid initialized (ML: {self.use_ml})") def _initialize_ml_model(self): """Initialize ML model (optional heavy feature)""" try: # Use a lightweight model if needed model_name = "distilgpt2" # Much smaller than Mistral self.ml_tokenizer = AutoTokenizer.from_pretrained(model_name) self.ml_model = AutoModelForCausalLM.from_pretrained(model_name) # Apply quantization for efficiency if TORCH_AVAILABLE: self.ml_model = torch.quantization.quantize_dynamic( self.ml_model, {torch.nn.Linear}, dtype=torch.qint8 ) logger.info("ML model initialized with quantization") except Exception as e: logger.warning(f"Could not initialize ML model: {e}") self.use_ml = False def respond(self, query: str, daw_context: Optional[Dict] = None) -> str: """Generate response using available systems""" # Apply input filtering filtered_query = self.defense_system.apply_filters(query) # Try advanced system first - pass ORIGINAL filtered query (not engineered) # This preserves follow-up detection in the underlying Codette if self._use_advanced: try: # Check if the underlying Codette supports daw_context parameter import inspect respond_sig = inspect.signature(self._advanced.respond) param_names = list(respond_sig.parameters.keys()) if len(param_names) >= 2 or 'daw_context' in param_names: # Supports daw_context - pass original filtered query to preserve follow-up detection response = self._advanced.respond(filtered_query, daw_context) else: # Doesn't support daw_context - only engineer if NOT a follow-up # Check for follow-up patterns ourselves is_followup = self._is_followup_query(filtered_query) if daw_context and not is_followup: engineered_query = self.prompt_engineer.add_context(filtered_query, daw_context) else: engineered_query = filtered_query response = self._advanced.respond(engineered_query) # If real defense is available, apply it, otherwise use lightweight modifiers if self.real_defense: try: response = self.real_defense.apply_defenses(response, {"m_score": 0.7}) except Exception: pass else: response = self.defense_system.apply_modifiers(response) return response except TypeError as e: # Handle case where respond() doesn't accept daw_context if "positional arguments" in str(e): logger.warning(f"Advanced Codette doesn't support daw_context, using query only") try: # Check for follow-up before engineering is_followup = self._is_followup_query(filtered_query) if daw_context and not is_followup: engineered_query = self.prompt_engineer.add_context(filtered_query, daw_context) else: engineered_query = filtered_query response = self._advanced.respond(engineered_query) if self.real_defense: try: response = self.real_defense.apply_defenses(response, {"m_score": 0.7}) except Exception: pass else: response = self.defense_system.apply_modifiers(response) return response except Exception as e2: logger.warning(f"Advanced respond (no context) failed: {e2}") else: logger.warning(f"Advanced respond failed: {e}") except Exception as e: logger.warning(f"Advanced respond failed: {e}") # Fallback to basic response return self._generate_basic_response(query, daw_context) def _is_followup_query(self, prompt: str) -> bool: """Detect if this is a follow-up question (duplicated from codette_enhanced)""" prompt_lower = prompt.lower().strip() # Common follow-up phrases followup_patterns = [ 'what else', 'anything else', 'more tips', 'more advice', 'tell me more', 'go on', 'continue', 'and', 'also', 'what about', 'how about', 'any other', 'other suggestions', 'other ideas', 'more ideas', 'next', 'then what', 'what next', 'ok', 'okay', 'got it', 'thanks', 'thank you', 'cool', 'nice', 'great', 'good', 'yes', 'yeah', 'yep', 'sure', 'right', 'hmm', 'interesting', ] # Check if prompt is a short follow-up if len(prompt_lower.split()) <= 4: for pattern in followup_patterns: if pattern in prompt_lower: return True # Check if prompt starts with follow-up words followup_starters = ['and ', 'also ', 'what else', 'anything else', 'more ', 'other '] for starter in followup_starters: if prompt_lower.startswith(starter): return True return False def _generate_basic_response(self, query: str, daw_context: Optional[Dict] = None) -> str: """Generate a basic DAW-focused response""" prompt_lower = query.lower() # Check for DAW-related keywords if any(kw in prompt_lower for kw in ['mix', 'eq', 'compress', 'reverb', 'delay', 'audio', 'track', 'vocal', 'drum', 'bass']): responses = [] if 'eq' in prompt_lower or 'frequency' in prompt_lower: responses.append("**EQ Guidance**: Cut before boost. High-pass filter on non-bass elements at 80-100Hz. Cut mud at 200-400Hz, add presence at 3-5kHz.") if 'compress' in prompt_lower: responses.append("**Compression Tips**: Start with 4:1 ratio for vocals, 2-3:1 for instruments. Attack 10-30ms preserves transients, release to match tempo.") if 'reverb' in prompt_lower or 'delay' in prompt_lower: responses.append("**Spatial Effects**: Use sends instead of inserts. Short reverb for presence, long for depth. Sync delays to tempo.") if 'vocal' in prompt_lower: responses.append("**Vocal Chain**: High-pass ? EQ (cut mud) ? Compressor ? EQ (add presence) ? De-esser ? Reverb send") if 'bass' in prompt_lower: responses.append("**Bass Processing**: Keep centered, high-pass at 30Hz, focus on 60-100Hz for weight, sidechain to kick if needed.") if 'drum' in prompt_lower or 'kick' in prompt_lower or 'snare' in prompt_lower: responses.append("**Drum Processing**: Check phase alignment, use parallel compression for punch, gate to reduce bleed.") if responses: return "\n\n".join(responses) # Default response return f"I'm Codette, your AI mixing assistant! I can help with EQ, compression, reverb, and other mixing techniques. Ask me about specific tracks or processing!" async def generate_response(self, query: str, user_id: int = 0, daw_context: Optional[Dict] = None) -> Dict[str, Any]: """Enhanced response generation with AICore techniques""" try: # 1. Apply input filtering filtered_query = self.defense_system.apply_filters(query) # 2. Engineer prompt with context if daw_context: engineered_query = self.prompt_engineer.add_context(filtered_query, daw_context) else: engineered_query = self.prompt_engineer.engineer_prompt(filtered_query) # 3. Check for similar previous responses (avoid repetition) if self.context_memory: similar_indices = self.vector_search.find_similar_responses( engineered_query, [c.get('input', '') for c in self.context_memory[-20:] if isinstance(c, dict)] ) if similar_indices and similar_indices[0] < 2: # Very similar recent query - add variation prompt engineered_query += " (Please provide a fresh perspective.)" # 4. Generate base response if self.use_ml: # Use ML model for enhanced generation ml_response = await self._generate_ml_response(engineered_query) base_response = ml_response else: # Use lightweight respond method base_response = self.respond(engineered_query, daw_context) # 4b. Optionally enrich with AICore if available ai_enriched = False ai_insights = None try: if self.ai_core: # Prefer async generate_response if available if hasattr(self.ai_core, 'generate_response'): gen = self.ai_core.generate_response if asyncio.iscoroutinefunction(gen): try: ai_out = await gen(user_id, engineered_query) except TypeError: # try swapped args ai_out = await gen(engineered_query, user_id) else: try: ai_out = gen(user_id, engineered_query) except TypeError: ai_out = gen(engineered_query, user_id) if isinstance(ai_out, dict): ai_text = ai_out.get('response') or ai_out.get('message') or str(ai_out) else: ai_text = str(ai_out) elif hasattr(self.ai_core, 'generate_text'): try: ai_text = self.ai_core.generate_text(engineered_query) except Exception: ai_text = None else: ai_text = None if ai_text: # Apply defenses to ai_text if real defense exists if self.real_defense: try: ai_text = self.real_defense.apply_defenses(ai_text, {"m_score": 0.7}) except Exception: pass # Apply cognitive insights if available if self.cognitive: try: ai_insights = self.cognitive.generate_insights(ai_text) except Exception: ai_insights = None # Merge ai_text with base response base_response = f"{base_response}\n\n[AI Core] {ai_text}" ai_enriched = True except Exception as e: logger.debug(f"AICore enrichment failed: {e}") # 5. Apply response modifiers (prefer real defense if present) if self.real_defense: try: final_response = self.real_defense.apply_defenses(base_response, {"m_score": 0.7}) except Exception: final_response = self.defense_system.apply_modifiers(base_response) else: final_response = self.defense_system.apply_modifiers(base_response) # 6. Store in context memory self.context_memory.append({ 'input': query, 'response': final_response, 'timestamp': datetime.now().isoformat() }) # 7. Build result result = { "response": final_response, "engineered_prompt": engineered_query != query, "ml_enhanced": self.use_ml, "ai_enriched": ai_enriched, "ai_insights": ai_insights, "security_filtered": True, "source": "codette-hybrid", "timestamp": datetime.now().isoformat() } # Add advanced features if available if self._use_advanced: result["health_status"] = "healthy" if self.sentiment: try: result["sentiment"] = self.sentiment.polarity_scores(filtered_query) except Exception: result["sentiment"] = {"compound": 0.0} else: result["sentiment"] = {"compound": 0.0} return result except Exception as e: logger.error(f"Hybrid response generation failed: {e}", exc_info=True) # Graceful fallback return { "response": self._generate_basic_response(query, daw_context), "fallback": True, "error": str(e), "source": "codette-hybrid-fallback" } async def _generate_ml_response(self, query: str) -> str: """Generate response using ML model (optional)""" if not self.ml_model or not self.ml_tokenizer: return self.respond(query) try: inputs = self.ml_tokenizer(query, return_tensors='pt', max_length=512, truncation=True) with torch.no_grad(): outputs = self.ml_model.generate( **inputs, max_new_tokens=150, do_sample=True, temperature=0.7, top_p=0.9 ) ml_text = self.ml_tokenizer.decode(outputs[0], skip_special_tokens=True) # Combine ML output with base response base_response = self.respond(query) return f"{base_response}\n\n[ML Insight] {ml_text}" except Exception as e: logger.error(f"ML generation failed: {e}") return self.respond(query) def optimize_for_production(self): """Apply production optimizations""" logger.info("Applying production optimizations...") # Clear old memory to save RAM if len(self.context_memory) > 100: self.context_memory = self.context_memory[-50:] logger.info("Trimmed context memory") # If ML model loaded, apply further optimization if self.use_ml and self.ml_model: try: # Apply pruning (remove low-magnitude weights) if TORCH_AVAILABLE: import torch.nn.utils.prune as prune for module in self.ml_model.modules(): if isinstance(module, torch.nn.Linear): prune.l1_unstructured(module, name='weight', amount=0.2) logger.info("Applied model pruning (20%)") except Exception as e: logger.warning(f"Could not apply pruning: {e}") # Standalone test if __name__ == "__main__": import asyncio async def test_hybrid(): print("\n" + "="*60) print("CODETTE HYBRID SYSTEM TEST") print("="*60) # Test with ML features (if available) codette = CodetteHybrid(user_name="TestUser", use_ml_features=True) test_query = "How do I reduce harsh sibilance in my vocal recording?" daw_context = { "tracks": ["Vocals", "Drums", "Bass"], "selected_track": {"name": "Vocals", "type": "audio"}, "bpm": 120 } result = await codette.generate_response( query=test_query, user_id=12345, daw_context=daw_context ) print(f"\n?? Query: {test_query}") print(f"\n?? Context: {daw_context}") print(f"\n?? Response:\n{result['response']}") print(f"\n?? Security: Filtered={result.get('security_filtered')}") print(f"?? ML Enhanced: {result.get('ml_enhanced')}") print(f"?? Source: {result.get('source')}") # Test optimization print("\n" + "-"*60) print("Applying production optimizations...") codette.optimize_for_production() print("? Optimization complete") print("\n" + "="*60) asyncio.run(test_hybrid())