Spaces:
Build error
Build error
| 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 | |
| } | |