Spaces:
Build error
Build error
File size: 8,458 Bytes
0bdeed7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 | 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
}
|