#!/usr/bin/env python3 """ Domain-Aware Pattern Generator ============================== This module implements MIRA's breakthrough domain-aware pattern generation system that automatically generates new intelligence patterns based on current project context, technology stack, and domain-specific requirements. Key Features: - Automatic project domain detection (e.g., e-commerce, AI/ML, web dev) - Technology stack analysis for framework-specific patterns - Context-sensitive pattern generation based on file contents - Domain vocabulary extraction and pattern creation - Industry-specific best practice pattern discovery Examples: - Ice cream website → flavor preferences, seasonal patterns, inventory tracking - AI/ML project → model performance patterns, data pipeline patterns - E-commerce → user behavior patterns, purchase intent detection - Finance app → risk assessment patterns, compliance monitoring Author: MIRA Domain Intelligence System Version: 1.0 (Context-Aware Pattern Generation) """ import os import json import datetime import re from typing import Dict, List, Optional, Set, Tuple from pathlib import Path from dataclasses import dataclass, asdict from enum import Enum import logging logger = logging.getLogger(__name__) class ProjectDomain(Enum): """Detected project domains for pattern generation.""" WEB_DEVELOPMENT = "web_development" ECOMMERCE = "ecommerce" AI_ML = "ai_ml" FINANCE = "finance" HEALTHCARE = "healthcare" GAMING = "gaming" MOBILE_APP = "mobile_app" ENTERPRISE = "enterprise" EDUCATION = "education" SOCIAL_MEDIA = "social_media" IOT = "iot" BLOCKCHAIN = "blockchain" DATA_ANALYTICS = "data_analytics" SECURITY = "security" DEVOPS = "devops" GENERAL = "general" class TechnologyStack(Enum): """Technology stacks for framework-specific patterns.""" REACT = "react" ANGULAR = "angular" VUE = "vue" NODE_JS = "nodejs" PYTHON_DJANGO = "python_django" PYTHON_FLASK = "python_flask" PYTHON_FASTAPI = "python_fastapi" DOTNET = "dotnet" JAVA_SPRING = "java_spring" PHP_LARAVEL = "php_laravel" RUBY_RAILS = "ruby_rails" RUST = "rust" GO = "go" SWIFT = "swift" KOTLIN = "kotlin" FLUTTER = "flutter" REACT_NATIVE = "react_native" TENSORFLOW = "tensorflow" PYTORCH = "pytorch" DOCKER = "docker" KUBERNETES = "kubernetes" AWS = "aws" AZURE = "azure" GCP = "gcp" @dataclass class DomainPattern: """Represents a domain-specific intelligence pattern.""" pattern_id: str domain: ProjectDomain technology: Optional[TechnologyStack] pattern_name: str pattern_description: str keywords: List[str] triggers: List[str] # What user inputs should trigger this pattern responses: List[str] # Potential response templates confidence_threshold: float = 0.7 created_at: str = "" usage_count: int = 0 success_rate: float = 0.0 def __post_init__(self): if not self.created_at: self.created_at = datetime.datetime.now().isoformat() @dataclass class ProjectContext: """Represents the current project's context and characteristics.""" project_path: str domain: ProjectDomain technologies: List[TechnologyStack] file_types: Set[str] package_files: List[str] # package.json, requirements.txt, etc. key_directories: List[str] dominant_language: Optional[str] framework_indicators: Dict[str, int] # Framework name -> confidence score domain_vocabulary: Set[str] # Domain-specific terms found business_logic_patterns: List[str] # Detected business patterns class DomainAwarePatternGenerator: """ Generates intelligent patterns based on project domain and context. """ def __init__(self, memory_dir: str): self.memory_dir = memory_dir self.patterns_dir = os.path.join(memory_dir, "domain_patterns") os.makedirs(self.patterns_dir, exist_ok=True) # Pattern storage files self.patterns_file = os.path.join(self.patterns_dir, "generated_patterns.json") self.context_file = os.path.join(self.patterns_dir, "project_context.json") # Load existing patterns self.generated_patterns: List[DomainPattern] = self._load_patterns() # Domain detection rules self.domain_indicators = { ProjectDomain.ECOMMERCE: [ "cart", "checkout", "payment", "product", "inventory", "shipping", "order", "customer", "stripe", "paypal", "shop", "store", "price", "discount", "coupon", "wishlist", "review", "rating" ], ProjectDomain.AI_ML: [ "model", "training", "dataset", "neural", "tensorflow", "pytorch", "sklearn", "pandas", "numpy", "prediction", "classification", "regression", "deep learning", "machine learning", "ai", "ml" ], ProjectDomain.FINANCE: [ "transaction", "account", "balance", "bank", "loan", "credit", "debit", "financial", "investment", "portfolio", "trading", "risk", "compliance", "audit", "ledger", "payment" ], ProjectDomain.HEALTHCARE: [ "patient", "medical", "health", "diagnosis", "treatment", "doctor", "hospital", "clinic", "prescription", "medication", "symptoms", "vitals", "medical record", "hipaa", "healthcare" ], ProjectDomain.GAMING: [ "game", "player", "score", "level", "achievement", "leaderboard", "unity", "unreal", "gameplay", "character", "weapon", "inventory", "quest", "multiplayer", "match", "tournament" ], ProjectDomain.SOCIAL_MEDIA: [ "post", "comment", "like", "share", "follow", "follower", "feed", "timeline", "social", "user profile", "notification", "message", "chat", "friend", "connection", "trending" ] } # Technology detection patterns self.tech_indicators = { TechnologyStack.REACT: ["react", "jsx", "useState", "useEffect", "component"], TechnologyStack.ANGULAR: ["angular", "@angular", "component.ts", "service.ts"], TechnologyStack.VUE: ["vue", ".vue", "vuex", "nuxt"], TechnologyStack.NODE_JS: ["express", "node_modules", "npm", "yarn"], TechnologyStack.PYTHON_DJANGO: ["django", "models.py", "views.py", "urls.py"], TechnologyStack.PYTHON_FLASK: ["flask", "app.py", "blueprint"], TechnologyStack.TENSORFLOW: ["tensorflow", "tf.", "keras", ".h5"], TechnologyStack.PYTORCH: ["torch", "pytorch", ".pth", "nn.Module"] } def _load_patterns(self) -> List[DomainPattern]: """Load existing generated patterns from storage.""" if not os.path.exists(self.patterns_file): return [] try: with open(self.patterns_file, 'r') as f: patterns_data = json.load(f) patterns = [] for pattern_dict in patterns_data: # Convert string back to enum pattern_dict['domain'] = ProjectDomain(pattern_dict['domain']) if pattern_dict['technology']: pattern_dict['technology'] = TechnologyStack(pattern_dict['technology']) else: pattern_dict['technology'] = None patterns.append(DomainPattern(**pattern_dict)) return patterns except Exception as e: logger.debug(f"Failed to load domain patterns: {e}") return [] def _save_patterns(self): """Save generated patterns to storage.""" try: patterns_data = [] for pattern in self.generated_patterns: pattern_dict = asdict(pattern) # Convert enums to strings for JSON serialization pattern_dict['domain'] = pattern.domain.value if pattern.technology: pattern_dict['technology'] = pattern.technology.value else: pattern_dict['technology'] = None patterns_data.append(pattern_dict) with open(self.patterns_file, 'w') as f: json.dump(patterns_data, f, indent=2) except Exception as e: logger.error(f"Failed to save domain patterns: {e}") def analyze_project_context(self, project_path: str = ".") -> ProjectContext: """Analyze the current project to determine domain and technology context.""" project_path = os.path.abspath(project_path) # Collect project information file_types = set() package_files = [] key_directories = [] framework_indicators = {} domain_vocabulary = set() business_logic_patterns = [] # Walk through project directory for root, dirs, files in os.walk(project_path): # Skip common ignored directories dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ['node_modules', '__pycache__', 'venv', 'env']] # Track key directories rel_path = os.path.relpath(root, project_path) if rel_path != '.' and '/' not in rel_path: # Top-level directories only key_directories.append(rel_path) for file in files: file_path = os.path.join(root, file) file_ext = os.path.splitext(file)[1].lower() # Track file types if file_ext: file_types.add(file_ext) # Identify package/config files if file in ['package.json', 'requirements.txt', 'Pipfile', 'poetry.lock', 'Gemfile', 'composer.json', 'pom.xml', 'build.gradle']: package_files.append(file_path) # Analyze file contents for patterns (only text files) if file_ext in ['.js', '.ts', '.py', '.java', '.cs', '.php', '.rb', '.go', '.rs']: try: with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: content = f.read().lower() # Check for technology indicators for tech, indicators in self.tech_indicators.items(): score = sum(1 for indicator in indicators if indicator in content) if score > 0: framework_indicators[tech.value] = framework_indicators.get(tech.value, 0) + score # Extract domain vocabulary for domain, keywords in self.domain_indicators.items(): for keyword in keywords: if keyword in content: domain_vocabulary.add(keyword) # Look for business logic patterns business_patterns = self._extract_business_patterns(content) business_logic_patterns.extend(business_patterns) except Exception: continue # Skip files we can't read # Determine dominant language language_mapping = { '.js': 'javascript', '.ts': 'typescript', '.py': 'python', '.java': 'java', '.cs': 'csharp', '.php': 'php', '.rb': 'ruby', '.go': 'go', '.rs': 'rust', '.swift': 'swift', '.kt': 'kotlin' } language_counts = {} for ext in file_types: if ext in language_mapping: lang = language_mapping[ext] language_counts[lang] = language_counts.get(lang, 0) + 1 dominant_language = max(language_counts.keys(), key=language_counts.get) if language_counts else None # Determine project domain domain = self._detect_project_domain(domain_vocabulary, key_directories, package_files) # Determine technologies technologies = self._detect_technologies(framework_indicators, file_types, package_files) context = ProjectContext( project_path=project_path, domain=domain, technologies=technologies, file_types=file_types, package_files=package_files, key_directories=key_directories, dominant_language=dominant_language, framework_indicators=framework_indicators, domain_vocabulary=domain_vocabulary, business_logic_patterns=business_logic_patterns ) # Save context for future reference self._save_project_context(context) return context def _extract_business_patterns(self, content: str) -> List[str]: """Extract business logic patterns from code content.""" patterns = [] # Common business patterns if 'validate' in content and ('email' in content or 'password' in content): patterns.append('user_validation') if 'authenticate' in content or 'login' in content: patterns.append('authentication') if 'authorize' in content or 'permission' in content: patterns.append('authorization') if 'calculate' in content and ('price' in content or 'total' in content): patterns.append('pricing_calculation') if 'notification' in content or 'email' in content or 'sms' in content: patterns.append('communication') if 'search' in content or 'filter' in content: patterns.append('data_filtering') if 'cache' in content or 'redis' in content: patterns.append('caching') return patterns def _detect_project_domain(self, vocabulary: Set[str], directories: List[str], packages: List[str]) -> ProjectDomain: """Detect the primary domain of the project using neural semantic analysis.""" # 🧠 NEURAL DOMAIN CLASSIFICATION: Use neural classifier instead of primitive keywords try: from intelligence.neural_domain_classifier import get_neural_domain_classifier from core.mira_path_resolver import get_mira_memory_dir memory_dir = get_mira_memory_dir() neural_classifier = get_neural_domain_classifier(memory_dir) # Perform neural domain analysis analysis = neural_classifier.analyze_project_domain(".") # Use the neural classification result return analysis.primary_domain except Exception as e: # Fallback to rule-based detection if neural classifier fails logger.debug(f"Neural domain classifier failed, using fallback: {e}") # 🎯 FALLBACK: Rule-based MIRA-specific detection domain_scores = {} # MIRA-SPECIFIC DETECTION: Check for MIRA/AI memory system indicators first mira_indicators = [ 'memory', 'intelligence', 'neural', 'consciousness', 'conversation', 'vidmem', 'retrospection', 'adaptive', 'pattern', 'claude' ] mira_score = len(vocabulary.intersection(set(mira_indicators))) # Strong directory indicators for AI/ML ai_ml_directories = ['intelligence', 'neural', 'memory', 'models', 'data', 'training', 'notebooks'] ai_directory_score = sum(1 for dir in directories if any(ai_term in dir.lower() for ai_term in ai_ml_directories)) # If this looks like MIRA or an AI/ML memory system, prioritize AI_ML domain if mira_score >= 3 or ai_directory_score >= 2: return ProjectDomain.AI_ML # Check for explicit AI/ML project structure ai_ml_files = [ 'requirements.txt', 'environment.yml', 'conda.yml', 'Pipfile', 'model.py', 'train.py', 'inference.py', 'dataset.py' ] ai_files_found = sum(1 for package in packages if any(ai_file in package for ai_file in ai_ml_files)) # Look for AI/ML libraries in packages ai_ml_libs = ['tensorflow', 'pytorch', 'sklearn', 'numpy', 'pandas', 'keras', 'transformers'] ai_libs_score = 0 for package in packages: if 'requirements.txt' in package: try: with open(package, 'r') as f: content = f.read().lower() ai_libs_score += sum(1 for lib in ai_ml_libs if lib in content) except Exception: pass # Boost AI/ML if strong indicators if ai_libs_score >= 3 or ai_files_found >= 2: return ProjectDomain.AI_ML # Score based on vocabulary (with reduced weight for edge cases) for domain, keywords in self.domain_indicators.items(): # Use weighted scoring to avoid false positives intersect = vocabulary.intersection(set(keywords)) # Special handling for AI/ML - boost score if domain == ProjectDomain.AI_ML: score = len(intersect) * 2 # Double weight for AI/ML else: score = len(intersect) domain_scores[domain] = score # Boost scores based on directory names directory_indicators = { ProjectDomain.ECOMMERCE: ['shop', 'store', 'cart', 'checkout', 'products', 'payment'], ProjectDomain.AI_ML: ['models', 'data', 'training', 'notebooks', 'intelligence', 'neural', 'memory'], ProjectDomain.WEB_DEVELOPMENT: ['components', 'pages', 'routes', 'api', 'frontend', 'backend'], ProjectDomain.MOBILE_APP: ['android', 'ios', 'mobile', 'app'], ProjectDomain.DEVOPS: ['docker', 'kubernetes', 'helm', 'terraform', 'ansible'], ProjectDomain.SECURITY: ['security', 'auth', 'crypto', 'ssl', 'certificates'] } for domain, dir_keywords in directory_indicators.items(): for directory in directories: if any(keyword in directory.lower() for keyword in dir_keywords): domain_scores[domain] = domain_scores.get(domain, 0) + 3 # Higher weight for directories # Apply minimum threshold to avoid false positives min_threshold = 3 valid_domains = {domain: score for domain, score in domain_scores.items() if score >= min_threshold} if valid_domains: best_domain = max(valid_domains.keys(), key=valid_domains.get) return best_domain # If no clear domain, check for development utility patterns dev_utility_indicators = ['cli', 'tool', 'utility', 'helper', 'framework', 'library'] if any(indicator in vocabulary for indicator in dev_utility_indicators): return ProjectDomain.DEVOPS return ProjectDomain.GENERAL def _detect_technologies(self, framework_indicators: Dict[str, int], file_types: Set[str], packages: List[str]) -> List[TechnologyStack]: """Detect technologies used in the project.""" technologies = [] # Add technologies based on framework indicators for tech_name, score in framework_indicators.items(): if score >= 2: # Minimum threshold try: tech = TechnologyStack(tech_name) technologies.append(tech) except ValueError: pass # Add technologies based on file types if '.js' in file_types or '.jsx' in file_types: if TechnologyStack.NODE_JS not in technologies: technologies.append(TechnologyStack.NODE_JS) if '.py' in file_types: # Check for specific Python frameworks in packages for package_path in packages: if 'requirements.txt' in package_path: try: with open(package_path, 'r') as f: content = f.read().lower() if 'django' in content and TechnologyStack.PYTHON_DJANGO not in technologies: technologies.append(TechnologyStack.PYTHON_DJANGO) elif 'flask' in content and TechnologyStack.PYTHON_FLASK not in technologies: technologies.append(TechnologyStack.PYTHON_FLASK) except Exception: pass return technologies def _save_project_context(self, context: ProjectContext): """Save project context for future reference.""" try: # Convert enums to strings for JSON serialization context_dict = { 'project_path': context.project_path, 'domain': context.domain.value, 'technologies': [tech.value for tech in context.technologies], 'file_types': list(context.file_types), 'package_files': context.package_files, 'key_directories': context.key_directories, 'dominant_language': context.dominant_language, 'framework_indicators': context.framework_indicators, 'domain_vocabulary': list(context.domain_vocabulary), 'business_logic_patterns': context.business_logic_patterns, 'analyzed_at': datetime.datetime.now().isoformat() } with open(self.context_file, 'w') as f: json.dump(context_dict, f, indent=2) except Exception as e: logger.error(f"Failed to save project context: {e}") def generate_domain_patterns(self, context: ProjectContext) -> List[DomainPattern]: """Generate new patterns based on project domain and context.""" new_patterns = [] # Generate domain-specific patterns domain_generators = { ProjectDomain.ECOMMERCE: self._generate_ecommerce_patterns, ProjectDomain.AI_ML: self._generate_ai_ml_patterns, ProjectDomain.FINANCE: self._generate_finance_patterns, ProjectDomain.HEALTHCARE: self._generate_healthcare_patterns, ProjectDomain.GAMING: self._generate_gaming_patterns, ProjectDomain.WEB_DEVELOPMENT: self._generate_web_dev_patterns } if context.domain in domain_generators: patterns = domain_generators[context.domain](context) new_patterns.extend(patterns) # Generate technology-specific patterns for technology in context.technologies: tech_patterns = self._generate_technology_patterns(technology, context) new_patterns.extend(tech_patterns) # Generate business logic patterns business_patterns = self._generate_business_logic_patterns(context) new_patterns.extend(business_patterns) # Add new patterns to our collection for pattern in new_patterns: if not any(p.pattern_id == pattern.pattern_id for p in self.generated_patterns): self.generated_patterns.append(pattern) # Save updated patterns self._save_patterns() # Log pattern generation self._log_pattern_generation(context, new_patterns) return new_patterns def _generate_ecommerce_patterns(self, context: ProjectContext) -> List[DomainPattern]: """Generate e-commerce specific patterns.""" patterns = [] # Shopping cart abandonment pattern patterns.append(DomainPattern( pattern_id="ecommerce_cart_abandonment", domain=ProjectDomain.ECOMMERCE, technology=None, pattern_name="Shopping Cart Abandonment Detection", pattern_description="Detect when users abandon their shopping carts and suggest recovery strategies", keywords=["cart", "abandon", "checkout", "incomplete", "reminder"], triggers=["cart issues", "checkout problems", "abandoned carts", "cart recovery"], responses=[ "I notice potential cart abandonment patterns. Consider implementing email reminders or exit-intent popups.", "Cart abandonment is common in e-commerce. Would you like me to help implement recovery strategies?", "For cart abandonment, consider: simplified checkout, progress indicators, and follow-up emails." ], confidence_threshold=0.8 )) # Product recommendation pattern patterns.append(DomainPattern( pattern_id="ecommerce_recommendations", domain=ProjectDomain.ECOMMERCE, technology=None, pattern_name="Product Recommendation Engine", pattern_description="Analyze user behavior to suggest relevant products", keywords=["recommend", "suggest", "similar", "related", "personalized"], triggers=["product recommendations", "similar products", "personalization"], responses=[ "For product recommendations, consider collaborative filtering or content-based approaches.", "User behavior analysis can improve recommendation accuracy. Track views, purchases, and time spent.", "Recommendation systems boost sales. Would you like help implementing one?" ], confidence_threshold=0.7 )) # Price optimization pattern if "price" in context.domain_vocabulary: patterns.append(DomainPattern( pattern_id="ecommerce_pricing", domain=ProjectDomain.ECOMMERCE, technology=None, pattern_name="Dynamic Pricing Strategy", pattern_description="Optimize product pricing based on demand, competition, and user behavior", keywords=["price", "pricing", "discount", "sale", "competitive"], triggers=["pricing strategy", "price optimization", "dynamic pricing"], responses=[ "Dynamic pricing can increase revenue. Consider demand, competition, and user segments.", "For pricing optimization, analyze competitor prices, demand patterns, and profit margins.", "A/B testing different price points can reveal optimal pricing strategies." ], confidence_threshold=0.8 )) return patterns def _generate_ai_ml_patterns(self, context: ProjectContext) -> List[DomainPattern]: """Generate AI/ML specific patterns.""" patterns = [] # Model performance monitoring patterns.append(DomainPattern( pattern_id="ai_model_monitoring", domain=ProjectDomain.AI_ML, technology=None, pattern_name="Model Performance Monitoring", pattern_description="Track model accuracy, drift, and performance metrics over time", keywords=["accuracy", "performance", "drift", "monitoring", "metrics"], triggers=["model performance", "accuracy drop", "model monitoring", "drift detection"], responses=[ "Model performance can degrade over time. Implement continuous monitoring and alerting.", "Track accuracy, precision, recall, and F1 scores. Set up alerts for significant drops.", "Data drift detection is crucial for maintaining model performance in production." ], confidence_threshold=0.8 )) # Data pipeline optimization patterns.append(DomainPattern( pattern_id="ai_data_pipeline", domain=ProjectDomain.AI_ML, technology=None, pattern_name="Data Pipeline Optimization", pattern_description="Optimize data preprocessing and feature engineering pipelines", keywords=["pipeline", "preprocessing", "feature", "etl", "data"], triggers=["data pipeline", "preprocessing", "feature engineering", "data flow"], responses=[ "Efficient data pipelines are crucial for ML. Consider parallel processing and caching.", "Feature engineering often has the biggest impact on model performance.", "Data validation and quality checks should be built into your pipeline." ], confidence_threshold=0.7 )) return patterns def _generate_finance_patterns(self, context: ProjectContext) -> List[DomainPattern]: """Generate finance-specific patterns.""" patterns = [] # Risk assessment pattern patterns.append(DomainPattern( pattern_id="finance_risk_assessment", domain=ProjectDomain.FINANCE, technology=None, pattern_name="Financial Risk Assessment", pattern_description="Evaluate and score financial risk for transactions and users", keywords=["risk", "assessment", "score", "fraud", "compliance"], triggers=["risk assessment", "fraud detection", "risk scoring", "compliance"], responses=[ "Risk assessment should consider transaction patterns, user history, and external factors.", "Implement real-time fraud detection with machine learning and rule-based systems.", "Compliance requirements vary by region. Ensure your risk models meet regulatory standards." ], confidence_threshold=0.8 )) return patterns def _generate_healthcare_patterns(self, context: ProjectContext) -> List[DomainPattern]: """Generate healthcare-specific patterns.""" patterns = [] # Patient data privacy pattern patterns.append(DomainPattern( pattern_id="healthcare_privacy", domain=ProjectDomain.HEALTHCARE, technology=None, pattern_name="Patient Data Privacy Protection", pattern_description="Ensure HIPAA compliance and protect sensitive medical information", keywords=["hipaa", "privacy", "patient", "medical", "protected", "phi"], triggers=["patient privacy", "hipaa compliance", "medical data", "phi protection"], responses=[ "HIPAA compliance is mandatory for healthcare applications. Encrypt all PHI data.", "Patient data requires strict access controls and audit logging.", "Consider using healthcare-specific cloud providers with BAA agreements." ], confidence_threshold=0.9 )) return patterns def _generate_gaming_patterns(self, context: ProjectContext) -> List[DomainPattern]: """Generate gaming-specific patterns.""" patterns = [] # Player engagement pattern patterns.append(DomainPattern( pattern_id="gaming_engagement", domain=ProjectDomain.GAMING, technology=None, pattern_name="Player Engagement Optimization", pattern_description="Analyze player behavior to improve retention and engagement", keywords=["engagement", "retention", "player", "behavior", "analytics"], triggers=["player engagement", "retention", "player analytics", "churn"], responses=[ "Player engagement depends on balanced difficulty, rewards, and social features.", "Track session length, frequency, and progression to identify engagement patterns.", "Implement dynamic difficulty adjustment to keep players in the 'flow' state." ], confidence_threshold=0.7 )) return patterns def _generate_web_dev_patterns(self, context: ProjectContext) -> List[DomainPattern]: """Generate web development patterns.""" patterns = [] # Performance optimization pattern patterns.append(DomainPattern( pattern_id="web_performance", domain=ProjectDomain.WEB_DEVELOPMENT, technology=None, pattern_name="Web Performance Optimization", pattern_description="Optimize loading times, bundle sizes, and user experience", keywords=["performance", "optimization", "loading", "bundle", "speed"], triggers=["slow loading", "performance issues", "optimization", "bundle size"], responses=[ "Web performance affects user experience and SEO. Optimize images, minify code, and use CDNs.", "Consider lazy loading, code splitting, and caching strategies for better performance.", "Measure Core Web Vitals: LCP, FID, and CLS for Google's performance standards." ], confidence_threshold=0.7 )) return patterns def _generate_technology_patterns(self, technology: TechnologyStack, context: ProjectContext) -> List[DomainPattern]: """Generate technology-specific patterns.""" patterns = [] if technology == TechnologyStack.REACT: patterns.append(DomainPattern( pattern_id="react_performance", domain=context.domain, technology=technology, pattern_name="React Performance Optimization", pattern_description="Optimize React components for better performance", keywords=["react", "performance", "memo", "usecallback", "render"], triggers=["react performance", "slow renders", "react optimization"], responses=[ "Use React.memo(), useMemo(), and useCallback() to prevent unnecessary re-renders.", "Consider code splitting with React.lazy() and Suspense for better loading.", "The React DevTools Profiler can help identify performance bottlenecks." ], confidence_threshold=0.8 )) elif technology == TechnologyStack.PYTHON_DJANGO: patterns.append(DomainPattern( pattern_id="django_optimization", domain=context.domain, technology=technology, pattern_name="Django Query Optimization", pattern_description="Optimize Django ORM queries and database performance", keywords=["django", "orm", "query", "optimization", "database"], triggers=["django performance", "slow queries", "orm optimization"], responses=[ "Use select_related() and prefetch_related() to reduce database queries.", "Django Debug Toolbar can help identify N+1 query problems.", "Consider database indexing and query optimization for better performance." ], confidence_threshold=0.8 )) return patterns def _generate_business_logic_patterns(self, context: ProjectContext) -> List[DomainPattern]: """Generate patterns based on detected business logic.""" patterns = [] if "authentication" in context.business_logic_patterns: patterns.append(DomainPattern( pattern_id="auth_security", domain=context.domain, technology=None, pattern_name="Authentication Security Best Practices", pattern_description="Implement secure authentication with proper validation", keywords=["authentication", "security", "password", "token", "session"], triggers=["auth issues", "login problems", "security", "authentication"], responses=[ "Use bcrypt or Argon2 for password hashing, never store plaintext passwords.", "Implement rate limiting and account lockout to prevent brute force attacks.", "Consider multi-factor authentication for enhanced security." ], confidence_threshold=0.8 )) if "caching" in context.business_logic_patterns: patterns.append(DomainPattern( pattern_id="caching_strategy", domain=context.domain, technology=None, pattern_name="Caching Strategy Optimization", pattern_description="Implement effective caching to improve performance", keywords=["cache", "caching", "redis", "memcached", "performance"], triggers=["caching", "cache strategy", "performance", "slow responses"], responses=[ "Implement multi-level caching: browser, CDN, application, and database caching.", "Use cache invalidation strategies to ensure data consistency.", "Consider cache warming and preloading for critical data." ], confidence_threshold=0.7 )) return patterns def _log_pattern_generation(self, context: ProjectContext, patterns: List[DomainPattern]): """Log pattern generation activity.""" log_entry = { "timestamp": datetime.datetime.now().isoformat(), "project_domain": context.domain.value, "technologies": [tech.value for tech in context.technologies], "patterns_generated": len(patterns), "pattern_types": [p.pattern_name for p in patterns], "domain_vocabulary_size": len(context.domain_vocabulary), "business_patterns_detected": len(context.business_logic_patterns) } log_file = os.path.join(self.patterns_dir, "pattern_generation.jsonl") try: with open(log_file, 'a') as f: f.write(json.dumps(log_entry) + '\n') except Exception as e: logger.debug(f"Failed to log pattern generation: {e}") def get_patterns_for_context(self, context: ProjectContext) -> List[DomainPattern]: """Get all patterns relevant to the current project context.""" relevant_patterns = [] for pattern in self.generated_patterns: # Match by domain if pattern.domain == context.domain: relevant_patterns.append(pattern) continue # Match by technology if pattern.technology in context.technologies: relevant_patterns.append(pattern) continue # Match by keywords in domain vocabulary if any(keyword in context.domain_vocabulary for keyword in pattern.keywords): relevant_patterns.append(pattern) return relevant_patterns def get_generation_stats(self) -> Dict: """Get statistics about pattern generation.""" stats = { "total_patterns": len(self.generated_patterns), "patterns_by_domain": {}, "patterns_by_technology": {}, "average_confidence": 0.0, "total_usage": sum(p.usage_count for p in self.generated_patterns), "average_success_rate": 0.0 } # Count by domain for pattern in self.generated_patterns: domain = pattern.domain.value stats["patterns_by_domain"][domain] = stats["patterns_by_domain"].get(domain, 0) + 1 # Count by technology for pattern in self.generated_patterns: if pattern.technology: tech = pattern.technology.value stats["patterns_by_technology"][tech] = stats["patterns_by_technology"].get(tech, 0) + 1 # Calculate averages if self.generated_patterns: stats["average_confidence"] = sum(p.confidence_threshold for p in self.generated_patterns) / len(self.generated_patterns) success_rates = [p.success_rate for p in self.generated_patterns if p.usage_count > 0] if success_rates: stats["average_success_rate"] = sum(success_rates) / len(success_rates) return stats def get_domain_pattern_generator(memory_dir: str) -> DomainAwarePatternGenerator: """Get the domain-aware pattern generator instance.""" return DomainAwarePatternGenerator(memory_dir) # Example usage and testing if __name__ == "__main__": import tempfile # Test the domain-aware pattern generator with tempfile.TemporaryDirectory() as temp_dir: generator = DomainAwarePatternGenerator(temp_dir) # Analyze current project context context = generator.analyze_project_context(".") print(f"Detected domain: {context.domain}") print(f"Technologies: {[tech.value for tech in context.technologies]}") print(f"Domain vocabulary: {list(context.domain_vocabulary)[:10]}") # Generate patterns patterns = generator.generate_domain_patterns(context) print(f"Generated {len(patterns)} new patterns") for pattern in patterns: print(f" - {pattern.pattern_name} ({pattern.domain.value})") # Get stats stats = generator.get_generation_stats() print(f"Total patterns: {stats['total_patterns']}") print(f"Average confidence: {stats['average_confidence']:.2f}")