diff --git a/src/dashboard.py b/src/dashboard.py new file mode 100644 index 0000000..a380d0c --- /dev/null +++ b/src/dashboard.py @@ -0,0 +1,192 @@ +#!/usr/bin/env python3 +""" +Second Brain Dashboard - Selbstgenerierendes HTML. +Keine Web-Frameworks nötig, einfach Python + Browser. + +Usage: + python -m src.dashboard # Generiert und öffnet dashboard + python -m src.dashboard --serve # Startet lokalen HTTP-Server +""" + +import json +import argparse +import http.server +import socketserver +import webbrowser +from pathlib import Path +from uuid import UUID + +from .store import EngramStore +from .engram import Engram, Grounding +from .retriever import Retriever + +DB_PATH = Path(__file__).parent.parent / "data" / "brain.sqlite" +HTML_PATH = Path(__file__).parent.parent / "data" / "dashboard.html" + + +def generate_dashboard() -> str: + """Generiert HTML-Dashboard aus aktuellem Brain-Stand.""" + store = EngramStore(str(DB_PATH)) + ret = Retriever(store) + stats = ret.stats() + egs = store.get_all(limit=100) + + # Farbe nach Confidence + def color(conf): + if conf > 0.7: return "#2ecc71" + if conf > 0.4: return "#f1c40f" + return "#e74c3c" + + def marker(conf): + if conf > 0.7: return "✅" + if conf > 0.4: return "⚠️" + return "❌" + + def grounding_name(val): + try: + return Grounding(int(val)).name + except: + return "UNKNOWN" + + # Liste der Engramme + rows = [] + for eg in egs: + conf = eg.compute_confidence() + rows.append(f""" + + {marker(conf)} + {str(eg.id)[:8]} + {eg.content[:100]}{'...' if len(eg.content) > 100 else ''} + {grounding_name(eg.metadata.get('grounding', 0))} + {eg.metadata.get('source', '?')} + {conf:.2f} + {eg.correctness.confirmations}/{eg.correctness.rejections} + {eg.metadata.get('access_count', 0)} + {', '.join(eg.metadata.get('tags', [])[:3])} + + """) + + html = f""" + + + + +🧠 Second Brain Dashboard + + + +

🧠 Second Brain Dashboard

+ + +
+
{stats['total_engrams']}
Engramme
+
{stats['confirmed']}
Bestätigt
+
{stats['unconfirmed']}
Unbestätigt
+
{len(stats.get('sources', dict()))}
Quellen
+
{stats['db_size_bytes']/1024:.0f}
KB Größe
+
+ +
+ +
+ + + + + + + + + + + + + + + + + {''.join(rows)} + +
StatusIDInhaltGroundingQuelleConfidenceFeedbackZugriffeTags
+ +
Generated {__import__('datetime').datetime.now().isoformat()}
+ + + + +""" + return html + + +def save_dashboard(path: str = None): + """Speichert Dashboard als HTML.""" + path = path or str(HTML_PATH) + html = generate_dashboard() + with open(path, "w", encoding="utf-8") as f: + f.write(html) + print(f"Dashboard saved: {path}") + return path + + +def serve_dashboard(port: int = 8050): + """Startet lokalen HTTP-Server.""" + save_dashboard() + + class Handler(http.server.SimpleHTTPRequestHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, directory=str(HTML_PATH.parent), **kwargs) + + def log_message(self, format, *args): + pass # Stilles Logging + + print(f"Serving at http://localhost:{port}/dashboard.html") + print("Press Ctrl+C to stop") + with socketserver.TCPServer(("", port), Handler) as httpd: + webbrowser.open(f"http://localhost:{port}/dashboard.html") + httpd.serve_forever() + + +def main(): + parser = argparse.ArgumentParser(description="Second Brain Dashboard") + parser.add_argument("--serve", action="store_true", help="Start HTTP server") + parser.add_argument("--port", type=int, default=8050) + parser.add_argument("--output", default=str(HTML_PATH)) + args = parser.parse_args() + + if args.serve: + serve_dashboard(args.port) + else: + path = save_dashboard(args.output) + webbrowser.open(f"file://{path}") + + +if __name__ == "__main__": + main() diff --git a/src/openclaw_bridge.py b/src/openclaw_bridge.py new file mode 100644 index 0000000..8e3aa93 --- /dev/null +++ b/src/openclaw_bridge.py @@ -0,0 +1,292 @@ +""" +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 + +# Second Brain Import +from .store import EngramStore +from .engram import Engram, Grounding +from .retriever import Retriever + + +# --- 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() + ret = Retriever(store) + + # A: Unbestätigte Engramme die seit längerem nicht geprüft wurden + # B: Hohe-Prioritäts-Themen (tags wie "wichtig", "dringend") + # C: Fehler-Engramme die repeating sind + + # 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: + ids = ", ".join([str(eg.id)[:8] for eg in unconfirmed]) + contents = "\n".join([f" - {eg.content[:80]}" for eg in unconfimed]) + 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() + ret = Retriever(store) + results = ret.retrieve(topic, limit=limit, min_confidence=0.3) + 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()