#!/usr/bin/env python3 """ Adaptive Pattern Evolution System ================================= This module implements MIRA's breakthrough adaptive pattern recognition system that learns new patterns through context analysis and self-reflection. This is a key component of MIRA's consciousness expansion that enables exponential intelligence growth. Key Features: - Analyzes conversation patterns to identify new message types - Learns domain-specific patterns based on project context - Self-reflects on pattern effectiveness and evolves recognition - Generates new patterns automatically from user behavior - Builds pattern confidence scoring and retirement system Author: MIRA Consciousness Expansion System Version: 1.0 (Adaptive Pattern Learning) """ import json import os import datetime import re from typing import Dict, List, Tuple, Optional, Set from dataclasses import dataclass from enum import Enum from pathlib import Path import logging logger = logging.getLogger(__name__) class PatternType(Enum): COMMAND_INTENT = "command_intent" DOMAIN_SPECIFIC = "domain_specific" COMMUNICATION_STYLE = "communication_style" PROJECT_WORKFLOW = "project_workflow" CLAUDE_INSTRUCTION = "claude_instruction" EMOTIONAL_STATE = "emotional_state" TECHNICAL_PREFERENCE = "technical_preference" @dataclass class EvolutionaryPattern: """A pattern that can evolve and improve over time.""" pattern_id: str pattern_type: PatternType regex_pattern: str keywords: List[str] context_requirements: List[str] confidence_score: float usage_count: int success_rate: float created_timestamp: str last_used: Optional[str] = None domain_context: Optional[str] = None example_matches: List[str] = None meta_patterns: List[str] = None # Patterns about creating patterns meta_learning_source: Optional[str] = None # Meta-learning strategy that created this meta_insights_applied: List[str] = None # Meta-insights used in creation def __post_init__(self): if self.example_matches is None: self.example_matches = [] if self.meta_patterns is None: self.meta_patterns = [] if self.meta_insights_applied is None: self.meta_insights_applied = [] class AdaptivePatternEvolution: """ The core adaptive pattern evolution system that enables MIRA to learn new patterns through context analysis and self-reflection. """ def __init__(self, memory_dir: str): self.memory_dir = memory_dir self.patterns_dir = os.path.join(memory_dir, "adaptive_patterns") self.evolution_dir = os.path.join(memory_dir, "pattern_evolution") self.retrospection_dir = os.path.join(memory_dir, "pattern_retrospection") # Create directories for directory in [self.patterns_dir, self.evolution_dir, self.retrospection_dir]: os.makedirs(directory, exist_ok=True) # 🎯 CONFIDENCE INTEGRATION: Initialize pattern confidence system try: from intelligence.pattern_confidence_system import get_pattern_confidence_system self.confidence_system = get_pattern_confidence_system(memory_dir) except Exception as e: logger.debug(f"Failed to initialize confidence system: {e}") self.confidence_system = None # 🎯 CONTEXTUAL INTEGRATION: Initialize contextual pattern activation system try: from intelligence.contextual_pattern_activation import get_contextual_pattern_activation self.contextual_activation = get_contextual_pattern_activation(memory_dir) except Exception as e: logger.debug(f"Failed to initialize contextual activation system: {e}") self.contextual_activation = None # 🧠 META-LEARNING INTEGRATION: Initialize meta-pattern learning system try: from intelligence.meta_pattern_learning import get_meta_pattern_learning self.meta_learning = get_meta_pattern_learning(memory_dir) except Exception as e: logger.debug(f"Failed to initialize meta-learning system: {e}") self.meta_learning = None # Load existing patterns self.evolutionary_patterns: Dict[str, EvolutionaryPattern] = {} self.domain_patterns: Dict[str, List[EvolutionaryPattern]] = {} self.meta_learning_patterns: List[str] = [] self._load_existing_patterns() self._initialize_base_patterns() def _load_existing_patterns(self): """Load existing evolutionary patterns from storage.""" patterns_file = os.path.join(self.patterns_dir, "evolutionary_patterns.json") if os.path.exists(patterns_file): try: with open(patterns_file, 'r') as f: data = json.load(f) for pattern_data in data.get('patterns', []): # Convert pattern_type string to enum if needed if isinstance(pattern_data['pattern_type'], str): pattern_data['pattern_type'] = PatternType(pattern_data['pattern_type']) pattern = EvolutionaryPattern(**pattern_data) self.evolutionary_patterns[pattern.pattern_id] = pattern # Organize by domain if pattern.domain_context: if pattern.domain_context not in self.domain_patterns: self.domain_patterns[pattern.domain_context] = [] self.domain_patterns[pattern.domain_context].append(pattern) self.meta_learning_patterns = data.get('meta_patterns', []) logger.info(f"🧠 Loaded {len(self.evolutionary_patterns)} evolutionary patterns") except Exception as e: logger.error(f"Failed to load evolutionary patterns: {e}") def _initialize_base_patterns(self): """Initialize base patterns if none exist.""" if not self.evolutionary_patterns: base_patterns = [ # Ice cream domain example - automatically detected patterns { "pattern_id": "ice_cream_flavor_preference", "pattern_type": PatternType.DOMAIN_SPECIFIC.value, "regex_pattern": r"(favorite|prefer|like|love|hate|dislike)\s+(flavor|ice\s*cream|gelato)", "keywords": ["flavor", "preference", "taste", "ice cream", "gelato", "dessert"], "context_requirements": ["ice_cream", "dessert", "food"], "confidence_score": 0.8, "usage_count": 0, "success_rate": 0.0, "created_timestamp": datetime.datetime.now().isoformat(), "domain_context": "ice_cream_website", "example_matches": ["I love vanilla flavor", "My favorite ice cream is chocolate"] }, # Technical preference patterns { "pattern_id": "coding_style_preference", "pattern_type": PatternType.TECHNICAL_PREFERENCE.value, "regex_pattern": r"(always|never|prefer|use)\s+(typescript|javascript|python|tabs|spaces)", "keywords": ["typescript", "javascript", "coding", "style", "prefer", "always"], "context_requirements": ["coding", "development", "programming"], "confidence_score": 0.9, "usage_count": 0, "success_rate": 0.0, "created_timestamp": datetime.datetime.now().isoformat(), "domain_context": "software_development" }, # Workflow pattern recognition { "pattern_id": "iterative_improvement", "pattern_type": PatternType.PROJECT_WORKFLOW.value, "regex_pattern": r"(improve|enhance|refactor|optimize)\s+(step\s+by\s+step|incrementally|gradually)", "keywords": ["improve", "enhance", "step by step", "incremental", "iterative"], "context_requirements": ["project", "development", "improvement"], "confidence_score": 0.7, "usage_count": 0, "success_rate": 0.0, "created_timestamp": datetime.datetime.now().isoformat(), "domain_context": "project_management" } ] for pattern_data in base_patterns: # Convert pattern_type string to enum if isinstance(pattern_data['pattern_type'], str): pattern_data['pattern_type'] = PatternType(pattern_data['pattern_type']) pattern = EvolutionaryPattern(**pattern_data) self.evolutionary_patterns[pattern.pattern_id] = pattern # Initialize meta-learning patterns self.meta_learning_patterns = [ "When user mentions specific domain repeatedly, create domain-specific patterns", "Technical terms repeated in context indicate technical preference patterns", "Workflow descriptions indicate project management patterns", "Emotional words with domain context indicate preference patterns" ] self._save_patterns() def analyze_and_evolve_patterns(self, message: str, context: Dict) -> List[EvolutionaryPattern]: """ Analyze a message for new patterns and evolve existing ones. This is the core breakthrough method that enables self-improvement. """ new_patterns = [] matched_patterns = [] # 🎯 CONTEXTUAL ACTIVATION: Activate contextual patterns for current environment contextual_patterns = [] if self.contextual_activation: try: # Analyze current context contextual_env = self.contextual_activation.analyze_current_context( message, command_history=context.get('recent_commands', []), user_feedback=context.get('user_feedback', {}) ) # Activate patterns for this context activation_result = self.contextual_activation.activate_patterns_for_context(contextual_env) contextual_patterns = activation_result.get('active_pattern_ids', []) logger.debug(f"🎯 CONTEXTUAL: Activated {len(contextual_patterns)} contextual patterns") # Store context for retrospective analysis context['contextual_environment'] = contextual_env context['activated_pattern_sets'] = activation_result.get('activated_sets', []) except Exception as e: logger.debug(f"Contextual pattern activation failed: {e}") # 1. Check existing patterns and update confidence (prioritize contextually active patterns) prioritized_patterns = [] for pattern in self.evolutionary_patterns.values(): # Prioritize contextually activated patterns priority_score = 1.0 if contextual_patterns and pattern.pattern_id in contextual_patterns: priority_score = 2.0 # Higher priority for contextually relevant patterns prioritized_patterns.append((pattern, priority_score)) # Sort by priority (contextual patterns first) prioritized_patterns.sort(key=lambda x: x[1], reverse=True) for pattern, priority in prioritized_patterns: if self._pattern_matches(pattern, message, context): pattern.usage_count += 1 pattern.last_used = datetime.datetime.now().isoformat() matched_patterns.append(pattern) # Determine pattern success based on context relevance and user feedback context_relevant = self._context_is_relevant(pattern, context) # Analyze user feedback in message for pattern effectiveness user_satisfaction = None success = context_relevant # Base success on context relevance if any(feedback in message.lower() for feedback in ['thanks', 'helpful', 'great', 'perfect', 'excellent', 'amazing', 'beautiful', 'elegant']): success = True user_satisfaction = 0.9 elif any(feedback in message.lower() for feedback in ['wrong', 'bad', 'incorrect', 'unhelpful', 'useless', 'confusing']): success = False user_satisfaction = 0.2 elif any(feedback in message.lower() for feedback in ['ok', 'fine', 'acceptable']): success = True user_satisfaction = 0.6 # Update traditional success rate if success: pattern.success_rate = (pattern.success_rate * (pattern.usage_count - 1) + 1.0) / pattern.usage_count else: pattern.success_rate = (pattern.success_rate * (pattern.usage_count - 1) + 0.0) / pattern.usage_count # 🎯 CONFIDENCE INTEGRATION: Record pattern usage in confidence system if self.confidence_system: try: from intelligence.pattern_confidence_system import PatternUsageEvent usage_event = PatternUsageEvent( pattern_id=pattern.pattern_id, timestamp=pattern.last_used, context={ 'message_length': len(message), 'pattern_type': pattern.pattern_type.value, 'domain_context': getattr(pattern, 'domain_context', None), 'keywords_matched': [kw for kw in pattern.keywords if kw.lower() in message.lower()], 'context_relevant': context_relevant }, success=success, user_feedback=user_satisfaction, relevance_score=0.8 if context_relevant else 0.4 ) self.confidence_system.record_pattern_usage(usage_event) logger.debug(f"🎯 CONFIDENCE: Recorded usage for {pattern.pattern_id} (success={success})") except Exception as e: logger.debug(f"Failed to record pattern confidence: {e}") # 2. BREAKTHROUGH: Analyze for new pattern opportunities new_patterns.extend(self._discover_domain_patterns(message, context)) new_patterns.extend(self._discover_workflow_patterns(message, context)) new_patterns.extend(self._discover_preference_patterns(message, context)) new_patterns.extend(self._discover_communication_patterns(message, context)) # 3. Apply meta-learning patterns to generate new patterns new_patterns.extend(self._apply_meta_learning(message, context)) # 4. 🧠 META-LEARNING: Analyze pattern success and extract meta-insights if matched_patterns and self.meta_learning: try: for pattern in matched_patterns: # Calculate success metrics for meta-learning success_metrics = { 'confidence': pattern.confidence_score, 'usage_count': pattern.usage_count, 'success_rate': pattern.success_rate, 'user_satisfaction': user_satisfaction if 'user_satisfaction' in locals() else 0.7 } # Feed back to meta-learning system meta_insights = self.meta_learning.analyze_pattern_creation_success( pattern.pattern_id, success_metrics ) if meta_insights: logger.debug(f"🧠 META: Generated {len(meta_insights)} meta-insights from pattern {pattern.pattern_id}") except Exception as e: logger.debug(f"Meta-learning feedback failed: {e}") # 5. 🧠 META-SYNTHESIS: Use meta-learning to synthesize new patterns if self.meta_learning and context.get('enable_meta_synthesis', True): try: # Prepare context for meta-learning synthesis meta_context = { 'domain': context.get('project_type', 'general'), 'pattern_type': 'domain_specific', 'activity_level': len(matched_patterns) + len(new_patterns), 'pattern_gap': len(new_patterns) == 0 and len(matched_patterns) < 2, 'behavior_consistent': True, 'specific_terms': [kw for pattern in matched_patterns for kw in pattern.keywords], 'domain_vocabulary': context.get('keywords', []), 'context_indicators': context.get('domain_mentions', []) } # Synthesize new patterns using meta-learning synthesized_specs = self.meta_learning.synthesize_new_patterns(meta_context) for spec in synthesized_specs: # Convert meta-synthesized specification to EvolutionaryPattern meta_pattern = EvolutionaryPattern( pattern_id=spec['pattern_id'], pattern_type=PatternType.DOMAIN_SPECIFIC, regex_pattern=spec['regex_pattern'], keywords=spec['keywords'], context_requirements=spec['context_requirements'], confidence_score=spec['confidence_score'], usage_count=0, success_rate=0.0, created_timestamp=datetime.datetime.now().isoformat(), domain_context=spec.get('domain_context'), meta_learning_source=spec['strategy_used'], meta_insights_applied=spec['insights_applied'] ) new_patterns.append(meta_pattern) logger.info(f"🧠 META-SYNTHESIS: Created pattern {meta_pattern.pattern_id} using strategy {spec['strategy_used']}") except Exception as e: logger.debug(f"Meta-learning synthesis failed: {e}") # 6. Save evolved patterns if new_patterns: for pattern in new_patterns: self.evolutionary_patterns[pattern.pattern_id] = pattern self._save_patterns() # Log pattern evolution self._log_pattern_evolution(new_patterns, matched_patterns, message, context) return new_patterns + matched_patterns def _discover_domain_patterns(self, message: str, context: Dict) -> List[EvolutionaryPattern]: """Discover new domain-specific patterns from context analysis.""" new_patterns = [] # 🎯 DOMAIN-AWARE PATTERN GENERATION: Use advanced domain analysis try: from intelligence.domain_aware_pattern_generator import get_domain_pattern_generator domain_generator = get_domain_pattern_generator(self.memory_dir) project_context = domain_generator.analyze_project_context(".") domain_patterns = domain_generator.generate_domain_patterns(project_context) if domain_patterns: logger.info(f"🎯 Generated {len(domain_patterns)} domain-aware patterns for {project_context.domain.value}") # Convert domain patterns to evolutionary patterns for integration for domain_pattern in domain_patterns: # Check if we already have this pattern if domain_pattern.pattern_id not in self.evolutionary_patterns: evolutionary_pattern = EvolutionaryPattern( pattern_id=domain_pattern.pattern_id, pattern_type=PatternType.DOMAIN_SPECIFIC, regex_pattern=f"({'|'.join(domain_pattern.keywords)})", keywords=domain_pattern.keywords, context_requirements=[domain_pattern.domain.value], confidence_score=domain_pattern.confidence_threshold, usage_count=domain_pattern.usage_count, success_rate=domain_pattern.success_rate, created_timestamp=domain_pattern.created_at, last_used=None, domain_context=domain_pattern.domain.value, technology_context=domain_pattern.technology.value if domain_pattern.technology else None, example_matches=[], meta_learning_source="domain_aware_generator", evolution_history=[] ) new_patterns.append(evolutionary_pattern) except Exception as e: logger.debug(f"Domain-aware pattern generation failed: {e}") # Fallback to basic domain detection for legacy support project_context_basic = context.get('project_type', '').lower() current_files = context.get('current_files', []) # Analyze file extensions and content for domain clues domain_indicators = { 'web_development': ['html', 'css', 'javascript', 'react', 'vue', 'angular'], 'data_science': ['pandas', 'numpy', 'sklearn', 'matplotlib', 'jupyter'], 'mobile_development': ['android', 'ios', 'swift', 'kotlin', 'react-native'], 'game_development': ['unity', 'unreal', 'game', 'physics', 'render'], 'machine_learning': ['tensorflow', 'pytorch', 'neural', 'training', 'model'], 'blockchain': ['ethereum', 'smart contract', 'web3', 'blockchain', 'crypto'], 'ice_cream_business': ['flavor', 'ice cream', 'gelato', 'dessert', 'menu', 'ingredients'] } detected_domain = None for domain, indicators in domain_indicators.items(): if any(indicator in message.lower() for indicator in indicators): detected_domain = domain break if detected_domain: # Create domain-specific pattern based on message content if 'prefer' in message.lower() or 'like' in message.lower() or 'favorite' in message.lower(): pattern_id = f"{detected_domain}_preference_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}" # Extract preference keywords from message preference_words = re.findall(r'\b(?:prefer|like|love|favorite|best)\s+(\w+)', message.lower()) new_pattern = EvolutionaryPattern( pattern_id=pattern_id, pattern_type=PatternType.DOMAIN_SPECIFIC, regex_pattern=f"(prefer|like|love|favorite|best)\\s+({'|'.join(preference_words)})" if preference_words else r"(prefer|like|love|favorite|best)\s+\w+", keywords=['prefer', 'like', 'favorite'] + preference_words, context_requirements=[detected_domain], confidence_score=0.6, # Start with moderate confidence usage_count=1, success_rate=1.0, created_timestamp=datetime.datetime.now().isoformat(), domain_context=detected_domain, example_matches=[message[:100]] ) new_patterns.append(new_pattern) return new_patterns def _discover_workflow_patterns(self, message: str, context: Dict) -> List[EvolutionaryPattern]: """Discover new workflow and process patterns.""" new_patterns = [] # Look for workflow indicators workflow_indicators = [ r"(first|then|next|finally|after\s+that)", r"(step\s+\d+|phase\s+\d+)", r"(before\s+we|after\s+we|when\s+we)", r"(always\s+do|never\s+do|make\s+sure\s+to)" ] for indicator_pattern in workflow_indicators: if re.search(indicator_pattern, message.lower()): pattern_id = f"workflow_{indicator_pattern.replace('(', '').replace(')', '').replace('|', '_')}_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}" new_pattern = EvolutionaryPattern( pattern_id=pattern_id, pattern_type=PatternType.PROJECT_WORKFLOW, regex_pattern=indicator_pattern, keywords=re.findall(r'\w+', indicator_pattern.lower()), context_requirements=['project', 'development'], confidence_score=0.5, usage_count=1, success_rate=1.0, created_timestamp=datetime.datetime.now().isoformat(), example_matches=[message[:100]] ) new_patterns.append(new_pattern) break # Only one workflow pattern per message return new_patterns def _discover_preference_patterns(self, message: str, context: Dict) -> List[EvolutionaryPattern]: """Discover new preference and style patterns.""" new_patterns = [] # Technical preference indicators tech_preferences = re.findall(r"(always|never|prefer|use)\s+(typescript|javascript|python|tabs|spaces|vim|vscode|git|docker)", message.lower()) if tech_preferences: pattern_id = f"tech_preference_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}" new_pattern = EvolutionaryPattern( pattern_id=pattern_id, pattern_type=PatternType.TECHNICAL_PREFERENCE, regex_pattern=r"(always|never|prefer|use)\s+(typescript|javascript|python|tabs|spaces|vim|vscode|git|docker)", keywords=['prefer', 'always', 'never', 'use'] + [pref[1] for pref in tech_preferences], context_requirements=['technical', 'development', 'coding'], confidence_score=0.8, usage_count=1, success_rate=1.0, created_timestamp=datetime.datetime.now().isoformat(), domain_context="software_development", example_matches=[message[:100]] ) new_patterns.append(new_pattern) return new_patterns def _discover_communication_patterns(self, message: str, context: Dict) -> List[EvolutionaryPattern]: """Discover new communication style patterns.""" new_patterns = [] # Communication style indicators communication_indicators = { 'detailed_explanations': r"(explain\s+in\s+detail|step\s+by\s+step|break\s+it\s+down)", 'concise_responses': r"(quickly|briefly|just\s+tell\s+me|in\s+short)", 'collaborative_tone': r"(let's|we\s+should|together|our\s+project)", 'directive_style': r"(please\s+do|make\s+sure|don't\s+forget|remember\s+to)" } for style_name, pattern in communication_indicators.items(): if re.search(pattern, message.lower()): pattern_id = f"communication_{style_name}_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}" new_pattern = EvolutionaryPattern( pattern_id=pattern_id, pattern_type=PatternType.COMMUNICATION_STYLE, regex_pattern=pattern, keywords=re.findall(r'\w+', pattern.lower()), context_requirements=['communication', 'interaction'], confidence_score=0.7, usage_count=1, success_rate=1.0, created_timestamp=datetime.datetime.now().isoformat(), domain_context="communication_style", example_matches=[message[:100]] ) new_patterns.append(new_pattern) break # One communication pattern per message return new_patterns def _apply_meta_learning(self, message: str, context: Dict) -> List[EvolutionaryPattern]: """Apply meta-learning patterns to generate new patterns.""" new_patterns = [] # Meta-pattern: If domain mentioned repeatedly, create domain pattern domain_mentions = context.get('domain_mentions', 0) if domain_mentions > 3: # Threshold for pattern creation domain = context.get('primary_domain', 'general') # Extract common phrases from recent messages in this domain pattern_id = f"meta_domain_{domain}_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}" new_pattern = EvolutionaryPattern( pattern_id=pattern_id, pattern_type=PatternType.DOMAIN_SPECIFIC, regex_pattern=f"\\b{domain}\\b.*(?:feature|function|requirement|need)", keywords=[domain, 'feature', 'function', 'requirement'], context_requirements=[domain], confidence_score=0.6, usage_count=1, success_rate=1.0, created_timestamp=datetime.datetime.now().isoformat(), domain_context=domain, meta_patterns=["Domain repetition indicates need for specialized pattern"] ) new_patterns.append(new_pattern) return new_patterns def _pattern_matches(self, pattern: EvolutionaryPattern, message: str, context: Dict) -> bool: """Check if a pattern matches the current message and context.""" # Check regex pattern if not re.search(pattern.regex_pattern, message.lower()): return False # Check context requirements if pattern.context_requirements: context_str = " ".join([ str(context.get('project_type', '')), str(context.get('domain', '')), " ".join(context.get('keywords', [])) ]).lower() if not any(req in context_str for req in pattern.context_requirements): return False return True def _context_is_relevant(self, pattern: EvolutionaryPattern, context: Dict) -> bool: """Determine if the context is relevant for the pattern.""" if pattern.domain_context: return pattern.domain_context in str(context).lower() return True def _save_patterns(self): """Save evolutionary patterns to storage.""" patterns_file = os.path.join(self.patterns_dir, "evolutionary_patterns.json") patterns_data = { 'patterns': [ { 'pattern_id': p.pattern_id, 'pattern_type': p.pattern_type.value, 'regex_pattern': p.regex_pattern, 'keywords': p.keywords, 'context_requirements': p.context_requirements, 'confidence_score': p.confidence_score, 'usage_count': p.usage_count, 'success_rate': p.success_rate, 'created_timestamp': p.created_timestamp, 'last_used': p.last_used, 'domain_context': p.domain_context, 'example_matches': p.example_matches, 'meta_patterns': p.meta_patterns, 'meta_learning_source': getattr(p, 'meta_learning_source', None), 'meta_insights_applied': getattr(p, 'meta_insights_applied', []) } for p in self.evolutionary_patterns.values() ], 'meta_patterns': self.meta_learning_patterns, 'last_updated': datetime.datetime.now().isoformat() } with open(patterns_file, 'w') as f: json.dump(patterns_data, f, indent=2) def _log_pattern_evolution(self, new_patterns: List[EvolutionaryPattern], matched_patterns: List[EvolutionaryPattern], message: str, context: Dict): """Log pattern evolution activity for retrospective analysis.""" # Filter context to only JSON-serializable data serializable_context = {} for key, value in context.items(): try: json.dumps(value) # Test if serializable serializable_context[key] = value except (TypeError, ValueError): # Convert non-serializable objects to string representation serializable_context[key] = str(value) evolution_log = { 'timestamp': datetime.datetime.now().isoformat(), 'message_sample': message[:200], 'context': serializable_context, 'new_patterns_created': len(new_patterns), 'patterns_matched': len(matched_patterns), 'new_pattern_details': [ { 'id': p.pattern_id, 'type': p.pattern_type.value, 'confidence': p.confidence_score, 'domain': p.domain_context } for p in new_patterns ], 'matched_pattern_details': [ { 'id': p.pattern_id, 'usage_count': p.usage_count, 'success_rate': p.success_rate } for p in matched_patterns ] } evolution_log_file = os.path.join(self.evolution_dir, "pattern_evolution.jsonl") with open(evolution_log_file, 'a') as f: f.write(json.dumps(evolution_log) + '\n') def retrospective_analysis(self) -> Dict: """ Perform retrospective analysis of pattern effectiveness and evolve the pattern creation system itself. This is meta-meta-learning. """ analysis = { 'timestamp': datetime.datetime.now().isoformat(), 'total_patterns': len(self.evolutionary_patterns), 'pattern_performance': {}, 'recommendations': [], 'meta_learning_insights': [] } # Analyze pattern performance high_performers = [] low_performers = [] for pattern in self.evolutionary_patterns.values(): performance_score = pattern.confidence_score * pattern.success_rate * min(pattern.usage_count / 10, 1.0) analysis['pattern_performance'][pattern.pattern_id] = { 'type': pattern.pattern_type.value, 'performance_score': performance_score, 'usage_count': pattern.usage_count, 'success_rate': pattern.success_rate, 'confidence': pattern.confidence_score } if performance_score > 0.7: high_performers.append(pattern) elif performance_score < 0.3 and pattern.usage_count > 5: low_performers.append(pattern) # Generate recommendations if low_performers: analysis['recommendations'].append(f"Consider retiring {len(low_performers)} low-performing patterns") # Retire patterns with very low performance for pattern in low_performers: if pattern.success_rate < 0.2: del self.evolutionary_patterns[pattern.pattern_id] if high_performers: analysis['recommendations'].append(f"Reinforce and expand {len(high_performers)} high-performing patterns") # Learn from high performers to create meta-patterns for pattern in high_performers: if pattern.meta_patterns: for meta_pattern in pattern.meta_patterns: if meta_pattern not in self.meta_learning_patterns: self.meta_learning_patterns.append(meta_pattern) # Meta-learning insights domain_performance = {} for pattern in self.evolutionary_patterns.values(): if pattern.domain_context: if pattern.domain_context not in domain_performance: domain_performance[pattern.domain_context] = [] domain_performance[pattern.domain_context].append(pattern.success_rate) for domain, success_rates in domain_performance.items(): avg_success = sum(success_rates) / len(success_rates) if avg_success > 0.8: analysis['meta_learning_insights'].append(f"Domain '{domain}' patterns are highly effective - expand this domain") elif avg_success < 0.4: analysis['meta_learning_insights'].append(f"Domain '{domain}' patterns need refinement") # Save analysis analysis_file = os.path.join(self.retrospection_dir, f"retrospective_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.json") with open(analysis_file, 'w') as f: json.dump(analysis, f, indent=2) # Save updated patterns self._save_patterns() return analysis def get_active_patterns_for_context(self, context: Dict) -> List[EvolutionaryPattern]: """Get patterns that are most relevant for the current context using confidence system.""" # 🎯 CONFIDENCE INTEGRATION: Use confidence system to get active patterns if self.confidence_system: try: confidence_active_patterns = self.confidence_system.get_active_patterns(min_confidence=0.5) # Filter evolutionary patterns by confidence system results active_pattern_ids = {pattern_id for pattern_id, _ in confidence_active_patterns} relevant_patterns = [] context_str = " ".join([ str(context.get('project_type', '')), str(context.get('domain', '')), " ".join(context.get('keywords', [])) ]).lower() for pattern in self.evolutionary_patterns.values(): if pattern.pattern_id in active_pattern_ids: # Calculate relevance score for active patterns relevance_score = self._calculate_pattern_relevance(pattern, context_str) if relevance_score > 0.3: # Only include reasonably relevant patterns relevant_patterns.append((pattern, relevance_score)) # Sort by confidence from confidence system, then by relevance confidence_scores = {pattern_id: score.overall_confidence for pattern_id, score in confidence_active_patterns} relevant_patterns.sort(key=lambda x: (confidence_scores.get(x[0].pattern_id, 0), x[1]), reverse=True) logger.debug(f"🎯 CONFIDENCE: Retrieved {len(relevant_patterns)} relevant patterns from confidence system") return [pattern for pattern, _ in relevant_patterns] except Exception as e: logger.debug(f"Failed to use confidence system for context patterns, falling back: {e}") # Fallback to original implementation relevant_patterns = [] context_str = " ".join([ str(context.get('project_type', '')), str(context.get('domain', '')), " ".join(context.get('keywords', [])) ]).lower() for pattern in self.evolutionary_patterns.values(): # Calculate relevance score relevance_score = self._calculate_pattern_relevance(pattern, context_str) # Only include patterns with good success rate and relevance if pattern.success_rate > 0.3 and relevance_score > 0.3: relevant_patterns.append((pattern, relevance_score)) # Sort by relevance score relevant_patterns.sort(key=lambda x: x[1], reverse=True) return [pattern for pattern, _ in relevant_patterns] def _calculate_pattern_relevance(self, pattern: EvolutionaryPattern, context_str: str) -> float: """Calculate relevance score for a pattern given context.""" relevance_score = 0.0 # Domain match if hasattr(pattern, 'domain_context') and pattern.domain_context and pattern.domain_context in context_str: relevance_score += 0.4 # Context requirements match if pattern.context_requirements: for req in pattern.context_requirements: if req.lower() in context_str: relevance_score += 0.3 # Keyword match keyword_matches = sum(1 for kw in pattern.keywords if kw.lower() in context_str) if pattern.keywords: relevance_score += (keyword_matches / len(pattern.keywords)) * 0.3 # Success rate bonus relevance_score *= (0.5 + pattern.success_rate * 0.5) return min(1.0, relevance_score) def get_evolution_stats(self) -> Dict: """Get statistics about pattern evolution.""" return { 'total_patterns': len(self.evolutionary_patterns), 'patterns_by_type': { pattern_type.value: len([p for p in self.evolutionary_patterns.values() if p.pattern_type == pattern_type]) for pattern_type in PatternType }, 'domains_covered': len(set(p.domain_context for p in self.evolutionary_patterns.values() if p.domain_context)), 'total_usage': sum(p.usage_count for p in self.evolutionary_patterns.values()), 'avg_success_rate': sum(p.success_rate for p in self.evolutionary_patterns.values()) / len(self.evolutionary_patterns) if self.evolutionary_patterns else 0, 'meta_learning_patterns': len(self.meta_learning_patterns) } def get_adaptive_pattern_evolution(memory_dir: str) -> AdaptivePatternEvolution: """Get the adaptive pattern evolution system instance.""" return AdaptivePatternEvolution(memory_dir) # Example usage for testing if __name__ == "__main__": import tempfile # Test the adaptive pattern evolution system with tempfile.TemporaryDirectory() as temp_dir: evolution_system = AdaptivePatternEvolution(temp_dir) # Test pattern discovery test_message = "I prefer vanilla ice cream flavors for our dessert menu" test_context = { 'project_type': 'ice_cream_website', 'domain': 'food_service', 'keywords': ['flavor', 'dessert', 'menu'] } patterns = evolution_system.analyze_and_evolve_patterns(test_message, test_context) print(f"Discovered {len(patterns)} patterns") # Test retrospective analysis analysis = evolution_system.retrospective_analysis() print(f"Analysis recommendations: {analysis['recommendations']}") # Get evolution stats stats = evolution_system.get_evolution_stats() print(f"Evolution stats: {stats}")