mission-control-v2/deploy/mem0-konsolidierung.py

#!/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/<id>).

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()