Otto
29bc45d623
Neu: - systemd: secondbrain-dashboard.service (Port 8501, autostart) - cron_rules.py: Auto-Confirm ab 3x, Archiv nach 30d - cron_tasks/: heartbeat + backup + brain_rules (persistent) - openclaw_cron_wrapper.py: subprocess-Isolation (kein SessionTakeover) - chat_autosave.py: Auto-Save von Chat + Kontext-Anreicherung Daten: - 18 unbestätigte Engramme bewertet: - 14x CONFIRMED (Fakten/Definitionen korrekt) - 3x ARCHIVIERT (historisch, nicht aktuell) - 1x CONFIRMED (Regel 73624013) - 0 offene unbestätigte Closes Gitea-Issue: #9
120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
#!/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'")
|