Otto/second-brain
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(integration): OpenClaw Bridge + Dashboard + Proaktivität

Browse Source 
- src/openclaw_bridge.py: Session-Save, Heartbeat, Fehlerbehandlung, Feedback
- src/dashboard.py: HTML-Dashboard (keine externen Abhängigkeiten)

Issues: #4, #5, #6
This commit is contained in:
Otto
2026-05-25 01:01:12 +02:00
parent 5e4f21e680
commit cf05dba944
2 changed files with 484 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
src/dashboard.py Normal file
View File

@@ -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"""
<tr>
<td><span style="color:{color(conf)};font-size:1.2em">{marker(conf)}</span></td>
<td><code>{str(eg.id)[:8]}</code></td>
<td>{eg.content[:100]}{'...' if len(eg.content) > 100 else ''}</td>
<td><span class="badge">{grounding_name(eg.metadata.get('grounding', 0))}</span></td>
<td><span class="badge">{eg.metadata.get('source', '?')}</span></td>
<td>{conf:.2f}</td>
<td>{eg.correctness.confirmations}/{eg.correctness.rejections}</td>
<td>{eg.metadata.get('access_count', 0)}</td>
<td>{', '.join(eg.metadata.get('tags', [])[:3])}</td>
</tr>
""")
html = f"""<!DOCTYPE html>
<html lang="de">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>🧠 Second Brain Dashboard</title>
<style>
:root {{ --bg: #1a1a2e; --card: #16213e; --text: #eee; --accent: #0f4c75; }}
body {{ font-family: -apple-system, BlinkMacSystemFont, sans-serif; background: var(--bg); color: var(--text); margin: 0; padding: 20px; }}
h1 {{ margin: 0 0 10px; font-size: 1.8em; }}
.stats {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 15px; margin-bottom: 25px; }}
.stat-card {{ background: var(--card); border-radius: 12px; padding: 15px; text-align: center; }}
.stat-card .num {{ font-size: 2em; font-weight: bold; color: #4fc3f7; }}
.stat-card .lbl {{ font-size: 0.85em; opacity: 0.7; }}
table {{ width: 100%; border-collapse: collapse; background: var(--card); border-radius: 12px; overflow: hidden; }}
th {{ text-align: left; padding: 10px; background: var(--accent); font-size: 0.85em; text-transform: uppercase; }}
td {{ padding: 8px 10px; border-bottom: 1px solid rgba(255,255,255,0.05); font-size: 0.9em; vertical-align: top; }}
tr:hover td {{ background: rgba(255,255,255,0.03); }}
.badge {{ display: inline-block; padding: 2px 8px; border-radius: 12px; background: rgba(79,195,247,0.2); font-size: 0.75em; }}
.search {{ margin-bottom: 20px; }}
.search input {{ width: 100%; max-width: 400px; padding: 10px 15px; border-radius: 8px; border: 1px solid rgba(255,255,255,0.1); background: var(--card); color: var(--text); font-size: 1em; }}
.refresh {{ position: fixed; top: 20px; right: 20px; background: #2ecc71; color: #000; border: none; padding: 10px 20px; border-radius: 8px; cursor: pointer; font-weight: bold; }}
.footer {{ margin-top: 30px; text-align: center; opacity: 0.5; font-size: 0.8em; }}
</style>
</head>
<body>
<h1>🧠 Second Brain Dashboard</h1>
<button class="refresh" onclick="location.reload()">🔄 Aktualisieren</button>
<div class="stats">
<div class="stat-card"><div class="num">{stats['total_engrams']}</div><div class="lbl">Engramme</div></div>
<div class="stat-card"><div class="num">{stats['confirmed']}</div><div class="lbl">Bestätigt</div></div>
<div class="stat-card"><div class="num">{stats['unconfirmed']}</div><div class="lbl">Unbestätigt</div></div>
<div class="stat-card"><div class="num">{len(stats.get('sources', dict()))}</div><div class="lbl">Quellen</div></div>
<div class="stat-card"><div class="num">{stats['db_size_bytes']/1024:.0f}</div><div class="lbl">KB Größe</div></div>
</div>
<div class="search">
<input type="text" placeholder="🔍 Suche nach Engrammen..." id="searchInput" onkeyup="filterTable()">
</div>
<table id="engramTable">
<thead>
<tr>
<th>Status</th>
<th>ID</th>
<th>Inhalt</th>
<th>Grounding</th>
<th>Quelle</th>
<th>Confidence</th>
<th>Feedback</th>
<th>Zugriffe</th>
<th>Tags</th>
</tr>
</thead>
<tbody>
{''.join(rows)}
</tbody>
</table>
<div class="footer">Generated {__import__('datetime').datetime.now().isoformat()}</div>
<script>
function filterTable() {{
var input = document.getElementById('searchInput');
var filter = input.value.toLowerCase();
var table = document.getElementById('engramTable');
var rows = table.getElementsByTagName('tr');
for (var i=1; i<rows.length; i++) {{
var txt = rows[i].textContent || rows[i].innerText;
rows[i].style.display = txt.toLowerCase().indexOf(filter) > -1 ? '' : 'none';
}}
}}
</script>
</body>
</html>
"""
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()

292
src/openclaw_bridge.py Normal file
View File

@@ -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 <id>`"
)
# 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()