| """
|
| mnemo — a memory layer for AI agents. (brand: Mnemosyne)
|
|
|
| The memory that runs an autonomous research OS over ~5,800 notes, distilled to a single file with
|
| no required dependencies. It does the four things agent memory actually needs, the way that held up
|
| in production:
|
|
|
| remember(text) append-only raw capture, stamped with an ABSOLUTE time (never rewritten)
|
| recall(query, k) value-ranked retrieval: relevance × the memory's accrued value, not just
|
| cosine similarity — the high-value memories surface first
|
| consolidate(cap) the "dream" pass: value-rank under a keep-budget, link near-duplicates, mark
|
| stale/superseded — it only ADDS a derived layer, it never edits the raw note
|
| contradictions() flag mutually-incompatible memories for REVIEW (never auto-delete)
|
|
|
| Design rules that are not optional (each one cost us to learn):
|
| • Raw capture is immutable. Consolidation adds links/markers; it never overwrites the source —
|
| that is what stops the slow accuracy drift of LLM-rewritten memory.
|
| • Absolute timestamps at write time. Relative/derived times rot the moment they're consolidated.
|
| • Value-ranked, capacity-aware consolidation. The payoff from ranking *what to keep* scales
|
| super-linearly as the budget shrinks (measured), so retention tracks value, not recency — and
|
| NOT access-frequency: decaying on reads keeps *popular* memories, but popularity != value, so a
|
| pure access-reset policy starves the rarely-read-but-load-bearing fact (measured: it retains
|
| ~3x less total value than a value blend under a tight budget). Forgetting blends value + recency.
|
| • Report value at the COHORT level (tag / time-block), never per-memory: per-item value at n-of-1
|
| is statistical noise; cohorts are where the signal lives.
|
| • Contradictions are flagged for review, not auto-resolved. Silent rewrites destroy trust.
|
|
|
| Bring your own embedder for semantic recall (any text->vector fn); with none, mnemo falls back to a
|
| lexical token overlap so it runs anywhere, today.
|
|
|
| from mnemo import Mnemo
|
| m = Mnemo("memory.json") # or Mnemo("memory.json", embed=my_embedder)
|
| m.remember("Pre-trend tests catch only ~31% of fatal DiD bias.", tags=["causal"], value=3)
|
| m.recall("difference in differences", k=5)
|
| m.consolidate(keep=200)
|
| m.contradictions()
|
|
|
| MIT-licensed. Part of Agora (https://github.com/DanceNitra/agora).
|
| """
|
| from __future__ import annotations
|
|
|
| import hashlib
|
| import json
|
| import math
|
| import os
|
| import re
|
| import time
|
| import uuid
|
| from pathlib import Path
|
|
|
| try:
|
| import numpy as _np
|
| except Exception:
|
| _np = None
|
|
|
| try:
|
| from cryptography.hazmat.primitives.asymmetric.ed25519 import (
|
| Ed25519PrivateKey as _Ed25519SK, Ed25519PublicKey as _Ed25519PK)
|
| from cryptography.hazmat.primitives import serialization as _ser
|
| _HAVE_ED = True
|
| except Exception:
|
| _HAVE_ED = False
|
|
|
| _GENESIS = "0" * 64
|
|
|
|
|
| def _canon(obj) -> bytes:
|
| return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
|
|
|
|
| def _sha256_hex(b: bytes) -> str:
|
| return hashlib.sha256(b).hexdigest()
|
|
|
|
|
| def new_receipt_keypair():
|
| """Return (private_key_hex, public_key_hex) for signing mnemo write receipts. Needs `cryptography`."""
|
| if not _HAVE_ED:
|
| raise RuntimeError("signing write receipts needs the `cryptography` package (pip install cryptography)")
|
| sk = _Ed25519SK.generate()
|
| return (sk.private_bytes(_ser.Encoding.Raw, _ser.PrivateFormat.Raw, _ser.NoEncryption()).hex(),
|
| sk.public_key().public_bytes(_ser.Encoding.Raw, _ser.PublicFormat.Raw).hex())
|
|
|
| __version__ = "0.4.1"
|
| _WORD = re.compile(r"[a-z0-9][a-z0-9\-']{2,}")
|
| _STOP = frozenset("the a an of for to in on and or is are was were be been with this that it its as "
|
| "by at from into our we us you your he she they them his her their not no".split())
|
|
|
|
|
| def _stem(w: str) -> str:
|
| return w[:-1] if (w.endswith("s") and len(w) > 4) else w
|
|
|
|
|
| def _tokens(text: str) -> set:
|
| return {_stem(w) for w in _WORD.findall((text or "").lower()) if w not in _STOP}
|
|
|
|
|
| def _token_counts(text: str) -> dict:
|
| """Term-frequency map with the SAME tokenization as _tokens (stem + stopword filter). BM25 needs TF;
|
| _tokens loses it by returning a set."""
|
| d: dict = {}
|
| for w in _WORD.findall((text or "").lower()):
|
| if w in _STOP:
|
| continue
|
| s = _stem(w)
|
| d[s] = d.get(s, 0) + 1
|
| return d
|
|
|
|
|
| def _cosine(a, b) -> float:
|
| dot = sum(x * y for x, y in zip(a, b))
|
| na = math.sqrt(sum(x * x for x in a)) or 1.0
|
| nb = math.sqrt(sum(y * y for y in b)) or 1.0
|
| return dot / (na * nb)
|
|
|
|
|
| class Mnemo:
|
| def __init__(self, path: str | None = None, embed=None, receipts: bool = False,
|
| receipt_key: str | None = None, receipt_pubkey: str | None = None):
|
| """path: optional JSON file to persist to. embed: optional fn(str)->list[float] for semantic
|
| recall; if omitted, recall uses lexical token overlap (zero dependencies).
|
|
|
| receipts/receipt_key (OPT-IN, default OFF -> identical legacy behavior): when enabled, every
|
| remember() appends a tamper-evident, hash-chained WRITE RECEIPT committing to the memory's
|
| content hash, persisted to a sidecar "<path>.receipts.json" (the main store format is unchanged).
|
| verify_writes() then proves the write history wasn't altered out-of-band — something an
|
| append-only store alone can't, because anyone who can edit the store file can rewrite a stored
|
| memory and the store would serve the altered text as original. The hash chain is zero-dependency;
|
| pass receipt_key (+ receipt_pubkey) from new_receipt_keypair() to also Ed25519-SIGN each receipt
|
| so a third party can verify it with the public key only. (Standalone version: agora-agent-receipts.)"""
|
| self.path = Path(path) if path else None
|
| self.embed = embed
|
| self.items: list[dict] = []
|
| self._tok_cache: dict[str, set] = {}
|
| self._tc_cache: dict[str, dict] = {}
|
|
|
|
|
| self.semantic_threshold = 300
|
| self._last_mode = "lexical"
|
| self._mat = None
|
| self._vec_rowof: dict[str, int] = {}
|
| self._mat_built_n = -1
|
| self._vec_mean = None
|
|
|
|
|
|
|
|
|
| self.center_embeddings = True
|
|
|
|
|
|
|
|
|
|
|
|
|
| self.two_tier_keep = True
|
| self.protect_frac = 0.30
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| self.supersede_requires_corroboration = False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| self.supersede_persistence = 0
|
|
|
|
|
|
|
|
|
| self._save_min_s = 5.0
|
| self._last_save = 0.0
|
| self._dirty = False
|
| if self.path and self.path.exists():
|
| try:
|
| self.items = json.loads(self.path.read_text(encoding="utf-8"))
|
| except Exception:
|
| self.items = []
|
|
|
| self.receipts_enabled = bool(receipts or receipt_key)
|
| self._receipt_sk = receipt_key
|
| self.receipt_pubkey = receipt_pubkey
|
| self._receipts: list[dict] = []
|
| self._receipts_path = (self.path.parent / (self.path.name + ".receipts.json")) if self.path else None
|
| if self.receipts_enabled and self._receipts_path and self._receipts_path.exists():
|
| try:
|
| self._receipts = json.loads(self._receipts_path.read_text(encoding="utf-8"))
|
| except Exception:
|
| self._receipts = []
|
|
|
|
|
| def remember(self, text: str, tags=None, value: float = 1.0, meta: dict | None = None,
|
| mtype: str | None = None, valid_from: float | None = None,
|
| source: dict | None = None, key: str | None = None) -> str:
|
| """Append-only raw capture. Stamped with an absolute UTC time; never edited afterward.
|
| mtype in {episodic, semantic, procedural} sets the decay prior (episodic fades fast,
|
| semantic slow, procedural barely); inferred from the text if not given. Pass it explicitly
|
| when the caller knows the kind — inference defaults to episodic (the conservative, fast-decay
|
| choice) and only promotes on clear markers.
|
|
|
| `key` (OPT-IN) is a deterministic supersession key — typically a (subject, relation) identifier,
|
| e.g. "billing-api::auth-method". When set, remembering a new value RETIRES every active record
|
| sharing the same key (status -> superseded), with NO similarity threshold and NO LLM call. This
|
| closes the 'supersession blind spot': cosine similarity cannot tell a contradicted fact from its
|
| replacement (we measured AUROC ~0.61, near chance — a contradiction is often MORE embedding-similar
|
| to the original than a rephrase is), so a similarity-based store silently serves the stale value
|
| (~42% of the time in our test). A deterministic (subject, relation, object) ledger drives that to
|
| ~0%. Bi-temporal: a back-filled record (earlier valid_from) does NOT overwrite a genuinely newer
|
| same-key value — the stale-on-arrival record is the one retired."""
|
| mid = uuid.uuid4().hex[:10]
|
| now = time.time()
|
| rec = {"id": mid, "text": text, "tags": list(tags or []), "value": float(value),
|
| "ts": now, "iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
| "valid_from": float(valid_from) if valid_from is not None else now,
|
| "source": dict(source) if source else None,
|
| "mtype": mtype or _infer_type(text), "last_access": now,
|
| "status": "active", "links": [], "meta": dict(meta or {})}
|
| if key is not None:
|
| rec["key"] = str(key)
|
| if self.embed:
|
| try:
|
| rec["vec"] = list(self.embed(text))
|
| except Exception:
|
| rec["vec"] = None
|
| self.items.append(rec)
|
| if key is not None:
|
| self._supersede_by_key(rec)
|
| self._save(force=True)
|
| if self.receipts_enabled:
|
| self._emit_write_receipt(rec)
|
| return mid
|
|
|
|
|
| def _write_commit(self, rec: dict) -> dict:
|
| """What a receipt commits to for a stored memory: its id + a hash of its content-bearing fields."""
|
| return {"id": rec["id"],
|
| "content_sha256": _sha256_hex(_canon({"text": rec.get("text"), "key": rec.get("key"),
|
| "mtype": rec.get("mtype")}))}
|
|
|
| def _emit_write_receipt(self, rec: dict) -> dict:
|
| prev = self._receipts[-1]["hash"] if self._receipts else _GENESIS
|
| r = {"seq": len(self._receipts), "ts": rec.get("ts"), "memory_id": rec["id"],
|
| "commit": self._write_commit(rec), "prev": prev}
|
| r["hash"] = _sha256_hex(_canon({k: r[k] for k in ("seq", "ts", "memory_id", "commit", "prev")}))
|
| if self._receipt_sk and _HAVE_ED:
|
| sk = _Ed25519SK.from_private_bytes(bytes.fromhex(self._receipt_sk))
|
| r["pubkey"] = self.receipt_pubkey
|
| r["sig"] = sk.sign(bytes.fromhex(r["hash"])).hex()
|
| self._receipts.append(r)
|
| if self._receipts_path:
|
| try:
|
| self._receipts_path.write_text(json.dumps(self._receipts, indent=2, ensure_ascii=False),
|
| encoding="utf-8")
|
| except Exception:
|
| pass
|
| return r
|
|
|
| def verify_writes(self, expected_pubkey: str | None = None) -> tuple[bool, list[str]]:
|
| """Verify the write-receipt chain AND that each stored memory still matches its write receipt.
|
| Returns (ok, problems). Catches out-of-band edits to the store the normal flow can't see.
|
| Requires receipts to have been enabled at write time."""
|
| problems: list[str] = []
|
| prev = _GENESIS
|
| by_id = {it["id"]: it for it in self.items}
|
| for i, r in enumerate(self._receipts):
|
| core = {k: r.get(k) for k in ("seq", "ts", "memory_id", "commit", "prev")}
|
| if r.get("prev") != prev:
|
| problems.append(f"receipt {i}: broken chain link (a prior receipt was altered/removed)")
|
| if _sha256_hex(_canon(core)) != r.get("hash"):
|
| problems.append(f"receipt {i}: receipt tampered (hash mismatch)")
|
| if "sig" in r and _HAVE_ED:
|
| try:
|
| _Ed25519PK.from_public_bytes(bytes.fromhex(r["pubkey"])).verify(
|
| bytes.fromhex(r["sig"]), bytes.fromhex(r["hash"]))
|
| if expected_pubkey and r.get("pubkey") != expected_pubkey:
|
| problems.append(f"receipt {i}: signed by an unexpected key")
|
| except Exception:
|
| problems.append(f"receipt {i}: invalid signature")
|
| elif expected_pubkey:
|
| problems.append(f"receipt {i}: unsigned, but a signature was required")
|
| cur = by_id.get(r["memory_id"])
|
| if cur is None:
|
| problems.append(f"memory {r['memory_id']}: written but missing from the store (deleted out-of-band)")
|
| elif self._write_commit(cur) != r["commit"]:
|
| problems.append(f"memory {r['memory_id']}: stored content no longer matches its write receipt (edited after write)")
|
| prev = r.get("hash")
|
| return (len(problems) == 0, problems)
|
|
|
| def _supersede_by_key(self, rec: dict) -> None:
|
| """Deterministic (subject, relation, object) supersession: retire active records that share
|
| rec['key']. No similarity threshold, no LLM call — the fix our Crucible replication validated
|
| (stale-fact recall 41.7% -> 0.0%, where cosine-based detection is near chance at AUROC ~0.61).
|
| Bi-temporal: only same-key records with valid_from <= rec's are retired; if an active same-key
|
| record is genuinely newer (later valid_from), the INCOMING rec is the stale one and is retired
|
| instead — a back-filled value never overwrites the current one. recall() hides superseded records
|
| by default, so a keyed store never surfaces a stale fact."""
|
| k = rec.get("key")
|
| if not k:
|
| return
|
| vf_new = rec.get("valid_from", rec["ts"])
|
| for r in self.items:
|
| if r is rec or r.get("status") != "active" or r.get("key") != k:
|
| continue
|
| vf_r = r.get("valid_from", r["ts"])
|
| if vf_r <= vf_new:
|
| r["status"] = "superseded"
|
| r["superseded_ts"] = time.time()
|
| r["invalidated_at"] = vf_new
|
| r.setdefault("meta", {})["superseded_by_toggle"] = rec["id"]
|
| else:
|
| rec["status"] = "superseded"
|
| rec["superseded_ts"] = time.time()
|
| rec["invalidated_at"] = vf_r
|
| rec.setdefault("meta", {})["superseded_by_toggle"] = r["id"]
|
|
|
| def remember_dedup(self, text: str, tags=None, value: float = 1.0, meta: dict | None = None,
|
| mtype: str | None = None, dup_threshold: float = 0.95) -> str:
|
| """OPT-IN write that skips redundant appends. If an active memory is near-identical (similarity >=
|
| dup_threshold) AND carries the SAME value(s) (no numeric clash), this returns that memory's id WITHOUT
|
| appending a duplicate raw row -- cutting raw-store bloat from repeated identical writes. A near-identical
|
| text with a DIFFERENT number (a value UPDATE) is NOT a duplicate: it appends, so the consolidation pass can
|
| supersede the stale value. Default `remember()` stays strictly append-only (the 'zero rewrites' contract);
|
| this is a separate opt-in path for high-duplicate ingest."""
|
| hits = self.recall(text, k=1)
|
| if hits:
|
| h = hits[0]
|
| s = self._similarity(text, h, self._qvec(text) if self.embed else None)
|
| if s >= dup_threshold and not _value_clash(text, h["text"]):
|
| return h["id"]
|
| return self.remember(text, tags=tags, value=value, meta=meta, mtype=mtype)
|
|
|
| def forget(self, ids=None, where=None, redact_links: bool = True) -> dict:
|
| """HARD-DELETE memories — the one operation that genuinely REMOVES content. mnemo is otherwise
|
| append-only: supersession / invalidation only DEMOTE a record (it still exists, recallable with
|
| include_superseded). forget() is for the cases where demotion is not enough: a right-to-be-forgotten
|
| / erasure request, a poisoned or libellous memory, or a hard correction.
|
|
|
| Select by `ids` (a single id or an iterable) and/or `where` (a predicate fn(record)->bool; e.g.
|
| lambda r: 'secret' in r['text']). VERIFIED FORGETTING: the matched records are deleted AND their ids
|
| are scrubbed from every surviving record's `links` and toggle-supersession pointers, and the cached
|
| vec matrix + token caches are dropped — so a forgotten memory cannot resurface via recall, via a
|
| consolidation link, or via a stale derived-summary pointer. This is complete because consolidation
|
| never copies raw text into other records (it only links ids and toggles status) — there is no merged
|
| blob left holding the forgotten content. Returns {forgotten, ids, scrubbed_links}."""
|
| target = set()
|
| if ids is not None:
|
| target |= ({ids} if isinstance(ids, str) else set(ids))
|
| if where is not None:
|
| for r in self.items:
|
| try:
|
| if where(r):
|
| target.add(r["id"])
|
| except Exception:
|
| pass
|
| target &= {r["id"] for r in self.items}
|
| if not target:
|
| return {"forgotten": 0, "ids": [], "scrubbed_links": 0}
|
| self.items = [r for r in self.items if r["id"] not in target]
|
| scrubbed = 0
|
| if redact_links:
|
| for r in self.items:
|
| if r.get("links"):
|
| before = len(r["links"])
|
| r["links"] = [l for l in r["links"] if l not in target]
|
| scrubbed += before - len(r["links"])
|
| meta = r.get("meta")
|
| if meta and meta.get("superseded_by_toggle") in target:
|
| meta.pop("superseded_by_toggle", None)
|
| for tid in target:
|
| self._tok_cache.pop(tid, None)
|
| self._mat = None; self._mat_built_n = -1
|
| self._save(force=True)
|
| return {"forgotten": len(target), "ids": sorted(target), "scrubbed_links": scrubbed}
|
|
|
|
|
| def _qvec(self, query: str):
|
| """Embed a query ONCE per scan, or None (no embedder / failure). Callers pass the result
|
| into _similarity so a recall over N memories costs 1 embedding, not N."""
|
| if not self.embed:
|
| return None
|
| try:
|
| return self.embed(query)
|
| except Exception:
|
| return None
|
|
|
| def _vec_matrix(self):
|
| """Cached L2-normalized matrix (numpy) of every memory that carries a vec — so a semantic
|
| recall is ONE matmul, not an O(N·d) pure-Python cosine loop. Rebuilt only when the item count
|
| changes (remember / bulk load); status changes (consolidate) don't touch the vectors."""
|
| if _np is None:
|
| return None
|
| if self._mat is None or self._mat_built_n != len(self.items):
|
| rows, ids = [], []
|
| for r in self.items:
|
| if r.get("vec"):
|
| rows.append(r["vec"]); ids.append(r["id"])
|
| if rows:
|
| M = _np.asarray(rows, dtype=_np.float32)
|
| self._vec_mean = M.mean(axis=0)
|
| if self.center_embeddings:
|
| M = M - self._vec_mean
|
| M /= (_np.linalg.norm(M, axis=1, keepdims=True) + 1e-9)
|
| self._mat = M
|
| self._vec_rowof = {i: k for k, i in enumerate(ids)}
|
| else:
|
| self._mat, self._vec_rowof, self._vec_mean = None, {}, None
|
| self._mat_built_n = len(self.items)
|
| return self._mat
|
|
|
| def _rec_tokens(self, rec: dict) -> set:
|
| """Token set for a memory, cached by id — recall over N memories shouldn't re-tokenize."""
|
| rid = rec.get("id") or id(rec)
|
| t = self._tok_cache.get(rid)
|
| if t is None:
|
| t = _tokens(rec["text"]); self._tok_cache[rid] = t
|
| return t
|
|
|
| def _rec_tokcount(self, rec: dict) -> dict:
|
| """Term-frequency map for a memory, cached by id (for the BM25 hybrid channel)."""
|
| rid = rec.get("id") or id(rec)
|
| c = self._tc_cache.get(rid)
|
| if c is None:
|
| c = _token_counts(rec["text"]); self._tc_cache[rid] = c
|
| return c
|
|
|
| def _bm25_scores(self, qtok: set, pool: list, k1: float = 1.5, b: float = 0.75) -> list:
|
| """Okapi BM25 score of `query` (token set) against every record in `pool` — the strong lexical
|
| channel for the hybrid. df/avgdl are computed over the pool (the live corpus). Returns a list of
|
| scores aligned to `pool`. Pure-Python, zero-dependency. We MEASURED BM25 (not token-overlap) as the
|
| lexical channel that makes the hybrid beat either alone (mnemo/probes/locomo_retrieval_map.py)."""
|
| N = len(pool)
|
| if N == 0:
|
| return []
|
| counts = [self._rec_tokcount(r) for r in pool]
|
| dl = [sum(c.values()) for c in counts]
|
| avgdl = (sum(dl) / N) or 1.0
|
| df: dict = {}
|
| for c in counts:
|
| for t in c:
|
| df[t] = df.get(t, 0) + 1
|
| idf = {t: math.log(1 + (N - n + 0.5) / (n + 0.5)) for t, n in df.items()}
|
| out = []
|
| for c, L in zip(counts, dl):
|
| s = 0.0
|
| for t in qtok:
|
| f = c.get(t, 0)
|
| if f:
|
| s += idf.get(t, 0.0) * (f * (k1 + 1)) / (f + k1 * (1 - b + b * L / avgdl))
|
| out.append(s)
|
| return out
|
|
|
| def _similarity(self, query: str, rec: dict, qvec=None, qtok: set | None = None) -> float:
|
| if qvec is not None and rec.get("vec"):
|
| return max(0.0, _cosine(qvec, rec["vec"]))
|
| q = qtok if qtok is not None else _tokens(query)
|
| t = self._rec_tokens(rec)
|
| if not q or not t:
|
| return 0.0
|
| return len(q & t) / min(len(q), len(t))
|
|
|
| def recall(self, query: str, k: int = 6, include_superseded: bool = False,
|
| include_hubs: bool = False, mode: str = "auto", min_relevance: float = 0.0,
|
| scope: str | None = None, as_of: float | None = None,
|
| where: dict | None = None, influence_only: bool = False,
|
| prefer: dict | None = None, prefer_trust: float = 1.0) -> list[dict]:
|
| """Top-k memories by RELEVANCE × VALUE — high-value memories outrank merely-similar ones.
|
| Memories the dream pass flagged as hubs (universal matchers) are skipped unless include_hubs.
|
|
|
| mode: 'auto' (default) uses LEXICAL token overlap while the store is small (< semantic_threshold
|
| active memories) and a LEXICAL+SEMANTIC HYBRID (Reciprocal Rank Fusion) once it grows past that —
|
| the hybrid robustly beat either channel alone in our agent-memory benchmark (details in the recall
|
| body / mnemo/probes/locomo_retrieval_map.py). Force a single channel with mode='lexical' /
|
| 'semantic', or the fusion explicitly with mode='hybrid'. Semantic/hybrid need an embedder (set on
|
| the store); without one, or if embedding fails, recall falls back to lexical automatically.
|
|
|
| where: an OPT-IN metadata pre-filter applied to the candidate pool BEFORE ranking — the cheap
|
| 'filter before you rank' lever (measured on LoCoMo: a metadata pre-filter can beat retriever choice;
|
| mnemo/probes/locomo_metadata_prefilter.py). A dict of field -> condition; a record must match ALL
|
| fields (AND). Each field is matched against the record's top-level attributes first, then its meta
|
| dict, so both `valid_from`/`mtype`/`key` and any `meta` key work. A condition is either a scalar
|
| (equality), a list/tuple/set (membership), or a dict of operators:
|
| {"$gte","$lte","$gt","$lt","$in","$nin","$ne","$contains"} — e.g. a time range
|
| where={"valid_from": {"$gte": t0, "$lte": t1}} (hard-filter the SOLVED half), or a closed-set
|
| entity where={"speaker": {"$in": ["Caroline","Mel"]}}. NOTE: this is a HARD filter — a record that
|
| doesn't match is removed, so on lossy/predicted extraction prefer a broad/loose filter (or rerank)
|
| over an aggressive one, since a wrong filter hard-deletes the answer (measured harm mode).
|
|
|
| influence_only (OPT-IN, default False -> zero behavior change): restrict the result to CORROBORATED
|
| memories — those that meet the same bar mnemo uses for episodic->semantic GRADUATION (an EARNED
|
| net-positive outcome via credit() [good>0 and good>=bad], OR already-graduated 'semantic' type, OR
|
| >=2 DISTINCT-canonical-source corroborating links). This is the retrieve-then-INFLUENCE split: recall
|
| freely for context, but call with influence_only=True for the set that is allowed to DRIVE an action.
|
| MEASURED (mnemo/probes/agentpoison_influence_gate*.py) against a real AgentPoison-style single-
|
| instance retrieval-poisoning attack (Chen et al., NeurIPS 2024, arXiv:2407.12784; PoisonedRAG, Zou
|
| et al., arXiv:2402.07867): a natural-sentence trigger hijacks RAW top-1 retrieval 88-100% and is
|
| scale-invariant (60->10k memories), and retrieval-time / embedding-geometry defenses do NOT
|
| generalize across encoders — but influence_only drops the single-instance poison's rank-1 hijack to
|
| 0% on all three tested retrievers (MiniLM/BGE/Contriever) and all scales, because an injected poison
|
| never earns corroboration while legitimate memories earn it through use. It GENERALIZES precisely
|
| because it lives in provenance metadata, not embedding geometry. HONEST COST (calibration tradeoff):
|
| a rare-but-true memory that has not yet earned corroboration is filtered too (measured recall 1.00
|
| corroborated vs 0.08 uncorroborated) — so this is for adversarial / untrusted-ingestion use, where a
|
| recalled-but-uncorroborated memory should inform but not unilaterally drive an action. It RAISES
|
| attacker cost (a single free injection is filtered; defeating it needs >=3 coordinated records with
|
| >=2 independent forged provenances) rather than making poisoning impossible. Reversible: default
|
| False = legacy recall.
|
|
|
| prefer / prefer_trust (OPT-IN, default None -> zero behavior change): a SOFT, trust-weighted metadata
|
| filter. Unlike `where` (a HARD filter that DELETES non-matching records — so a wrong filter hard-
|
| deletes the answer), `prefer` takes the same condition dict but only BOOSTS matching records'
|
| score by (1 + prefer_trust * gain), leaving non-matching records rankable. prefer_trust in [0,1] is
|
| HOW MUCH to trust the filter cue this call: pass a low value when your metadata extractor is unsure
|
| (weak/ambiguous match) so the filter gracefully backs off toward plain recall; prefer_trust=0 == no
|
| filter. This is the a-priori-trust lever: weight the filter by the RELIABILITY of the extraction
|
| (e.g. alias-match strength: exact-name hit -> ~1.0, no-name/ambiguous guess -> ~0.0), NOT by the
|
| extractor model's own self-reported confidence (which is corrupted in the overconfident-on-wrong
|
| case). MEASURED (mnemo/probes/locomo_soft_prefer_filter.py) on LoCoMo: soft prefer weighted by
|
| alias-strength gives the filter's benefit on reliable (exact-name) queries while backing off on
|
| ambiguous ones where extraction fails -- beating both no filter and a hard `where` filter under
|
| imperfect extraction. Reversible: prefer=None = legacy recall."""
|
| def _eligible(r: dict) -> bool:
|
| s = r["status"]
|
| if as_of is not None:
|
|
|
|
|
|
|
| vf = r.get("valid_from", r["ts"])
|
| inv = r.get("invalidated_at")
|
| if vf > as_of or (inv is not None and inv <= as_of):
|
| return False
|
| return include_hubs if s == "hub" else True
|
| if s == "active":
|
| return True
|
| if s == "hub":
|
| return include_hubs
|
| return include_superseded
|
| pool = [r for r in self.items if _eligible(r)]
|
|
|
|
|
|
|
| if scope is not None:
|
| pool = [r for r in pool if (r.get("meta") or {}).get("scope") == scope]
|
|
|
|
|
| if where:
|
| pool = [r for r in pool if self._cond_match(r, where)]
|
|
|
|
|
|
|
| if influence_only:
|
| _byid = {x["id"]: x for x in self.items}
|
| pool = [r for r in pool if self._is_corroborated(r, _byid)]
|
|
|
|
|
|
|
|
|
|
|
|
|
| has_embed = self.embed is not None
|
| if mode == "lexical" or not has_embed:
|
| sel = "lexical"
|
| elif mode in ("semantic", "hybrid"):
|
| sel = mode
|
| else:
|
| sel = "hybrid" if len(pool) >= self.semantic_threshold else "lexical"
|
| qvec = self._qvec(query) if sel in ("semantic", "hybrid") else None
|
| if qvec is None and sel != "lexical":
|
| sel = "lexical"
|
| self._last_mode = sel
|
| qtok = _tokens(query)
|
|
|
| sims_vec = None
|
| if qvec is not None and _np is not None:
|
| M = self._vec_matrix()
|
| if M is not None:
|
| qv = _np.asarray(qvec, dtype=_np.float32)
|
| if self.center_embeddings and self._vec_mean is not None:
|
| qv = qv - self._vec_mean
|
| sims_vec = M @ (qv / (float(_np.linalg.norm(qv)) or 1.0))
|
| _now = time.time()
|
| _by_id = {x["id"]: x for x in self.items}
|
| def _semsim(r) -> float:
|
| if sims_vec is not None and r.get("vec") and r["id"] in self._vec_rowof:
|
| return max(0.0, float(sims_vec[self._vec_rowof[r["id"]]]))
|
| return max(0.0, _cosine(qvec, r["vec"])) if (qvec is not None and r.get("vec")) else 0.0
|
| def _lexsim(r) -> float:
|
| t = self._rec_tokens(r)
|
| return (len(qtok & t) / min(len(qtok), len(t))) if (qtok and t) else 0.0
|
| def _candrec(r, sim):
|
|
|
|
|
|
|
|
|
| stale = bool(r.get("links")) and any(
|
| (_by_id.get(lid, {}).get("meta") or {}).get("superseded_by_toggle") for lid in r["links"])
|
| r["_stale_derived"] = stale
|
| return (sim, 0.5 if stale else 1.0, self._effective_value(r, _now), r)
|
| cands = []
|
|
|
|
|
|
|
|
|
| if sel == "hybrid":
|
| bm = self._bm25_scores(qtok, pool)
|
| scn = []
|
| for r, bx in zip(pool, bm):
|
| sem = _semsim(r)
|
| if (sem <= 0 or sem < min_relevance) and bx <= 0:
|
| continue
|
| scn.append((r, sem, bx))
|
| if scn:
|
| order_sem = sorted(range(len(scn)), key=lambda i: -scn[i][1])
|
| order_bm = sorted(range(len(scn)), key=lambda i: -scn[i][2])
|
| rrf = [0.0] * len(scn)
|
| for rank, i in enumerate(order_sem): rrf[i] += 1.0 / (60 + rank)
|
| for rank, i in enumerate(order_bm): rrf[i] += 1.0 / (60 + rank)
|
| mx = max(rrf) or 1.0
|
| for i, (r, sem, bx) in enumerate(scn):
|
| cands.append(_candrec(r, rrf[i] / mx))
|
| else:
|
| for r in pool:
|
| sim = _semsim(r) if sel == "semantic" else _lexsim(r)
|
| if sim <= 0 or sim < min_relevance:
|
| continue
|
| cands.append(_candrec(r, sim))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| mode = getattr(self, "cal_mode", "full")
|
| gate_off = False
|
| if mode == "gated" and cands:
|
| top = max(c[0] for c in cands)
|
| near = [c for c in cands if c[0] >= top * 0.95]
|
| D = len(near)
|
| if D >= 2:
|
| g = sum(float(c[3].get("good", 0) or 0) for c in near)
|
| b = sum(float(c[3].get("bad", 0) or 0) for c in near)
|
| if (g + 1.0) / (g + b + 2.0) <= 1.0 / (1.0 + D):
|
| gate_off = True
|
|
|
|
|
|
|
|
|
| _pt = max(0.0, min(1.0, float(prefer_trust))) if prefer else 0.0
|
| scored = []
|
| for sim, prov, evalue, r in cands:
|
| if gate_off:
|
| cal = 1.0
|
| else:
|
| cal = 0.5 + self._reliability(r)
|
| if mode == "boost" and cal < 1.0:
|
| cal = 1.0
|
| pref = (1.0 + _pt * _PREFER_GAIN) if (_pt > 0.0 and self._cond_match(r, prefer)) else 1.0
|
| score = sim * (1.0 + math.log1p(max(0.0, evalue))) * prov * cal * pref
|
| scored.append((score, sim, r))
|
| scored.sort(key=lambda x: -x[0])
|
| out = []
|
| _top_sim = scored[0][1] if scored else 1.0
|
| for score, sim, r in scored[:k]:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| rel = (sim / _top_sim) if _top_sim > 0 else 1.0
|
| r["value"] += 0.25 * rel
|
| r["last_access"] = _now
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _good = float(r.get("good", 0) or 0); _bad = float(r.get("bad", 0) or 0)
|
| corroborated = (_good > 0 and _good >= _bad) or self._distinct_sources(r.get("links"), _by_id) >= 2
|
| if r.get("mtype") == "episodic" and r["value"] >= _GRADUATE_VALUE and corroborated:
|
| r["mtype"] = "semantic"
|
| r.setdefault("meta", {})["graduated_from_episodic"] = True
|
| out.append({"id": r["id"], "text": r["text"], "tags": r["tags"], "iso": r["iso"],
|
| "value": round(r["value"], 2), "relevance": round(sim, 3),
|
| "score": round(score, 3), "links": r["links"],
|
| "reliability": round(self._reliability(r), 3),
|
| "source": r.get("source"),
|
| "stale_derived": bool(r.get("_stale_derived"))})
|
|
|
|
|
|
|
|
|
| if out:
|
| self._dirty = True
|
| return out
|
|
|
| @staticmethod
|
| def _canon_source(doc) -> str:
|
| """Entity-resolution canonicalization of a source identifier, so sybil variants of one origin
|
| ('Wikipedia', 'wikipedia.org', 'https://www.wikipedia.org/wiki/X') collapse to a single key."""
|
| s = str(doc or "").strip().lower()
|
| s = re.sub(r"^[a-z]+://", "", s)
|
| s = re.sub(r"^www\.", "", s)
|
| s = s.split("/")[0].split("?")[0]
|
| s = re.sub(r"\.(org|com|net|io|gov|edu|co|ai|dev|info|news)$", "", s)
|
| s = re.sub(r"[^a-z0-9]+", "", s)
|
| return s
|
|
|
| @staticmethod
|
| def _distinct_sources(links, by_id) -> int:
|
| """Count DISTINCT canonical sources among corroborating links — entity resolution BEFORE counting,
|
| so 'three names for one source' sybil variants count as one. A link whose record carries no source
|
| counts as its own id, so genuinely source-less corroboration is not penalised (no regression)."""
|
| keys = set()
|
| for lid in (links or []):
|
| lr = by_id.get(lid)
|
| if lr is None:
|
| continue
|
| src = lr.get("source")
|
| doc = src.get("doc") if isinstance(src, dict) else (src if isinstance(src, str) else None)
|
| keys.add(Mnemo._canon_source(doc) if doc else "id:" + lid)
|
| return len(keys)
|
|
|
| @staticmethod
|
| def _is_corroborated(rec: dict, by_id: dict) -> bool:
|
| """The corroboration bar shared by episodic->semantic graduation and the recall influence gate:
|
| an EARNED net-positive outcome (good>0 and good>=bad — set by credit() on real work, not
|
| self-assertable), OR an already-graduated 'semantic' memory, OR >=2 DISTINCT-canonical-source
|
| corroborating links (sybil variants of one source collapse to one). A single fresh self-asserted
|
| memory (the AgentPoison single-instance poison) meets none of these."""
|
| good = float(rec.get("good", 0) or 0)
|
| bad = float(rec.get("bad", 0) or 0)
|
| if good > 0 and good >= bad:
|
| return True
|
| if rec.get("mtype") == "semantic":
|
| return True
|
| return Mnemo._distinct_sources(rec.get("links"), by_id) >= 2
|
|
|
| @staticmethod
|
| def _cond_match(r: dict, conds: dict) -> bool:
|
| """Does record r satisfy ALL of `conds` (a where/prefer dict)? Each field is matched against the
|
| record's top-level attributes first, then its meta dict. A condition is a scalar (equality), a
|
| list/tuple/set (membership), or a dict of operators ($eq/$ne/$in/$nin/$gte/$lte/$gt/$lt/$contains).
|
| Shared by the hard `where` pre-filter and the soft `prefer` trust-weighted boost."""
|
| meta = r.get("meta") or {}
|
| for field, cond in conds.items():
|
| val = r[field] if field in r else meta.get(field)
|
| if isinstance(cond, dict):
|
| for op, cv in cond.items():
|
| if op in ("$eq", "eq"):
|
| if val != cv: return False
|
| elif op in ("$ne", "ne"):
|
| if val == cv: return False
|
| elif op in ("$in", "in"):
|
| if val not in cv: return False
|
| elif op in ("$nin", "nin"):
|
| if val in cv: return False
|
| elif op in ("$gte", "gte"):
|
| if val is None or val < cv: return False
|
| elif op in ("$lte", "lte"):
|
| if val is None or val > cv: return False
|
| elif op in ("$gt", "gt"):
|
| if val is None or val <= cv: return False
|
| elif op in ("$lt", "lt"):
|
| if val is None or val >= cv: return False
|
| elif op in ("$contains", "contains"):
|
| if val is None or cv not in val: return False
|
| else:
|
| raise ValueError(f"recall condition: unknown operator {op!r}")
|
| elif isinstance(cond, (list, tuple, set)):
|
| if val not in cond: return False
|
| else:
|
| if val != cond: return False
|
| return True
|
|
|
| @staticmethod
|
| def _reliability(r: dict) -> float:
|
| """Per-memory track record as a Beta(1+good, 1+bad) posterior MEAN: 0.5 with no outcomes yet,
|
| ->1 if recalls into it kept resolving WELL, ->0 if they kept resolving badly. Counts only grow."""
|
| g = float(r.get("good", 0) or 0)
|
| b = float(r.get("bad", 0) or 0)
|
| return (g + 1.0) / (g + b + 2.0)
|
|
|
| def credit(self, ids, outcome, weight: float = 1.0) -> dict:
|
| """Close the accuracy loop onto the substrate. When the work a set of memories was recalled into
|
| gets a real verdict (a forecast resolves, a replication is ruled REPRODUCED/FAILED, a hypothesis is
|
| severe-tested), call credit(recalled_ids, outcome): each memory's Beta(good,bad) track record is
|
| nudged so future recall ranks by WAS-IT-RIGHT, not merely was-it-recalled. Append-only to the
|
| counts; never edits raw text. `outcome` may be a bool, a sign (>0 good), or a verdict string
|
| (good/right/correct/reproduced/hit vs bad/wrong/failed/miss)."""
|
| if isinstance(outcome, bool):
|
| good = outcome
|
| elif isinstance(outcome, (int, float)):
|
| good = outcome > 0
|
| else:
|
| s = str(outcome).strip().lower()
|
| good = s in ("good", "right", "correct", "reproduced", "hit", "true", "win", "+")
|
| by_id = {x["id"]: x for x in self.items}
|
| key, updated = ("good" if good else "bad"), []
|
| for i in (ids or []):
|
| rec = by_id.get(i)
|
| if rec is None:
|
| continue
|
| rec[key] = float(rec.get(key, 0) or 0) + float(weight)
|
| updated.append(i)
|
| if updated:
|
| self._save()
|
| return {"updated": updated, "outcome": key, "weight": weight}
|
|
|
| def _effective_value(self, r: dict, now: float) -> float:
|
| """Recall weight = stored value decayed by time since last access, at the memory's TYPE
|
| half-life (episodic fades fast, semantic slow, procedural barely). Access resets the clock,
|
| so memories that keep being useful stay alive while stored-but-never-recalled ones fade.
|
| Reversible: raw value/text are untouched; only the effective ranking weight decays."""
|
| hl = _HALFLIFE_S.get(r.get("mtype", "episodic"), _HALFLIFE_S["episodic"])
|
| age = max(0.0, now - r.get("last_access", r.get("ts", now)))
|
| return r["value"] * (0.5 ** (age / hl))
|
|
|
|
|
| def _common_vocab(self, active: list[dict], min_df_frac: float = 0.002):
|
| """Token sets per memory + the corpus's COMMON vocabulary (tokens shared by enough
|
| memories to be real content, not one-off noise). Cheap, O(total tokens)."""
|
| from collections import Counter
|
| df: Counter = Counter()
|
| toks = []
|
| for r in active:
|
| tk = _tokens(r["text"]); toks.append(tk); df.update(tk)
|
| min_df = max(3, int(min_df_frac * len(active)))
|
| common = {w for w, c in df.items() if c >= min_df}
|
| return toks, common
|
|
|
| def recall_iterative(self, query: str, ask_followup, k: int = 6, rounds: int = 1,
|
| **recall_kw) -> list[dict]:
|
| """Multi-hop recall. One-shot top-k misses evidence reachable only via a BRIDGE entity (a fact whose
|
| detail lives in a memory NOT similar to the query). This does: retrieve -> let a capable model read the
|
| results and name what's missing, emitting follow-up queries -> retrieve again -> merge (dedup by id).
|
| `ask_followup(query, current_results) -> list[str]` is caller-supplied, so mnemo stays model-agnostic
|
| (inject any model/LLM). MEASURED ~3.3x multi-hop full-evidence recall vs one-shot top-k on LoCoMo
|
| (0.057 -> 0.186, n=70 across 3 conversations) — the one mechanism that moved the multi-hop bottleneck
|
| where static retrieval tricks (dense-neighbor, lexical bridges) did not. More expensive (a model call
|
| in the loop), so it's an explicit mode, not the default."""
|
| seen: dict = {}
|
| for r in self.recall(query, k=k, **recall_kw):
|
| seen[r["id"]] = r
|
| for _ in range(max(0, int(rounds))):
|
| try:
|
| followups = ask_followup(query, list(seen.values())) or []
|
| except Exception:
|
| followups = []
|
| for fq in followups:
|
| if not isinstance(fq, str) or not fq.strip():
|
| continue
|
| for r in self.recall(fq, k=k, **recall_kw):
|
| seen.setdefault(r["id"], r)
|
| return list(seen.values())
|
|
|
| def consolidate(self, keep: int | None = None, dup_threshold: float = 0.82,
|
| hub_coverage: float = 0.12, link_duplicates: bool = True) -> dict:
|
| """The dream pass. ADDS a derived layer (status + links); never edits raw text. Three steps:
|
|
|
| 1. HUB PASS — flag indiscriminate "universal-matcher" memories. Under lexical recall the
|
| similarity is the overlap coefficient |q∩t|/min(|q|,|t|), so a memory whose token set
|
| covers a large fraction of the corpus's common vocabulary scores ~1.0 against ALMOST ANY
|
| query and drowns the specific memory the user actually wanted (measured on a 6k-note
|
| vault: such hubs sat in the top-10 for ~47% of queries). We mark them `status:'hub'`
|
| (reversible; recall skips them unless include_hubs) — measured to lift recall@5 ~+22%.
|
| 2. near-duplicate LINKING (dedup without delete) — EXCEPT a polarity clash, which is a
|
| STATE TOGGLE (preference flip): supersede the OLDER, since a contradiction is not a dup.
|
| 3. keep-budget: mark the lowest-value surplus `superseded`.
|
|
|
| hub_coverage: a memory covering ≥ this fraction of the common vocabulary is a hub (0 disables).
|
| link_duplicates: the dup pass is O(n²); pass False to skip it on large stores."""
|
| active = [r for r in self.items if r["status"] == "active"]
|
| hubs = 0
|
| if hub_coverage and len(active) >= 50:
|
| toks, common = self._common_vocab(active)
|
| nv = len(common) or 1
|
| for r, tk in zip(active, toks):
|
| shared = len(tk & common)
|
| cov = shared / nv
|
|
|
|
|
|
|
|
|
|
|
|
|
| if shared >= 3 and cov >= hub_coverage:
|
| r["status"] = "hub"
|
| r.setdefault("meta", {})["hub"] = True
|
| r["meta"]["hub_coverage"] = round(cov, 3)
|
| r["superseded_ts"] = time.time()
|
| hubs += 1
|
| active = [r for r in active if r["status"] == "active"]
|
| active.sort(key=lambda r: -r["value"])
|
| linked = toggled = 0
|
| if link_duplicates:
|
|
|
|
|
|
|
|
|
|
|
| for i, a in enumerate(active):
|
| if a["status"] != "active":
|
| continue
|
| avec = self._qvec(a["text"])
|
| for b in active[i + 1:]:
|
| if b["status"] != "active" or b["id"] in a["links"]:
|
| continue
|
| if self._similarity(a["text"], b, avec) >= dup_threshold:
|
| if _negation_clash(a["text"], b["text"]) or _value_clash(a["text"], b["text"]):
|
|
|
|
|
|
|
|
|
|
|
| _vf = lambda r: r.get("valid_from", r["ts"])
|
| older, newer = (a, b) if _vf(a) <= _vf(b) else (b, a)
|
|
|
|
|
|
|
|
|
| if self.supersede_requires_corroboration:
|
| _ng = float(newer.get("good", 0) or 0); _nb = float(newer.get("bad", 0) or 0)
|
| if not ((_ng > 0 and _ng >= _nb) or len(newer.get("links") or []) >= 2):
|
| a["links"].append(b["id"]); linked += 1
|
| continue
|
|
|
|
|
|
|
|
|
| if self.supersede_persistence > 1:
|
| nvec = self._qvec(newer["text"])
|
| support = sum(
|
| 1 for r in active if r["status"] == "active"
|
| and self._similarity(newer["text"], r, nvec) >= dup_threshold
|
| and not _value_clash(newer["text"], r["text"])
|
| and not _negation_clash(newer["text"], r["text"])
|
| and (_value_clash(older["text"], r["text"]) or _negation_clash(older["text"], r["text"])))
|
| if support < self.supersede_persistence:
|
| a["links"].append(b["id"]); linked += 1
|
| continue
|
| older["status"] = "superseded"
|
| older["superseded_ts"] = time.time()
|
| older["invalidated_at"] = _vf(newer)
|
| older.setdefault("meta", {})["superseded_by_toggle"] = newer["id"]
|
|
|
|
|
|
|
|
|
| older["bad"] = float(older.get("bad", 0) or 0) + 1.0
|
| newer["good"] = float(newer.get("good", 0) or 0) + 1.0
|
| toggled += 1
|
| if older is a:
|
| break
|
| else:
|
| a["links"].append(b["id"]); linked += 1
|
| staled = 0
|
| if keep is not None and len(active) > keep:
|
|
|
|
|
|
|
|
|
| if self.two_tier_keep:
|
| now = time.time()
|
| kprot = int(self.protect_frac * keep)
|
| protected, rest = active[:kprot], active[kprot:]
|
| rest_keep = set(id(r) for r in
|
| sorted(rest, key=lambda r: -self._effective_value(r, now))[:keep - kprot])
|
| drop = [r for r in rest if id(r) not in rest_keep]
|
| else:
|
| drop = active[keep:]
|
| for r in drop:
|
| r["status"] = "superseded"; r["superseded_ts"] = time.time(); staled += 1
|
| self._save()
|
| return {"active": len([r for r in self.items if r["status"] == "active"]),
|
| "hubs_flagged": hubs, "linked_pairs": linked, "toggled": toggled,
|
| "staled": staled, "kept": keep, "total": len(self.items)}
|
|
|
|
|
| def _cluster_active(self, sim_threshold: float = 0.5) -> list[list[dict]]:
|
| """Cheap greedy single-pass clustering of ACTIVE memories by similarity (O(n·#clusters)).
|
| Highest-value member is the cluster representative; each memory joins the most-similar
|
| cluster above the threshold, else starts its own. Lexical or semantic per the store's mode."""
|
| active = sorted([r for r in self.items if r["status"] == "active"], key=lambda r: -r["value"])
|
| cents: list[dict] = []
|
| for r in active:
|
| rvec = self._qvec(r["text"])
|
| best = None
|
| for c in cents:
|
| s = self._similarity(c["rec"]["text"], r, c["vec"])
|
| if s >= sim_threshold and (best is None or s > best[1]):
|
| best = (c, s)
|
| if best:
|
| best[0]["members"].append(r)
|
| else:
|
| cents.append({"rec": r, "vec": rvec, "members": [r]})
|
| return [c["members"] for c in cents]
|
|
|
| def consolidate_clusters(self, threshold: int = 15, cluster_sim: float = 0.5,
|
| dup_threshold: float = 0.82, keep_per_cluster: int | None = None) -> dict:
|
| """Cluster-TRIGGERED consolidation: consolidate a semantic cluster only once it has grown past
|
| `threshold` members — not a global nightly blanket. Avoids (1) prematurely consolidating sparse
|
| topics, where the raw episodes are still the best representation, and (2) unbounded growth in
|
| dense ones. Cheap to call often (no-op until a cluster is ripe). Runs dedup + the state-toggle
|
| guard (+ optional keep-budget) WITHIN each ripe cluster only."""
|
| clusters = self._cluster_active(cluster_sim)
|
| fired = linked = toggled = staled = 0
|
| for members in clusters:
|
| if len(members) < threshold:
|
| continue
|
| fired += 1
|
| members.sort(key=lambda r: -r["value"])
|
| for i, a in enumerate(members):
|
| if a["status"] != "active":
|
| continue
|
| avec = self._qvec(a["text"])
|
| for b in members[i + 1:]:
|
| if b["status"] != "active" or b["id"] in a["links"]:
|
| continue
|
| if self._similarity(a["text"], b, avec) >= dup_threshold:
|
| if _negation_clash(a["text"], b["text"]) or _value_clash(a["text"], b["text"]):
|
| older, newer = (a, b) if a["ts"] <= b["ts"] else (b, a)
|
| older["status"] = "superseded"; older["superseded_ts"] = time.time()
|
| older.setdefault("meta", {})["superseded_by_toggle"] = newer["id"]
|
| toggled += 1
|
| if older is a:
|
| break
|
| else:
|
| a["links"].append(b["id"]); linked += 1
|
| if keep_per_cluster is not None:
|
| act = sorted([r for r in members if r["status"] == "active"], key=lambda r: -r["value"])
|
| for r in act[keep_per_cluster:]:
|
| r["status"] = "superseded"; r["superseded_ts"] = time.time(); staled += 1
|
| self._save()
|
| return {"clusters_total": len(clusters), "clusters_fired": fired, "threshold": threshold,
|
| "linked_pairs": linked, "toggled": toggled, "staled": staled}
|
|
|
|
|
| def contradictions(self, sim_threshold: float = 0.5, incompatible=None) -> list[dict]:
|
| """Flag mutually-incompatible memories among RELATED ones (similarity-gated) for human review.
|
| `incompatible(a_text, b_text)->bool` defaults to a negation/polarity heuristic."""
|
| inc = incompatible or _negation_clash
|
| active = [r for r in self.items if r["status"] == "active"]
|
| flags = []
|
| for i, a in enumerate(active):
|
| avec = self._qvec(a["text"])
|
| for b in active[i + 1:]:
|
| if self._similarity(a["text"], b, avec) >= sim_threshold and inc(a["text"], b["text"]):
|
| flags.append({"a": a["id"], "b": b["id"],
|
| "a_text": a["text"][:120], "b_text": b["text"][:120]})
|
| return flags
|
|
|
|
|
| def value_by_cohort(self) -> dict:
|
| """Per-TAG value rollup. Deliberately not per-memory: at n-of-1, per-item value is noise;
|
| the cohort (tag / time-block) is where the signal is real."""
|
| out: dict[str, dict] = {}
|
| for r in self.items:
|
| if r["status"] != "active":
|
| continue
|
| for tag in (r["tags"] or ["(untagged)"]):
|
| c = out.setdefault(tag, {"count": 0, "value": 0.0})
|
| c["count"] += 1; c["value"] += r["value"]
|
| return {k: {"count": v["count"], "value": round(v["value"], 2),
|
| "avg": round(v["value"] / v["count"], 2)} for k, v in out.items()}
|
|
|
| def _save(self, force: bool = False):
|
| if not self.path:
|
| return
|
|
|
|
|
| now = time.time()
|
| if not force and (now - self._last_save) < self._save_min_s:
|
| self._dirty = True
|
| return
|
| try:
|
|
|
|
|
|
|
|
|
|
|
| slim = [{k: v for k, v in r.items() if k != "vec"} for r in self.items]
|
|
|
|
|
| data = json.dumps(slim, ensure_ascii=False, indent=1)
|
| tmp = self.path.with_name(self.path.name + ".tmp")
|
| tmp.write_text(data, encoding="utf-8")
|
| os.replace(tmp, self.path)
|
| self._last_save = now
|
| self._dirty = False
|
| except Exception:
|
| pass
|
|
|
| def flush(self):
|
| """Force-persist any pending throttled changes (call on clean shutdown)."""
|
| if self._dirty:
|
| self._save(force=True)
|
|
|
|
|
|
|
|
|
|
|
| _HALFLIFE_S = {"episodic": 7 * 86400, "semantic": 180 * 86400, "procedural": 3650 * 86400}
|
|
|
|
|
| _GRADUATE_VALUE = 5.0
|
|
|
|
|
|
|
|
|
| _PREFER_GAIN = 3.0
|
| _PROCEDURAL_RE = re.compile(r"\b(always|never|prefers?|rule|workflow|convention|policy|habit|"
|
| r"setting|must|should|avoid|don't|do not)\b", re.I)
|
| _SEMANTIC_RE = re.compile(r"\b(means|defined|definition|theorem|law of|equals|consists? of|"
|
| r"is a |is an |is the |refers to)\b", re.I)
|
|
|
|
|
| def _infer_type(text: str) -> str:
|
| """Conservative type inference: default EPISODIC (fast decay) and only promote on clear markers.
|
| Callers that know the kind should pass mtype explicitly."""
|
| t = text or ""
|
| if _PROCEDURAL_RE.search(t):
|
| return "procedural"
|
| if _SEMANTIC_RE.search(t):
|
| return "semantic"
|
| return "episodic"
|
|
|
|
|
| def _negation_clash(a: str, b: str) -> bool:
|
| """Cheap default: two highly-related statements where exactly one negates. Replace with an
|
| LLM judge for production — but gate it behind similarity first to keep it O(neighbourhood)."""
|
| neg = re.compile(r"\b(not|no|never|cannot|can't|doesn't|isn't|won't|fails?|false)\b", re.I)
|
| return bool(neg.search(a)) != bool(neg.search(b))
|
|
|
|
|
| _NUM = re.compile(r"-?\d+(?:\.\d+)?")
|
|
|
|
|
| def _value_clash(a: str, b: str) -> bool:
|
| """A VALUE UPDATE: two already-near-duplicate statements that are identical EXCEPT for a differing
|
| numeric value ('retry limit is 5' -> '... is 12'). This is a state toggle (the fact's value changed),
|
| NOT a duplicate — so the older should be superseded, not merged. Gated behind the caller's similarity
|
| check; the tight 'non-numeric remainder is identical' condition keeps genuinely-distinct facts safe."""
|
|
|
|
|
|
|
|
|
|
|
|
|
| na, nb = _NUM.findall(a), _NUM.findall(b)
|
| if not na or len(na) != len(nb):
|
| return False
|
| if sum(1 for x, y in zip(na, nb) if x != y) != 1:
|
| return False
|
|
|
|
|
|
|
| return _tokens(_NUM.sub("", a)) == _tokens(_NUM.sub("", b))
|
|
|
|
|
| if __name__ == "__main__":
|
| m = Mnemo()
|
| m.remember("SGD converges slowly due to gradient variance.", tags=["optimization"], value=3)
|
| m.remember("SGD does not converge slowly.", tags=["optimization"], value=1)
|
| m.remember("Pre-trend tests catch only 31% of fatal DiD bias.", tags=["causal"], value=2)
|
| print("recall 'SGD variance':", [r["text"][:46] for r in m.recall("SGD variance", k=3)])
|
| print("consolidate:", m.consolidate(keep=10))
|
| print("contradictions:", m.contradictions())
|
| print("value_by_cohort:", m.value_by_cohort())
|
| print("(For semantic recall, pass embed=your_model to Mnemo(); lexical is the zero-dep fallback.)")
|
|
|
