import hashlib import json import os from typing import Dict, List, Optional, Any from datetime import datetime import logging logger = logging.getLogger(__name__) class ErrorPatternLearner: """ Tracks error patterns and solutions to build a knowledge base. Stores anonymized error signatures and their solutions. Upgraded to support Semantic Retrieval via embeddings. """ def __init__(self, storage_dir: str = "./error_patterns", embeddings: Any = None, chroma_client: Any = None): self.storage_dir = storage_dir os.makedirs(storage_dir, exist_ok=True) self.patterns_file = os.path.join(storage_dir, "patterns.json") self.patterns = self._load_patterns() self.embeddings = embeddings self.chroma_client = chroma_client self.collection = None if chroma_client and embeddings: try: self.collection = chroma_client.get_or_create_collection( name="learned_error_patterns", metadata={"hnsw:space": "cosine"} ) logger.info("Initialized semantic collection for ErrorPatternLearner") except Exception as e: logger.error(f"Failed to initialize Chroma collection for patterns: {e}") def _load_patterns(self) -> Dict: """Load existing patterns from storage""" if os.path.exists(self.patterns_file): try: with open(self.patterns_file, 'r') as f: return json.load(f) except Exception as e: logger.error(f"Failed to load patterns: {e}") return {} return {} def _save_patterns(self): """Save patterns to storage""" try: with open(self.patterns_file, 'w') as f: json.dump(self.patterns, f, indent=2) except Exception as e: logger.error(f"Failed to save patterns: {e}") def _create_signature(self, error_text: str, language: str = "unknown") -> str: """ Create an anonymized signature for an error. Removes specific values but keeps structure. """ # Normalize error text normalized = error_text.lower() # Remove specific values (numbers, URLs, paths) import re normalized = re.sub(r'\d+', 'N', normalized) # Numbers normalized = re.sub(r'https?://\S+', 'URL', normalized) # URLs normalized = re.sub(r'/[\w/]+', '/PATH', normalized) # File paths normalized = re.sub(r'[a-f0-9]{8,}', 'HASH', normalized) # Hashes # Create hash of normalized error signature = hashlib.sha256(f"{language}:{normalized}".encode()).hexdigest()[:16] return signature def record_solution( self, error_text: str, solution: str, root_cause: str = "", language: str = "unknown", confidence: str = "medium", metadata: Optional[Dict] = None ): """ Record a successful solution for an error pattern. """ signature = self._create_signature(error_text, language) if signature not in self.patterns: self.patterns[signature] = { 'signature': signature, 'language': language, 'first_seen': datetime.now().isoformat(), 'occurrences': 0, 'solutions': [] } # Add to semantic index if available if self.collection and self.embeddings: try: import asyncio vector = self.embeddings.embed_query(error_text) self.collection.add( embeddings=[vector], documents=[error_text], ids=[signature], metadatas=[{"language": language}] ) except Exception as e: logger.warning(f"Failed to add pattern {signature} to semantic index: {e}") # Increment occurrence count self.patterns[signature]['occurrences'] += 1 self.patterns[signature]['last_seen'] = datetime.now().isoformat() # Add solution if not duplicate solution_hash = hashlib.sha256(f"{root_cause}:{solution}".encode()).hexdigest()[:8] existing_solutions = [s['hash'] for s in self.patterns[signature]['solutions']] if solution_hash not in existing_solutions: self.patterns[signature]['solutions'].append({ 'hash': solution_hash, 'summary': solution[:200], # Keep legacy summary for UI compatibility 'full_fix': solution, 'root_cause': root_cause, 'confidence': confidence, 'added': datetime.now().isoformat(), 'metadata': metadata or {} }) logger.info(f"📚 Recorded new solution for pattern {signature}") self._save_patterns() def find_similar_patterns( self, error_text: str, language: str = "unknown", top_k: int = 3 ) -> List[Dict]: """ Find similar error patterns with known solutions. Combines exact signature matching with semantic search. """ results = [] signature = self._create_signature(error_text, language) # 1. Check for Exact Match if signature in self.patterns: pattern = self.patterns[signature] results.append({ 'match_type': 'exact', 'pattern': pattern, 'similarity': 1.0 }) # 2. Check for Semantic Match (if enough time/resources) if self.collection and self.embeddings and len(results) < top_k: try: query_vector = self.embeddings.embed_query(error_text) semantic_results = self.collection.query( query_embeddings=[query_vector], n_results=top_k, where={"language": language} ) if semantic_results and semantic_results['ids'][0]: for i, found_id in enumerate(semantic_results['ids'][0]): if found_id == signature: continue # Already added as exact distance = semantic_results['distances'][0][i] if 'distances' in semantic_results else 0.5 similarity = 1 - distance if similarity > 0.6 and found_id in self.patterns: results.append({ 'match_type': 'semantic', 'pattern': self.patterns[found_id], 'similarity': similarity }) except Exception as e: logger.warning(f"Semantic pattern search failed: {e}") # 3. Fallback to Fuzzy Language Match (Legacy) if not results: for sig, pattern in self.patterns.items(): if pattern['language'] == language and pattern['occurrences'] >= 2: results.append({ 'match_type': 'fuzzy', 'pattern': pattern, 'similarity': 0.5 }) # Sort by similarity and return top-K results.sort(key=lambda x: x['similarity'], reverse=True) return results[:top_k] def get_stats(self) -> Dict: """Get statistics about learned patterns""" total_patterns = len(self.patterns) total_solutions = sum(len(p['solutions']) for p in self.patterns.values()) total_occurrences = sum(p['occurrences'] for p in self.patterns.values()) languages = {} for pattern in self.patterns.values(): lang = pattern['language'] languages[lang] = languages.get(lang, 0) + 1 return { 'total_patterns': total_patterns, 'total_solutions': total_solutions, 'total_occurrences': total_occurrences, 'languages': languages }