341 lines
10 KiB
Python
341 lines
10 KiB
Python
"""
|
|
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 <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()
|
|
|
|
# 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()
|