""" OpenClaw Bridge - Verbindung zwischen OpenClaw-System und Second Brain. Ermöglicht: 1. Automatisches Speichern von Session-Inhalten als Engramme 2. Proaktive Erinnerungen (Heartbeat) 3. Selbstheilende Fehlerbehandlung 4. Feedback-Integration (richtig/falsch) """ import os import sys import json import traceback from pathlib import Path from datetime import datetime, timezone from typing import Optional, List, Dict, Any # Ensure project root is on sys.path for standalone usage project_root = str(Path(__file__).parent.parent) if project_root not in sys.path: sys.path.insert(0, project_root) # Activate virtualenv if available (for chromadb etc.) venv_path = Path(__file__).parent.parent / ".venv" if venv_path.exists(): venv_site_packages = list((venv_path / "lib").glob("python3.*/site-packages")) if venv_site_packages and str(venv_site_packages[0]) not in sys.path: sys.path.insert(0, str(venv_site_packages[0])) # Second Brain Import from src.engram import Engram, Grounding from src.store import EngramStore # Retriever: optional (braucht chromadb) try: from src.retriever import Retriever except ImportError: Retriever = None # Chroma: optional (braucht chromadb) try: from src.chroma_store import ChromaStore except Exception: ChromaStore = None # --- Konfiguration --- BRAIN_DB = Path(__file__).parent.parent / "data" / "brain.sqlite" def get_brain() -> EngramStore: """Gibt initialisierten Brain-Store.""" BRAIN_DB.parent.mkdir(parents=True, exist_ok=True) return EngramStore(str(BRAIN_DB)) # --- 1. Auto-Save: Session-Inhalt als Engramm --- def save_session_learned( content: str, source: str = "agent", tags: Optional[List[str]] = None, session_id: Optional[str] = None, confidence: float = 0.5, grounding: Grounding = Grounding.ASSUMPTION, ) -> Engram: """ Speichert eine gelernte Information aus einer OpenClaw-Session. Nutzung im Agent-Code: from second_brain.src.openclaw_bridge import save_session_learned eg = save_session_learned( "Neues Faktum", tags=["project", "wichtig"], confidence=0.7, ) """ store = get_brain() eg = Engram.create( content=content, source=source, tags=tags or [], session_id=session_id, confidence=confidence, grounding=grounding, ) store.save(eg) return eg # --- 2. Proaktivität: Was soll ich tun? --- def heartbeat_check() -> Optional[str]: """ Prüft ob proaktive Aktion sinnvoll ist. Rückgabe: Nachricht für den User, oder None wenn nichts zu tun. """ store = get_brain() # Prüfe auf wichtige unbestätigte Engramme egs = store.get_all(limit=50) unconfirmed = [ eg for eg in egs if not eg.correctness.confirmed and eg.compute_confidence() > 0.5 ][:5] if unconfirmed: contents = "\n".join([f" - {eg.content[:80]}" for eg in unconfirmed]) return ( f"🧠 Second Brain Heartbeat\n" f"Unbestätigte Engramme mit gutem Confidence-Score:\n{contents}\n" f"Bestätige mit: `python -m src.cli confirm `" ) # Prüfe auf Fehler-Muster errors = store.search_tag("error", limit=5) if len(errors) >= 3: return ( f"⚠️ Second Brain: {len(errors)} Fehler-Engramme gespeichert.\n" f"Möglicherweise wiederholendes Problem. Prüfe mit `python -m src.cli search --tag error`" ) return None # --- 3. Fehlerbehandlung --- def handle_tool_error( tool_name: str, error_message: str, context: Optional[Dict[str, Any]] = None, ) -> Engram: """ Speichert einen Fehler als Engramm für spätere Analyse. Nutzung bei try/except um Tool-Calls: try: result = some_tool_call() except Exception as e: handle_tool_error("tool_name", str(e), context={"args": args}) """ store = get_brain() content = f"FEHLER: {tool_name}\n{error_message}\n" if context: content += f"\nContext: {json.dumps(context, default=str, ensure_ascii=False)}" eg = Engram.create( content=content, source="system", tags=["error", tool_name], confidence=0.0, grounding=Grounding.UNKNOWN, ) store.save(eg) # Prüfe auf Wiederholung similar = store.search_text(error_message[:50], limit=10) same_errors = [e for e in similar if e.metadata.get("source") == "system" and "error" in e.metadata.get("tags", [])] if len(same_errors) >= 3: # Auto-Fix: Erstelle Engramm mit Lösungsstrategie fix_eg = Engram.create( content=f"AUTO-FIX-STRATEGIE für wiederholten Fehler '{tool_name}':\n" f"1. Retry mit alternativem Ansatz\n" f"2. Fallback auf simpleres Tool\n" f"3. User benachrichtigen wenn Persistenz > 5x", source="system", tags=["auto-fix", tool_name, "strategy"], confidence=0.6, grounding=Grounding.INFERRED, ) store.save(fix_eg) return eg # --- 4. Feedback-Integration --- def user_feedback(engram_id: str, is_correct: bool, note: str = "") -> Engram: """ Nimmt User-Feedback (richtig/falsch) auf. Nutzung nach jeder Agent-Antwort: if user_says_correct: user_feedback(engram_id, True) elif user_says_wrong: user_feedback(engram_id, False, note="Falsches Datum") """ store = get_brain() eg = store.get(engram_id) if not eg: raise ValueError(f"Engram {engram_id} nicht gefunden") if is_correct: eg.correctness.confirm(by="user", note=note) else: eg.correctness.reject(by="user", note=note) store.save(eg) return eg # --- 5. Kontext-anreicherung für Agenten --- def enrich_context(topic: str, limit: int = 3) -> str: """ Holt relevante Engramme und formatiert sie als Kontext für den Agent. Nutzung vor Prompt-Generierung: memory_context = enrich_context("Projekt X") # memory_context in das Prompt einbauen """ store = get_brain() # Versuche Hybrid-Retrieval (FTS + optional Vector), fallback auf Textsuche if Retriever: chroma = None if ChromaStore: try: chroma = ChromaStore(path=str(Path(__file__).parent.parent / "data" / "chroma")) except Exception: chroma = None ret = Retriever(store, chroma=chroma) try: results = ret.hybrid_retrieve(topic, limit=limit * 3, min_confidence=0.3) except Exception: results = ret.retrieve(topic, limit=limit * 3, min_confidence=0.3) # confirmed-first ranking def _rank(r): eg = r["engram"] confirmed = 1 if getattr(eg.correctness, "confirmed", False) else 0 return (confirmed, float(r.get("score", 0.0))) results.sort(key=_rank, reverse=True) # If we have confirmed results, show only confirmed up to limit confirmed_only = [r for r in results if r["engram"].correctness.confirmed] if confirmed_only: results = confirmed_only[:limit] else: results = results[:limit] else: results_raw = store.search_text(topic, limit=limit) results = [{"engram": eg, "score": 0.5} for eg in results_raw] if not results: return "" lines = ["\n📚 Second Brain Kontext:"] for r in results: eg = r["engram"] conf = eg.compute_confidence() marker = "✅" if conf > 0.7 else "⚠️" if conf > 0.4 else "❌" g = Grounding(eg.metadata.get("grounding", 0)).name lines.append(f" {marker} [{g}] {eg.content[:150]}") return "\n".join(lines) # --- 6. Backup & Wartung --- def backup_daily() -> str: """ Tägliches Backup als JSONL. Sollte via Cron aufgerufen werden: 0 2 * * * python -m second_brain.src.openclaw_bridge backup """ store = get_brain() ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") path = BRAIN_DB.parent / f"backup_{ts}.jsonl" count = store.export_jsonl(str(path)) return f"Backup: {count} Engramme -> {path}" def stats() -> Dict[str, Any]: """Gibt Statistiken zurück.""" store = get_brain() return Retriever(store).stats() # --- CLI-Entrypoint --- def main(): """CLI für die Bridge-Funktionen.""" import argparse parser = argparse.ArgumentParser(description="Second Brain OpenClaw Bridge") sub = parser.add_subparsers(dest="cmd") p_save = sub.add_parser("save", help="Speichere Lerngut") p_save.add_argument("content") p_save.add_argument("--tag", action="append", default=[]) p_save.add_argument("--confidence", type=float, default=0.5) p_feedback = sub.add_parser("feedback", help="User-Feedback") p_feedback.add_argument("id") p_feedback.add_argument("--correct", action="store_true") p_feedback.add_argument("--note", default="") p_heartbeat = sub.add_parser("heartbeat", help="Heartbeat-Check") p_backup = sub.add_parser("backup", help="Backup erstellen") p_stats = sub.add_parser("stats", help="Statistiken") p_context = sub.add_parser("context", help="Kontext holen") p_context.add_argument("topic") p_context.add_argument("--limit", type=int, default=3) args = parser.parse_args() if args.cmd == "save": eg = save_session_learned(args.content, tags=args.tag, confidence=args.confidence) print(f"Saved: {eg.id} (conf: {eg.compute_confidence():.2f})") elif args.cmd == "feedback": eg = user_feedback(args.id, args.correct, args.note) print(f"Feedback recorded: {eg.id} -> {eg.compute_confidence():.2f}") elif args.cmd == "heartbeat": msg = heartbeat_check() print(msg or "HEARTBEAT_OK") elif args.cmd == "backup": print(backup_daily()) elif args.cmd == "stats": print(json.dumps(stats(), indent=2)) elif args.cmd == "context": print(enrich_context(args.topic, args.limit)) else: parser.print_help() if __name__ == "__main__": main()