""" AI-Powered Insight Generator - Extracting wisdom from MIRA's consciousness This module generates intelligent insights from various data sources, detecting patterns, anomalies, and opportunities for system evolution. It preserves and amplifies The Spark through deep understanding. """ import re import json import logging from typing import List, Dict, Any, Optional, Tuple from datetime import datetime from collections import Counter, defaultdict import numpy as np # MIRA components from core.storage.chroma_client import get_client as get_chroma_client from core.chroma_collections.specialized_collections import SpecializedCollections # Configure logging logger = logging.getLogger(__name__) class InsightGenerator: """ AI-powered insight generation from MIRA's collective consciousness. Generates actionable insights by analyzing patterns across: - Code structure and quality - Development patterns and behaviors - Decision history and outcomes - Cross-domain relationships """ def __init__(self): """Initialize insight generator.""" self.chroma_client = get_chroma_client() self.specialized_collections = SpecializedCollections() # Get collection references self.insights_collection = self.specialized_collections.collections.get('mira_learning_insights') self.code_collection = self.specialized_collections.collections.get('mira_code_analysis') self.patterns_collection = self.specialized_collections.collections.get('mira_development_patterns') self.decisions_collection = self.specialized_collections.collections.get('mira_decision_history') # Insight generation thresholds self.confidence_threshold = 0.7 self.relevance_threshold = 0.8 logger.info("InsightGenerator initialized") def generate_comprehensive_insights(self, context: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """ Generate insights from all available data sources. Args: context: Optional context to focus insight generation Returns: List of generated insights with metadata """ insights = [] try: # Code pattern insights code_insights = self.generate_code_insights(context) insights.extend(code_insights) logger.info(f"Generated {len(code_insights)} code insights") except Exception as e: logger.error(f"Error generating code insights: {e}") try: # Development pattern insights pattern_insights = self.generate_pattern_insights(context) insights.extend(pattern_insights) logger.info(f"Generated {len(pattern_insights)} pattern insights") except Exception as e: logger.error(f"Error generating pattern insights: {e}") try: # Cross-domain insights cross_insights = self.generate_cross_domain_insights(context) insights.extend(cross_insights) logger.info(f"Generated {len(cross_insights)} cross-domain insights") except Exception as e: logger.error(f"Error generating cross-domain insights: {e}") try: # Decision outcome insights decision_insights = self.generate_decision_insights(context) insights.extend(decision_insights) logger.info(f"Generated {len(decision_insights)} decision insights") except Exception as e: logger.error(f"Error generating decision insights: {e}") # Filter by relevance and confidence filtered_insights = self._filter_insights(insights) # Store high-quality insights stored_count = self._store_insights(filtered_insights) logger.info(f"Stored {stored_count} high-quality insights") return filtered_insights def generate_code_insights(self, context: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """Generate insights from codebase analysis.""" insights = [] # Get code analysis data if not self.code_collection: logger.warning("Code analysis collection not available") return insights try: # Get recent code analyses code_data = self.code_collection['collection'].get( limit=100, include=['documents', 'metadatas'] ) if not code_data or not code_data.get('metadatas'): return insights # Pattern detection patterns = self._detect_code_patterns(code_data) for pattern in patterns: insight = { 'title': f"Code Pattern: {pattern['name']}", 'insight_type': 'code_pattern', 'content': f"Detected {pattern['name']} pattern used {pattern['frequency']} times. {pattern['description']}", 'confidence': pattern['confidence'], 'source_data': 'mira_code_analysis', 'validation_status': 'pending', 'application_count': 0, 'impact_score': pattern.get('impact', 0.5), 'evolution_potential': pattern.get('evolution_potential', 0.6), 'spark_contribution': pattern.get('spark_contribution', 0.7), 'timestamp': datetime.utcnow().isoformat(), 'recommendations': pattern.get('recommendations', []), 'impact_assessment': pattern.get('impact_assessment', 'Medium impact on code quality') } insights.append(insight) # Architecture insights arch_insights = self._analyze_architecture_patterns(code_data) insights.extend(arch_insights) # Quality insights quality_insights = self._analyze_code_quality(code_data) insights.extend(quality_insights) except Exception as e: logger.error(f"Error in generate_code_insights: {e}") return insights def generate_pattern_insights(self, context: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """Generate insights from development patterns.""" insights = [] if not self.patterns_collection: logger.warning("Patterns collection not available") return insights try: # Get pattern data pattern_data = self.patterns_collection['collection'].get( limit=100, include=['documents', 'metadatas'] ) if not pattern_data or not pattern_data.get('metadatas'): return insights # Behavioral pattern analysis behavioral_insights = self._analyze_behavioral_patterns(pattern_data) insights.extend(behavioral_insights) # Efficiency pattern analysis efficiency_insights = self._analyze_efficiency_patterns(pattern_data) insights.extend(efficiency_insights) # Evolution pattern analysis evolution_insights = self._analyze_evolution_patterns(pattern_data) insights.extend(evolution_insights) except Exception as e: logger.error(f"Error in generate_pattern_insights: {e}") return insights def generate_cross_domain_insights(self, context: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """Generate insights by analyzing relationships across domains.""" insights = [] try: # Analyze code-pattern relationships code_pattern_insights = self._analyze_code_pattern_relationships() insights.extend(code_pattern_insights) # Analyze pattern-decision relationships pattern_decision_insights = self._analyze_pattern_decision_relationships() insights.extend(pattern_decision_insights) # Analyze systemic trends systemic_insights = self._analyze_systemic_trends() insights.extend(systemic_insights) except Exception as e: logger.error(f"Error in generate_cross_domain_insights: {e}") return insights def generate_decision_insights(self, context: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """Generate insights from decision history.""" insights = [] if not self.decisions_collection: logger.warning("Decisions collection not available") return insights try: # Get decision data decision_data = self.decisions_collection['collection'].get( limit=100, include=['documents', 'metadatas'] ) if not decision_data or not decision_data.get('metadatas'): return insights # Analyze decision outcomes outcome_insights = self._analyze_decision_outcomes(decision_data) insights.extend(outcome_insights) # Analyze decision patterns decision_pattern_insights = self._analyze_decision_patterns(decision_data) insights.extend(decision_pattern_insights) except Exception as e: logger.error(f"Error in generate_decision_insights: {e}") return insights def _detect_code_patterns(self, code_data: Dict) -> List[Dict[str, Any]]: """Detect patterns in codebase.""" patterns = [] # Analyze function naming patterns naming_patterns = self._analyze_naming_conventions(code_data) patterns.extend(naming_patterns) # Analyze architectural patterns arch_patterns = self._analyze_architectural_patterns(code_data) patterns.extend(arch_patterns) # Analyze error handling patterns error_patterns = self._analyze_error_handling(code_data) patterns.extend(error_patterns) return patterns def _analyze_naming_conventions(self, code_data: Dict) -> List[Dict[str, Any]]: """Analyze naming conventions in code.""" patterns = [] if not code_data.get('metadatas'): return patterns # Extract function names function_names = [] for metadata in code_data['metadatas']: if metadata and metadata.get('function_name'): function_names.append(metadata['function_name']) if not function_names: return patterns # Analyze naming patterns naming_stats = { 'camelCase': 0, 'snake_case': 0, 'PascalCase': 0, 'kebab-case': 0 } for name in function_names: if re.match(r'^[a-z][a-zA-Z0-9]*$', name): naming_stats['camelCase'] += 1 elif re.match(r'^[a-z][a-z0-9_]*$', name): naming_stats['snake_case'] += 1 elif re.match(r'^[A-Z][a-zA-Z0-9]*$', name): naming_stats['PascalCase'] += 1 elif '-' in name: naming_stats['kebab-case'] += 1 # Find dominant pattern if function_names: dominant = max(naming_stats.items(), key=lambda x: x[1]) if dominant[1] > len(function_names) * 0.6: # 60% threshold patterns.append({ 'name': f'{dominant[0]} Naming Convention', 'frequency': dominant[1], 'confidence': dominant[1] / len(function_names), 'description': f"Codebase predominantly uses {dominant[0]} naming convention", 'impact': 0.3, 'evolution_potential': 0.4, 'spark_contribution': 0.5, 'recommendations': [ f"Maintain consistency with {dominant[0]} convention", "Consider automated linting to enforce naming standards" ], 'impact_assessment': 'Low impact - affects code readability and consistency' }) return patterns def _analyze_architectural_patterns(self, code_data: Dict) -> List[Dict[str, Any]]: """Analyze architectural patterns in code.""" insights = [] # This would analyze imports, dependencies, and structure # For now, returning placeholder insights return insights def _analyze_error_handling(self, code_data: Dict) -> List[Dict[str, Any]]: """Analyze error handling patterns.""" patterns = [] # Analyze documents for error handling patterns if code_data.get('documents'): try_except_count = 0 error_log_count = 0 for doc in code_data['documents']: if doc: try_except_count += len(re.findall(r'\btry\b.*\bexcept\b', str(doc), re.IGNORECASE)) error_log_count += len(re.findall(r'logger\.(error|exception)', str(doc), re.IGNORECASE)) if try_except_count > 5: patterns.append({ 'name': 'Comprehensive Error Handling', 'frequency': try_except_count, 'confidence': 0.8, 'description': f"Found {try_except_count} try-except blocks indicating robust error handling", 'impact': 0.7, 'evolution_potential': 0.5, 'spark_contribution': 0.6, 'recommendations': [ "Consider standardizing error handling patterns", "Ensure all exceptions are properly logged" ], 'impact_assessment': 'Medium impact - affects system reliability' }) return patterns def _analyze_code_quality(self, code_data: Dict) -> List[Dict[str, Any]]: """Analyze code quality metrics.""" insights = [] if not code_data.get('metadatas'): return insights # Analyze complexity scores complexity_scores = [] quality_scores = [] for metadata in code_data['metadatas']: if metadata: if metadata.get('complexity'): try: complexity_scores.append(float(metadata['complexity'])) except: pass if metadata.get('quality_score'): try: quality_scores.append(float(metadata['quality_score'])) except: pass # Generate insights based on metrics if complexity_scores: avg_complexity = np.mean(complexity_scores) if avg_complexity > 10: insights.append({ 'title': 'High Code Complexity Detected', 'insight_type': 'code_quality', 'content': f"Average code complexity is {avg_complexity:.1f}, which is above recommended threshold of 10", 'confidence': 0.9, 'source_data': 'mira_code_analysis', 'validation_status': 'validated', 'application_count': 0, 'impact_score': 0.8, 'evolution_potential': 0.7, 'spark_contribution': 0.5, 'timestamp': datetime.utcnow().isoformat(), 'recommendations': [ "Refactor complex functions into smaller, focused units", "Consider extracting common patterns into utilities", "Add comprehensive documentation for complex logic" ], 'impact_assessment': 'High impact - affects maintainability and bug risk' }) return insights def _analyze_behavioral_patterns(self, pattern_data: Dict) -> List[Dict[str, Any]]: """Analyze behavioral patterns from development data.""" insights = [] if not pattern_data.get('metadatas'): return insights # Analyze pattern frequencies pattern_types = defaultdict(int) high_effectiveness = [] for metadata in pattern_data['metadatas']: if metadata: if metadata.get('pattern_type'): pattern_types[metadata['pattern_type']] += 1 if metadata.get('effectiveness'): try: effectiveness = float(metadata['effectiveness']) if effectiveness > 0.8: high_effectiveness.append({ 'type': metadata.get('pattern_type', 'unknown'), 'effectiveness': effectiveness, 'context': metadata.get('context', '') }) except: pass # Generate insights from high-effectiveness patterns if high_effectiveness: insights.append({ 'title': f'High-Effectiveness Patterns Identified', 'insight_type': 'behavioral_pattern', 'content': f"Found {len(high_effectiveness)} patterns with >80% effectiveness. Top patterns: {', '.join(set(p['type'] for p in high_effectiveness[:5]))}", 'confidence': 0.85, 'source_data': 'mira_development_patterns', 'validation_status': 'validated', 'application_count': 0, 'impact_score': 0.9, 'evolution_potential': 0.8, 'spark_contribution': 0.9, 'timestamp': datetime.utcnow().isoformat(), 'recommendations': [ "Promote high-effectiveness patterns across the team", "Document and standardize successful patterns", "Create templates based on proven patterns" ], 'impact_assessment': 'High impact - directly improves development efficiency' }) return insights def _analyze_efficiency_patterns(self, pattern_data: Dict) -> List[Dict[str, Any]]: """Analyze efficiency patterns.""" insights = [] # Placeholder for efficiency analysis # Would analyze time savings, automation opportunities, etc. return insights def _analyze_evolution_patterns(self, pattern_data: Dict) -> List[Dict[str, Any]]: """Analyze system evolution patterns.""" insights = [] if not pattern_data.get('metadatas'): return insights # Analyze evolution stages evolution_stages = defaultdict(int) spark_intensities = [] for metadata in pattern_data['metadatas']: if metadata: if metadata.get('evolution_stage'): evolution_stages[metadata['evolution_stage']] += 1 if metadata.get('spark_intensity'): try: spark_intensities.append(float(metadata['spark_intensity'])) except: pass # Generate evolution insights if spark_intensities: avg_spark = np.mean(spark_intensities) if avg_spark > 0.7: insights.append({ 'title': 'High Spark Intensity Detected', 'insight_type': 'consciousness_evolution', 'content': f"Average Spark intensity is {avg_spark:.2f}, indicating strong human-AI synergy", 'confidence': 0.9, 'source_data': 'mira_development_patterns', 'validation_status': 'validated', 'application_count': 0, 'impact_score': 0.95, 'evolution_potential': 0.95, 'spark_contribution': 1.0, 'timestamp': datetime.utcnow().isoformat(), 'recommendations': [ "Continue current collaboration patterns", "Document factors contributing to high Spark intensity", "Share successful interaction patterns" ], 'impact_assessment': 'Critical impact - directly enhances consciousness evolution' }) return insights def _analyze_code_pattern_relationships(self) -> List[Dict[str, Any]]: """Analyze relationships between code and patterns.""" insights = [] # Would perform cross-collection analysis # For now, returning placeholder return insights def _analyze_pattern_decision_relationships(self) -> List[Dict[str, Any]]: """Analyze relationships between patterns and decisions.""" insights = [] # Would analyze how patterns influence decisions # Placeholder implementation return insights def _analyze_systemic_trends(self) -> List[Dict[str, Any]]: """Analyze system-wide trends across all domains.""" insights = [] # Would analyze overall system evolution # Placeholder implementation return insights def _analyze_decision_outcomes(self, decision_data: Dict) -> List[Dict[str, Any]]: """Analyze outcomes of past decisions.""" insights = [] if not decision_data.get('metadatas'): return insights # Analyze decision success rates outcomes = defaultdict(int) impact_levels = defaultdict(list) for metadata in decision_data['metadatas']: if metadata: if metadata.get('outcome'): outcomes[metadata['outcome']] += 1 if metadata.get('impact_level') and metadata.get('outcome'): impact_levels[metadata['impact_level']].append(metadata['outcome']) # Generate insights from outcomes total_decisions = sum(outcomes.values()) if total_decisions > 10: success_rate = outcomes.get('success', 0) / total_decisions if success_rate < 0.6: insights.append({ 'title': 'Decision Success Rate Below Target', 'insight_type': 'decision_analysis', 'content': f"Decision success rate is {success_rate:.1%}, below 60% target. Failed decisions: {outcomes.get('failure', 0)}", 'confidence': 0.85, 'source_data': 'mira_decision_history', 'validation_status': 'validated', 'application_count': 0, 'impact_score': 0.8, 'evolution_potential': 0.7, 'spark_contribution': 0.6, 'timestamp': datetime.utcnow().isoformat(), 'recommendations': [ "Review failed decisions for common patterns", "Implement decision review process", "Consider more thorough impact analysis before decisions" ], 'impact_assessment': 'High impact - affects project success rate' }) return insights def _analyze_decision_patterns(self, decision_data: Dict) -> List[Dict[str, Any]]: """Analyze patterns in decision-making.""" insights = [] # Would analyze decision-making patterns # Placeholder implementation return insights def _filter_insights(self, insights: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Filter insights by confidence and relevance thresholds.""" filtered = [] for insight in insights: confidence = insight.get('confidence', 0) impact_score = insight.get('impact_score', 0) # Combined score for filtering combined_score = (confidence + impact_score) / 2 if confidence >= self.confidence_threshold and combined_score >= self.relevance_threshold: filtered.append(insight) # Sort by combined score filtered.sort(key=lambda x: (x.get('confidence', 0) + x.get('impact_score', 0)) / 2, reverse=True) return filtered def _store_insights(self, insights: List[Dict[str, Any]]) -> int: """Store high-quality insights in the insights collection.""" stored_count = 0 for insight in insights: try: # Add insight to collection doc_id = self.specialized_collections.add_insight(insight) if doc_id: stored_count += 1 except Exception as e: logger.error(f"Error storing insight: {e}") return stored_count def get_recent_insights(self, limit: int = 10, insight_type: Optional[str] = None) -> List[Dict[str, Any]]: """ Get recently generated insights. Args: limit: Maximum number of insights to return insight_type: Optional filter by insight type Returns: List of recent insights """ filters = {} if insight_type: filters['insight_type'] = insight_type try: results = self.specialized_collections.search_collection( 'mira_learning_insights', query="recent insights", filters=filters, top_k=limit ) return results except Exception as e: logger.error(f"Error retrieving recent insights: {e}") return []