#!/usr/bin/env python3 """mem0-Konsolidierung (monatlich, 19.07.2026 — Nachtrag Phase 2 Drei-Welten-Plan). mem0 OSS v2 ist ADD-only: Es legt Fakten an, loest aber Widersprueche nie auf und legt Aehnliches nie zusammen — ohne Pflege waechst das Gedaechtnis unbegrenzt und der Recall verrauscht. Der Steward-Dedupe faengt nur exakte Dubletten. Dieser Lauf macht die fehlende Konsolidierung: 1. Cluster aehnlicher Fakten holen — ueber den bestehenden /graph-Endpunkt des Sidecars (Kosinus der gespeicherten Embeddings, kein Re-Embedding, kein Chroma-Zugriff noetig). 2. Je Cluster entscheidet das lokale Hirn (Alias `fast`, Thinking aus wie im Sidecar): zusammenlegen / Widerspruch aufloesen (neuester Zeitstempel gewinnt) / NICHTS tun. 3. Anwenden ueber die normalen Sidecar-Endpunkte (PUT/DELETE /memory/). Sicherheitsleitplanken (hart im Code, nicht im Prompt): - Pro Cluster ueberlebt IMMER mindestens ein Fakt. - Nur ids aus dem Cluster, max. 1 Update pro Cluster, Inhalt <= 600 Zeichen. - LLM-Fehler / kaputtes JSON => Cluster bleibt unangetastet. - Kappung pro Lauf (MAX_CLUSTERS), damit ein Lauf nie das halbe Gedaechtnis umpfluegt — der naechste Monat macht weiter. Aufruf: mem0-konsolidierung.py [--apply] (Default: Dry-Run, druckt nur Plan) """ import json import sys import time import urllib.request MEM0 = "http://127.0.0.1:8765" LLM = "http://127.0.0.1:8080/v1/chat/completions" LLM_MODEL = "fast" MIN_SCORE = 0.82 MAX_CLUSTERS = 15 MAX_FACTS_PER_CLUSTER = 6 APPLY = "--apply" in sys.argv def _http(method: str, url: str, body: dict | None = None, timeout: float = 120.0): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode() or "{}") def clusters_from_graph() -> tuple[dict, list]: g = _http("GET", f"{MEM0}/graph?min_score={MIN_SCORE}&top_k=5", timeout=180) nodes = {n["id"]: n for n in g.get("nodes", [])} parent: dict = {} def find(x): parent.setdefault(x, x) while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x for e in g.get("edges", []): a, b = find(e["source"]), find(e["target"]) if a != b: parent[a] = b groups: dict = {} for nid in parent: groups.setdefault(find(nid), []).append(nid) clusters = [sorted(v) for v in groups.values() if len(v) >= 2] clusters.sort(key=len, reverse=True) return nodes, clusters[:MAX_CLUSTERS] def judge(facts: list[dict]) -> list[dict]: """LLM entscheidet pro Cluster. Fehler => [] (nichts tun).""" listing = "\n".join( f"- id={f['id']} | kategorie={f.get('category')} | stand={f.get('updated_at', '')[:10]} | {f.get('content', '')}" for f in facts) system = ( "Du konsolidierst ein Langzeitgedaechtnis. Unten steht EINE Gruppe semantisch " "aehnlicher Fakten. Entscheide konservativ: " "(a) Sind es Facetten DESSELBEN Fakts: fasse sie in EINEM praezisen deutschen Satz " "zusammen -> genau ein update auf die beste id, die anderen delete. " "(b) Widersprechen sie sich: der Fakt mit dem NEUESTEN Stand gewinnt -> veraltete delete. " "(c) Sind es VERSCHIEDENE Fakten (nur thematisch verwandt): TUE NICHTS. " "Im Zweifel IMMER (c). Antworte NUR als JSON: " '{"actions":[{"op":"update","id":"...","content":"..."},{"op":"delete","id":"..."}]} ' 'oder {"actions":[]}.' ) try: resp = _http("POST", LLM, { "model": LLM_MODEL, "messages": [{"role": "system", "content": system}, {"role": "user", "content": "Fakten-Gruppe:\n" + listing}], "temperature": 0.1, "max_tokens": 800, "response_format": {"type": "json_object"}, "chat_template_kwargs": {"enable_thinking": False}, }) raw = resp["choices"][0]["message"]["content"] or "{}" actions = json.loads(raw).get("actions", []) except Exception as exc: print(f" (LLM-Fehler, Gruppe unangetastet: {exc})") return [] # Harte Validierung — Leitplanken siehe Docstring. ids = {f["id"] for f in facts} updates = [a for a in actions if a.get("op") == "update" and a.get("id") in ids and isinstance(a.get("content"), str) and 0 < len(a["content"]) <= 600] deletes = [a for a in actions if a.get("op") == "delete" and a.get("id") in ids] if len(updates) > 1: return [] del_ids = {a["id"] for a in deletes} - {a["id"] for a in updates} if len(del_ids) >= len(ids): return [] return updates + [{"op": "delete", "id": i} for i in sorted(del_ids)] def main() -> None: t0 = time.time() nodes, clusters = clusters_from_graph() total = len(nodes) print(f"mem0-Konsolidierung {'(APPLY)' if APPLY else '(DRY-RUN)'} — " f"{total} Fakten, {len(clusters)} Gruppen >= {MIN_SCORE}") n_upd = n_del = 0 for ci, cl in enumerate(clusters, 1): facts = [nodes[i] for i in cl][:MAX_FACTS_PER_CLUSTER] actions = judge(facts) if not actions: continue print(f"Gruppe {ci} ({len(facts)} Fakten):") for f in facts: print(f" [{f['id'][:8]}] {f.get('content', '')[:90]}") for a in actions: if a["op"] == "update": print(f" -> UPDATE {a['id'][:8]}: {a['content'][:90]}") n_upd += 1 if APPLY: _http("PUT", f"{MEM0}/memory/{a['id']}", {"content": a["content"]}) else: print(f" -> DELETE {a['id'][:8]}") n_del += 1 if APPLY: _http("DELETE", f"{MEM0}/memory/{a['id']}") print(f"Ergebnis: {n_upd} zusammengefasst, {n_del} geloescht, " f"{total - (n_del if APPLY else 0)} Fakten verbleiben. " f"({int(time.time() - t0)} s)") if __name__ == "__main__": main()