Hitonabi/mission-control-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Gedaechtnis-Bereiche: alltag/projekt-Feld statt zweitem mem0

Browse Source 
- classify_facts ordnet neue Fakten zusaetzlich einem Bereich zu (alltag=Person,
  projekt=Technik); Metadatum bereich, in /memory-Items sichtbar.
- GET /memory?bereich=... filtert; Alt-Fakten ohne Feld laufen IMMER mit.
- docs/wissen/GEDAECHTNIS-BEREICHE.md: Entscheid (kein zweites mem0), Stand,
  4-Wochen-Selbstpruefung und kompletter Bauauftrag fuer den Recall-Filter -
  damit die Werkstatt das ohne externe Hilfe fertigbauen kann.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
2026-07-19 14:38:12 +02:00
parent a1d7a4b6b9
commit 6b03a2c3cf
2 changed files with 85 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mem0_service/app.py
+40 -11
View File
@@ -166,6 +166,7 @@ def _to_item(r: dict, score=None) -> dict:
"id": r.get("id"),
"content": r.get("memory", ""),
"category": meta.get("category") or "stable",
"bereich": meta.get("bereich") or "",
"source": meta.get("source") or "auto",
"created_at": r.get("created_at") or "",
"updated_at": r.get("updated_at") or r.get("created_at") or "",
@@ -235,8 +236,10 @@ def _rerank(query: str, items: list[dict]) -> list[dict]:
@app.get("/memory")
def list_memory(q: str = "", category: str = "") -> list[dict]:
"""q gesetzt → semantische Suche (mit Relevanz-Score). Sonst → alle Fakten."""
def list_memory(q: str = "", category: str = "", bereich: str = "") -> list[dict]:
"""q gesetzt → semantische Suche (mit Relevanz-Score). Sonst → alle Fakten.
bereich='alltag'|'projekt' filtert (Fakten OHNE Bereichs-Feld laufen immer mit,
damit Alt-Bestand nie verschwindet)."""
if q:
res = mem().search(q, filters={"user_id": USER_ID}, top_k=SEARCH_TOP_K)
items = [_to_item(r, r.get("score")) for r in res.get("results", [])]
@@ -248,6 +251,8 @@ def list_memory(q: str = "", category: str = "") -> list[dict]:
items.sort(key=lambda i: i.get("created_at", ""), reverse=True)
if category:
items = [i for i in items if i["category"] == category]
if bereich in ("alltag", "projekt"):
items = [i for i in items if i.get("bereich") in (bereich, "")]
return items
@@ -349,8 +354,14 @@ def graph(min_score: float = 0.45, top_k: int = 3) -> dict:
def classify_facts(facts: list[tuple[str, str]]) -> dict:
"""Ordnet jeden Fakt per LLM GENAU einer Kategorie zu → {id: category}.
Nutzt dasselbe (Thinking-freie) Hirn wie die Extraktion. Fehler = leeres Mapping (Fallback)."""
"""Ordnet jeden Fakt per LLM einer Kategorie UND einem Bereich zu →
{id: {"category": <slug>, "bereich": "alltag"|"projekt"}}.
Bereich (19.07.2026, Trennung der Welten): "alltag" = der Nutzer als Person
(Vorlieben, Leben, Arbeitsweise, Entscheidungen), "projekt" = Technik-/Code-/
Infrastruktur-Stände. Grundlage für einen späteren Recall-Filter, damit
Projekt-Schutt nicht in Alltags-Gespräche streut — OHNE zweites Gedächtnis.
Nutzt dasselbe (Thinking-freie) Hirn wie die Extraktion. Fehler = leeres Mapping."""
if not facts:
return {}
import json as _json
@@ -359,7 +370,11 @@ def classify_facts(facts: list[tuple[str, str]]) -> dict:
system = (
"Du bist ein präziser Klassifikator für ein Langzeitgedächtnis. Ordne jeden Fakt GENAU EINER "
"Kategorie zu. Verfügbare Kategorien (nur diese Slugs verwenden):\n" + cat_lines +
"\nAntworte NUR als JSON-Objekt der Form {\"<id>\": \"<slug>\", …}."
"\nOrdne zusätzlich jeden Fakt GENAU EINEM Bereich zu: 'alltag' (der Nutzer als Person: "
"Vorlieben, Leben, Anrede, Arbeitsweise, grundsätzliche Entscheidungen) oder 'projekt' "
"(Technik: Code, Modelle, Server, Infrastruktur, Projekt- und Fehlerstände). "
"Antworte NUR als JSON-Objekt der Form "
"{\"<id>\": {\"kategorie\": \"<slug>\", \"bereich\": \"alltag|projekt\"}, …}."
)
body = "\n".join(f"{fid}: {txt}" for fid, txt in facts)
try:
@@ -370,7 +385,18 @@ def classify_facts(facts: list[tuple[str, str]]) -> dict:
)
m = _re.search(r"\{.*\}", resp or "", _re.S)
data = _json.loads(m.group(0)) if m else {}
return {str(k): v for k, v in data.items() if v in CATEGORIES}
out: dict = {}
for k, v in data.items():
if isinstance(v, str): # Alt-Form {"id": "slug"} weiter akzeptieren
if v in CATEGORIES:
out[str(k)] = {"category": v, "bereich": ""}
elif isinstance(v, dict):
cat = v.get("kategorie") or v.get("category") or ""
ber = v.get("bereich") or ""
if cat in CATEGORIES or ber in ("alltag", "projekt"):
out[str(k)] = {"category": cat if cat in CATEGORIES else DEFAULT_CATEGORY,
"bereich": ber if ber in ("alltag", "projekt") else ""}
return out
except Exception:
log.exception("Auto-Klassifikation fehlgeschlagen")
return {}
@@ -434,17 +460,20 @@ def learn(body: LearnIn) -> dict:
if r.get("id") and r.get("memory") and r.get("event") in ("ADD", "UPDATE")]
cats = classify_facts(to_classify)
for r in results:
cat = cats.get(r.get("id"))
if cat:
info = cats.get(r.get("id"))
if info:
meta = {"category": info["category"], "source": body.source}
if info.get("bereich"):
meta["bereich"] = info["bereich"]
try:
mem().update(r["id"], data=r.get("memory", ""),
metadata={"category": cat, "source": body.source})
mem().update(r["id"], data=r.get("memory", ""), metadata=meta)
except Exception:
log.debug("Kategorie-Update fehlgeschlagen für %s", r.get("id"))
return {"results": [
{"id": r.get("id"), "content": r.get("memory"), "event": r.get("event"),
"category": cats.get(r.get("id"), body.category)}
"category": (cats.get(r.get("id")) or {}).get("category", body.category),
"bereich": (cats.get(r.get("id")) or {}).get("bereich", "")}
for r in results
]}