dbf2497cd6
- openmemory_client.py: REST-API Client für search/add/list - tools/openmemory.py: openmemory_add, openmemory_search, Kumpel SYSTEM_PROMPT_EXTRA - llm.py: OpenMemory-Block in Prompt (Text + Bild) - homelab.conf: OPENMEMORY Optionen dokumentiert
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
"""OpenMemory REST-Client (CT 122).
|
|
|
|
Nutzt die REST-API von OpenMemory (mem0/Qdrant) für langfristige User-Memories.
|
|
Ergänzt den Memory-Service (CT 117) um semantische/langfristige Erinnerungen.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import requests
|
|
|
|
from core import config
|
|
|
|
log = logging.getLogger("openmemory_client")
|
|
|
|
_cfg = None
|
|
_base_url = None
|
|
_user_id = None
|
|
|
|
|
|
def _ensure_config():
|
|
global _cfg, _base_url, _user_id
|
|
if _base_url is not None:
|
|
return
|
|
_cfg = config.parse_config()
|
|
_base_url = (_cfg.raw.get("OPENMEMORY_API_URL") or "http://10.10.10.122:8765").rstrip("/")
|
|
_user_id = _cfg.raw.get("OPENMEMORY_USER_ID") or "orbitalo"
|
|
if not _base_url:
|
|
log.warning("OPENMEMORY_API_URL nicht in homelab.conf")
|
|
|
|
|
|
def search(query: str, limit: int = 10) -> list[dict]:
|
|
"""Sucht Memories per Text (GET mit search_query). Liefert Liste von {content, id, ...}."""
|
|
_ensure_config()
|
|
if not _base_url:
|
|
return []
|
|
try:
|
|
r = requests.get(
|
|
f"{_base_url}/api/v1/memories/",
|
|
params={"user_id": _user_id, "search_query": query, "size": limit},
|
|
timeout=5,
|
|
)
|
|
if r.ok:
|
|
data = r.json()
|
|
items = data.get("items", [])
|
|
return [{"content": i.get("content", ""), "id": str(i.get("id", ""))} for i in items]
|
|
log.debug("OpenMemory search %s: %s", r.status_code, r.text[:150])
|
|
except Exception as e:
|
|
log.debug("OpenMemory search: %s", e)
|
|
return []
|
|
|
|
|
|
def add(text: str, app: str = "hausmeister") -> Optional[dict]:
|
|
"""Fügt eine neue Memory hinzu. Gibt {id, ...} oder None zurück."""
|
|
_ensure_config()
|
|
if not _base_url or not text.strip():
|
|
return None
|
|
try:
|
|
r = requests.post(
|
|
f"{_base_url}/api/v1/memories/",
|
|
json={"user_id": _user_id, "text": text.strip(), "metadata": {}, "infer": True, "app": app},
|
|
timeout=10,
|
|
)
|
|
if r.ok:
|
|
return r.json()
|
|
log.warning("OpenMemory add %s: %s", r.status_code, r.text[:200])
|
|
except Exception as e:
|
|
log.warning("OpenMemory add: %s", e)
|
|
return None
|
|
|
|
|
|
def list_memories(limit: int = 15) -> list[dict]:
|
|
"""Listet die neuesten Memories (ohne semantische Suche)."""
|
|
_ensure_config()
|
|
if not _base_url:
|
|
return []
|
|
try:
|
|
r = requests.get(
|
|
f"{_base_url}/api/v1/memories/",
|
|
params={"user_id": _user_id, "page": 1, "size": limit},
|
|
timeout=5,
|
|
)
|
|
if r.ok:
|
|
data = r.json()
|
|
items = data.get("items", [])
|
|
return [{"content": i.get("content", ""), "id": str(i.get("id", ""))} for i in items]
|
|
except Exception as e:
|
|
log.debug("OpenMemory list: %s", e)
|
|
return []
|
|
|
|
|
|
def get_openmemory_for_prompt(query: str, top_k: int = 8) -> str:
|
|
"""Holt OpenMemory-Ergebnisse und formatiert sie für den System-Prompt."""
|
|
items = search(query, limit=top_k)
|
|
if not items:
|
|
items = list_memories(limit=5)
|
|
if not items:
|
|
return ""
|
|
lines = ["", "=== OPENMEMORY (langfristige Erinnerungen) ==="]
|
|
for item in items:
|
|
c = (item.get("content") or "").strip()
|
|
if c:
|
|
lines.append(f"• {c}")
|
|
lines.append("=== ENDE OPENMEMORY ===")
|
|
return "\n".join(lines)
|