#!/usr/bin/env python3 """ Chat-Auto-Save: Wertvolle User-Nachrichten → Engramm. Wird am Ende jeder Main-Session-Antwort aufgerufen. """ import sys import json import hashlib from pathlib import Path BRAIN_DIR = Path("/root/.openclaw/workspace/second-brain") sys.path.insert(0, str(BRAIN_DIR)) from src.engram import Engram, Grounding from src.store import EngramStore DB_PATH = BRAIN_DIR / "data" / "brain.sqlite" def _hash(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12] def is_fluff(content: str) -> bool: """Prüft ob Inhalt nur Floskel ist.""" lower = content.lower().strip().rstrip(".?!") short_fluff = [ "hallo", "hi", "hey", "guten tag", "guten morgen", "guten abend", "danke", "ok", "okay", "ja", "nein", "bitte", "gerne", "tschüss", "bis später", "bis morgen", "alles klar", "in ordnung", ] if lower in short_fluff: return True if len(content) < 10 and all(c in " ?,!.;:-" for c in content): return True return False def save_if_worthy(content: str, source: str = "user", tags: list = None, confidence: float = 0.7, session_id: str = None, reasoning: str = None) -> dict: """ Speichert Nachricht als Engramm wenn sie Wert hat. Wird in jeder Antwort aufgerufen. """ if is_fluff(content): return {"saved": False, "reason": "fluff"} store = EngramStore(str(DB_PATH)) content_hash = _hash(content) recent = store.get_all(limit=200) for eg in recent: if _hash(eg.content) == content_hash: return {"saved": False, "reason": "duplicate", "id": str(eg.id)} eg = Engram.create( content=content, source=source, tags=tags or ["auto-save", "chat"], session_id=session_id, confidence=confidence, grounding=Grounding.ASSUMPTION, ) store.save(eg) return { "saved": True, "id": str(eg.id), "confidence": eg.compute_confidence(), "first8": str(eg.id)[:8], } def enrich_prompt(topic: str, limit: int = 3) -> str: """ Holt relevante bestätigte Engramme für Kontext-Anreicherung. Wird VOR jeder Antwort aufgerufen. """ store = EngramStore(str(DB_PATH)) recent = store.get_all(limit=100) # Einfache Text-Suche (kein FTS wegen Satzzeichen) topic_lower = topic.lower() matches = [] for eg in recent: if eg.correctness.confirmed and topic_lower in eg.content.lower(): matches.append(eg) elif len(matches) < limit and any(t in topic_lower for t in [t.lower() for t in eg.metadata.get("tags", [])]): matches.append(eg) if len(matches) >= limit: break if not matches: return "" lines = ["\n📚 Relevantes Wissen:"] for eg in matches[:limit]: lines.append(f" • [{eg.compute_confidence():.0%}] {eg.content[:120]}") return "\n".join(lines) def check_pending(session_id: str = None) -> list: """Gibt unbestätigte Engramme zurück.""" store = EngramStore(str(DB_PATH)) egs = store.get_all(limit=50) pending = [eg for eg in egs if not eg.correctness.confirmed] return pending if __name__ == "__main__": import sys if len(sys.argv) > 1: result = save_if_worthy(sys.argv[1]) print(json.dumps(result, indent=2)) else: print("Usage: python3 chat_autosave.py 'Nachricht'")