import json import numpy as np from pathlib import Path from sentence_transformers import SentenceTransformer import datetime import traceback from config.debug_config import DEBUG_MODULES import sys from config.paths_config import MEMORY_PATH class MemoryStore: def __init__(self, memory_file="agent_memory.json", model_name='all-MiniLM-L6-v2'): self.memory_file = MEMORY_PATH / memory_file self._log(__name__, f"🔧 MemoryStore initialisiert mit Datei: {self.memory_file}") self.memory_file.parent.mkdir(parents=True, exist_ok=True) # Verzeichnis anlegen self.memory = self.load_memory() # Embedding-Modell laden (für semantische Ähnlichkeit) self._log(__name__, f"📦 Lade Embedding-Modell: {model_name}") try: self.model = SentenceTransformer(model_name) self._log(__name__, f"✅ Modell geladen: {model_name}") except Exception as e: self._log(__name__, f"❌ Fehler beim Laden des Modells: {e}") raise # Embeddings aus vorhandenen Speichern laden/berechnen self._ensure_embeddings() self._log(__name__, f"📊 Memory geladen: {len(self.memory['memories'])} Einträge") def _log(self, module, message): """Debug-Ausgabe mit Timestamp und Modulsteuerung""" if DEBUG_MODULES.get(module, False): timestamp = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3] log_line = f"📝 [{module} {timestamp}] {message}" print(log_line, file=sys.stderr, flush=True) # ← stderr + flush def _log_memory(self, memory, prefix="📌"): """Detaillierte Ausgabe eines Memory-Eintrags""" if DEBUG_MODULES.get(__name__, False): print(f"{prefix} Eintrag:") print(f" Error: {memory.get('error', 'N/A')[:50]}...") print(f" Context: {memory.get('context', 'N/A')[:30]}...") print(f" Confidence: {memory.get('confidence', 0):.2f}") print(f" Success: {memory.get('success_count', 0)}") print(f" Patch: {memory.get('patch_command', 'None')[:30]}...") def load_memory(self): """Lädt Memory aus JSON-Datei""" self._log(__name__, f"📂 Lade Memory von: {self.memory_file}") if self.memory_file.exists(): try: with open(self.memory_file, 'r') as f: data = json.load(f) self._log(__name__, f"✅ Erfolgreich geladen: {len(data.get('memories', []))} Einträge") return data except json.JSONDecodeError as e: self._log(__name__, f"❌ JSON-Fehler: {e}") return {"memories": []} except Exception as e: self._log(__name__, f"❌ Fehler beim Laden: {e}") return {"memories": []} else: self._log(__name__, f"⚠️ Datei nicht gefunden: {self.memory_file}") return {"memories": []} def save_memory(self): """Speichert Memory in JSON-Datei""" self._log(__name__, f"💾 Speichere Memory nach: {self.memory_file}") try: # Backup erstellen if self.memory_file.exists(): backup = self.memory_file.with_suffix('.json.bak') import shutil shutil.copy2(self.memory_file, backup) self._log(__name__, f"📦 Backup erstellt: {backup}") with open(self.memory_file, 'w') as f: json.dump(self.memory, f, indent=2) self._log(__name__, f"✅ Erfolgreich gespeichert: {len(self.memory['memories'])} Einträge") return True except Exception as e: self._log(__name__, f"❌ Fehler beim Speichern: {e}") self._log(__name__, traceback.format_exc()) return False def _ensure_embeddings(self): """Stellt sicher, dass alle Einträge Embeddings haben""" self._log(__name__, "🔄 Prüfe auf fehlende Embeddings...") updated = 0 for idx, memory in enumerate(self.memory["memories"]): if "embedding" not in memory: # Embedding für Fehlermeldung berechnen text = memory.get("error", "") + " " + memory.get("context", "") if text.strip(): self._log(__name__, f"⚙️ Berechne Embedding für Eintrag {idx}: {text[:50]}...") embedding = self.model.encode(text).tolist() memory["embedding"] = embedding updated += 1 else: self._log(__name__, f"⚠️ Eintrag {idx} hat keinen Text für Embedding") if updated > 0: self._log(__name__, f"✅ {updated} neue Embeddings berechnet") self.save_memory() else: self._log(__name__, "✅ Alle Einträge haben bereits Embeddings") def add_memory(self, error, solution, context="", confidence=0.8, patch_command=""): """Fügt neue Fehlerlösung mit Embedding hinzu""" self._log(__name__, f"➕ Füge neuen Memory-Eintrag hinzu:") self._log(__name__, f" Error: {error[:50]}...") self._log(__name__, f" Context: {context[:30]}...") self._log(__name__, f" Solution: {solution[:30]}...") self._log(__name__, f" Confidence: {confidence}") # Embedding berechnen text = error + " " + context self._log(__name__, f"⚙️ Berechne Embedding für {len(text)} Zeichen") embedding = self.model.encode(text).tolist() new_entry = { "error": error[:200], "context": context[:300], "solution": solution, "patch_command": patch_command, "confidence": confidence, "success_count": 1, "embedding": embedding, "created": datetime.datetime.now().isoformat() } self.memory["memories"].append(new_entry) self._log_memory(new_entry, "🆕") if self.save_memory(): self._log(__name__, f"✅ Memory erfolgreich hinzugefügt (ID: {len(self.memory['memories'])-1})") else: self._log(__name__, f"❌ Fehler beim Speichern des neuen Eintrags") # Dabei wird die Übergabe nach Reihenfolge überschrieben! context bleibt default! context enthält vom Parser weitere Infos aus dem Logfile def find_similar(self, query, context="", threshold=0.3, top_k=3): """Findet semantisch ähnliche Fehler mittels Embeddings""" self._log(__name__, f"🔍 Suche ähnliche Einträge für:") self._log(__name__, f" Query: {query[:50]}...") self._log(__name__, f" Context: {context[:30]}...") self._log(__name__, f" Threshold: {threshold}, Top-K: {top_k}") # Existieren Einträge in der JSON-Datei if not self.memory["memories"]: self._log(__name__, "⚠️ Keine Einträge in Memory") return [] # Query-Embedding berechnen query_text = query + " " + context #die reine Fehlermeldung wird mit weiteren Loginfos angereichert! self._log(__name__, f"⚙️ Berechne Query-Embedding") query_embedding = self.model.encode(query_text) matches = [] # enumerate fügt Index jedem Eintrag in der JSON-Datei hinzu! in idx steht somit der Index und in memory das Dictionary for idx, memory in enumerate(self.memory["memories"]): if "embedding" not in memory: #falls zu dem Dictionary kein Embedding existiert weil KI bei Start keins generiert hat self._log(__name__, f"⚠️ Eintrag {idx} hat kein Embedding") continue # Kosinus-Ähnlichkeit berechnen mem_embedding = np.array(memory["embedding"]) similarity = np.dot(query_embedding, mem_embedding) / ( np.linalg.norm(query_embedding) * np.linalg.norm(mem_embedding) ) if similarity >= threshold: self._log(__name__, f"🎯 Treffer {idx}: Similarity={similarity:.3f}, Confidence={memory['confidence']:.2f}") matches.append({ "index": idx, "solution": memory["solution"], "patch_command": memory.get("patch_command", ""), "confidence": similarity * memory["confidence"], "similarity": float(similarity), "error": memory["error"] }) else: if similarity > 0.5: # Zeige knapp daneben self._log(__name__, f"📉 Knapp daneben {idx}: Similarity={similarity:.3f} < {threshold}") # Sortieren nach Ähnlichkeit matches.sort(key=lambda x: x["similarity"], reverse=True) top_matches = matches[:top_k] self._log(__name__, f"📊 Ergebnisse: {len(matches)} Treffer, Top {len(top_matches)} angezeigt") for i, m in enumerate(top_matches): self._log(__name__, f" #{i+1}: Sim={m['similarity']:.3f}, Conf={m['confidence']:.3f}") return top_matches def update_success(self, error, context=""): """Erhöht Erfolgszähler bei bestätigter Lösung""" self._log(__name__, f"📈 Update Success für: {error[:50]}...") text = error + " " + context query_embedding = self.model.encode(text) for idx, memory in enumerate(self.memory["memories"]): if "embedding" in memory: mem_embedding = np.array(memory["embedding"]) similarity = np.dot(query_embedding, mem_embedding) / ( np.linalg.norm(query_embedding) * np.linalg.norm(mem_embedding) ) if similarity > 0.95: # Sehr ähnlich = gleicher Fehler old_count = memory.get("success_count", 0) old_conf = memory.get("confidence", 0.5) memory["success_count"] = old_count + 1 memory["confidence"] = min(1.0, old_conf + 0.1) self._log(__name__, f"✅ Eintrag {idx} aktualisiert: Success {old_count}→{memory['success_count']}, Confidence {old_conf:.2f}→{memory['confidence']:.2f}") if self.save_memory(): return True else: self._log(__name__, f"❌ Fehler beim Speichern nach Update") return False self._log(__name__, f"⚠️ Kein passender Eintrag für Update gefunden") return False def execute_patch(self, patch_command): """ Führt einen gespeicherten Patch-Befehl aus. Muss in core.py integriert werden. """ self._log(__name__, f"⚡ Execute Patch: {patch_command}") if not patch_command: self._log(__name__, "❌ Kein Patch-Befehl vorhanden") return "❌ Kein Patch-Befehl vorhanden" # FIXME: Hier wird der Patch-Befehl an den Patcher übergeben self._log(__name__, "⚠️ execute_patch muss noch in core.py integriert werden") return f"✅ Patch ausgeführt: {patch_command}" def get_stats(self): """Gibt Statistiken über Memory aus""" stats = { "total_entries": len(self.memory["memories"]), "avg_confidence": 0, "total_success": 0, "with_patch": 0, "with_context": 0 } for m in self.memory["memories"]: stats["avg_confidence"] += m.get("confidence", 0) stats["total_success"] += m.get("success_count", 0) if m.get("patch_command"): stats["with_patch"] += 1 if m.get("context"): stats["with_context"] += 1 if stats["total_entries"] > 0: stats["avg_confidence"] /= stats["total_entries"] self._log(__name__, f"📊 Memory Stats: {stats}") return stats