Add optional embedding reranker

CHANGELOG.md

@@ -10,6 +10,15 @@
   and the exact command.
 - CI now smoke-tests the LongMemEval harness against a tiny local fixture,
   keeping the public dataset download out of the test matrix.
+- Added an optional `MIND_EMBED_CMD` recall re-ranker. The command reads text
+  from stdin and returns a numeric vector as JSON or whitespace-separated
+  floats; failures, timeouts, invalid vectors, and unset commands fall back to
+  the deterministic hash embedder without mixing vector spaces. Transient
+  failures are retried after a short cache window, and oversized output stays
+  off the Python heap.
+- Benchmarks now report the default offline recall column and, when
+  `MIND_EMBED_CMD` is set, a second embedded re-rank column so semantic
+  backend gains can be measured without changing the offline baseline.
 
 ## 6.2.10 — 2026-07-12
 

README.md

@@ -39,8 +39,9 @@ No server, no vector store, no embedding model, no configuration file.
 
 ## Measured, not vibes
 
-The first table comes from `python3 bench/bench.py`; each later measurement
-names its own reproducible command — rerun them yourself
+The first table comes from `python3 bench/bench.py`; when `MIND_EMBED_CMD`
+is configured it also prints an embedded re-rank column next to the offline
+score. Each later measurement names its own reproducible command — rerun them yourself
 (Python 3.14 on Apple M-series — latencies are environment-dependent, rerun
 for your hardware; 20 bilingual queries with known answers against
 distractor-filled graphs):
@@ -142,8 +143,8 @@ Layer 1  WORKING MEMORY   .mind/ACTIVE.md  → injected into agent rule files
 Layer 2  HIPPOCAMPUS      .mind/graph.json → weighted concept graph
          recall = spreading activation (≤3 hops) fused with direct keyword
          matches via Reciprocal Rank Fusion + IDF, re-ranked by offline
-         hash embeddings; near-duplicate results are diversified (pattern
-         separation); fuzzy fallback finds memories from partial cues
+         hash embeddings or optional `MIND_EMBED_CMD`; near-duplicate
+         results are diversified (pattern separation); fuzzy fallback finds memories from partial cues
          (pattern completion)
 
 Layer 3  CORTEX           .mind/cortex/*.md → consolidated durable knowledge
@@ -246,6 +247,29 @@ and restrengthens its edges. The exported agent instructions teach this
 loop, and every dream weakens all edges slightly (synaptic homeostasis),
 so connections that never earn a confirmation decay and prune away.
 
+## Optional semantic re-ranking
+
+The default recall path remains fully offline and stdlib-only. To compare a
+learned embedding backend without changing storage or the graph algorithm,
+set `MIND_EMBED_CMD` to a trusted local command that reads text on stdin and
+prints either a JSON list of floats or whitespace-separated floats. The
+command is parsed into arguments and executed directly, never through a shell:
+
+```bash
+export MIND_EMBED_CMD='python3 local_embed.py'
+python3 mind.py recall "what css framework"
+python3 bench/bench.py    # prints offline vs embedded re-rank columns
+```
+
+The hook is used only for recall head re-ranking. If the command is missing,
+times out, exits non-zero, or returns an invalid vector, `mind` silently falls
+back to the built-in hash embeddings for the whole comparison, so vectors from
+different embedding spaces are never mixed. Failed texts are retried after a
+short cache window. Set `MIND_EMBED_TIMEOUT` to control the per-text timeout
+in seconds (default `2`, clamped to `0.1`–`30`). Dream clustering, fuzzy
+fallback, and pattern separation continue to use the deterministic offline
+embedder.
+
 ## Safety properties
 
 - **Atomic, durable, symlink-refusing writes** everywhere — directory-handle
@@ -273,9 +297,10 @@ so connections that never earn a confirmation decay and prune away.
   sentry→errors...). Cross-domain synonymy *outside* that seed and the
   corpus still misses; polysemous words (black, express, spring...) are
   deliberately excluded from the seed because a false category on an
-  everyday sentence is worse than a missed synonym. True embeddings would
-  close the remainder at the cost of the zero-dependency promise; a
-  pluggable backend is on the roadmap.
+  everyday sentence is worse than a missed synonym. `MIND_EMBED_CMD` can
+  re-rank the recall head with a learned backend for those cases, but
+  quality then depends on the command you provide; the built-in fallback
+  remains deterministic and local.
 - Arabic stemming is light (prefix/suffix + broken-plural seed), not a full
   morphological analyzer. Other languages get no stemming or stopword
   lists at all — IDF and character n-grams carry them (measured above),

bench/bench.py

@@ -11,6 +11,7 @@
 Numbers in README.md come from this script — re-run it yourself.
 """
 import random
+import os
 import shutil
 import statistics
 import sys
@@ -62,6 +63,25 @@ def build(n_distractors):
     return tmp, h
 
 
+class embed_env:
+    def __init__(self, cmd):
+        self.cmd = cmd
+        self.old = None
+
+    def __enter__(self):
+        self.old = os.environ.get("MIND_EMBED_CMD")
+        if self.cmd:
+            os.environ["MIND_EMBED_CMD"] = self.cmd
+        else:
+            os.environ.pop("MIND_EMBED_CMD", None)
+
+    def __exit__(self, *_):
+        if self.old is None:
+            os.environ.pop("MIND_EMBED_CMD", None)
+        else:
+            os.environ["MIND_EMBED_CMD"] = self.old
+
+
 def run_recall(h):
     hits1 = hits5 = 0
     lat = []
@@ -77,6 +97,15 @@ def run_recall(h):
     return hits1 / len(FACTS), hits5 / len(FACTS), lat
 
 
+def measure(n_distractors, embed_cmd=None):
+    with embed_env(embed_cmd):
+        tmp, h = build(n_distractors)
+        try:
+            return run_recall(h)
+        finally:
+            shutil.rmtree(tmp, ignore_errors=True)
+
+
 def dream_determinism():
     t1, h1 = build(30)
     t2, h2 = build(30)
@@ -94,17 +123,25 @@ def dream_determinism():
 def main():
     print("mind benchmark (python %s)" % sys.version.split()[0])
     print("=" * 56)
+    embed_cmd = os.environ.get("MIND_EMBED_CMD")
+    if embed_cmd:
+        print("embedded rerank: enabled through MIND_EMBED_CMD")
+    else:
+        print("embedded rerank: n/a (set MIND_EMBED_CMD to compare)")
     ok = True
     for n_distract, label in ((80, "100 nodes"), (980, "1,000 nodes")):
-        tmp, h = build(n_distract)
-        r1, r5, lat = run_recall(h)
-        print("%-12s recall@1 %.2f | recall@5 %.2f | "
-              "median %.2f ms | p95 %.2f ms"
+        r1, r5, lat = measure(n_distract, embed_cmd=None)
+        if embed_cmd:
+            er1, er5, elat = measure(n_distract, embed_cmd=embed_cmd)
+            embedded = " | embedded r@1 %.2f r@5 %.2f med %.2f ms" % (
+                er1, er5, statistics.median(elat))
+        else:
+            embedded = " | embedded n/a"
+        print("%-12s offline r@1 %.2f r@5 %.2f med %.2f ms p95 %.2f ms%s"
               % (label, r1, r5, statistics.median(lat),
-                 sorted(lat)[int(len(lat) * 0.95) - 1]))
+                 sorted(lat)[int(len(lat) * 0.95) - 1], embedded))
         if r1 < 0.9:            # regression gate for CI
             ok = False
-        shutil.rmtree(tmp, ignore_errors=True)
     det = dream_determinism()
     print("dream determinism: %s"
           % ("PASS (identical plan on identical state)" if det else "FAIL"))

mind.py

@@ -25,7 +25,7 @@
 
 Design: docs/DESIGN.md  |  License: MIT  |  https://github.com/Da7-Tech/mind
 """
-import sys, os, json, re, time, math, hashlib, tempfile, shlex, stat, threading
+import sys, os, json, re, time, math, hashlib, tempfile, shlex, stat, threading, subprocess
 from datetime import datetime, timedelta
 from pathlib import Path
 from collections import Counter, defaultdict, deque
@@ -1075,6 +1075,151 @@ def similarity(self, a, b):
         return max(0.0, dot / (na * nb))
 
 
+class CommandEmbed:
+    """Optional command-backed embeddings for recall re-ranking.
+
+    The command is read from MIND_EMBED_CMD by default. It receives the text
+    on stdin and should print either a JSON list of numbers or whitespace/
+    comma-separated floats. Any failure falls back to HashEmbed, so default
+    offline behaviour and zero-dependency installs stay unchanged.
+    """
+
+    MAX_OUTPUT_BYTES = 1_000_000
+    FAILURE_CACHE_SECONDS = 5.0
+
+    def __init__(self, cmd=None, fallback=None, timeout=None):
+        raw_cmd = cmd if cmd is not None else os.environ.get("MIND_EMBED_CMD", "")
+        self.cmd = str(raw_cmd or "").strip()
+        self.fallback = fallback if fallback is not None else HashEmbed()
+        self.timeout = self._timeout(timeout)
+        self._cache = {}
+
+    @staticmethod
+    def _timeout(value):
+        if value is None:
+            value = os.environ.get("MIND_EMBED_TIMEOUT", "2.0")
+        try:
+            value = float(value)
+        except (TypeError, ValueError):
+            return 2.0
+        return max(0.1, min(30.0, value))
+
+    @staticmethod
+    def _parse_vector(payload):
+        text = (payload or b"").decode("utf-8", "replace").strip()
+        if not text:
+            return None
+        try:
+            data = json.loads(text)
+            if isinstance(data, dict):
+                for key in ("embedding", "vector", "values"):
+                    if key in data:
+                        data = data[key]
+                        break
+            if isinstance(data, list):
+                vec = [float(v) for v in data]
+                return (
+                    vec if vec and any(vec)
+                    and all(math.isfinite(v) for v in vec) else None
+                )
+        except (TypeError, ValueError, json.JSONDecodeError):
+            pass
+        parts = [p for p in re.split(r"[\s,]+", text) if p]
+        try:
+            vec = [float(p) for p in parts]
+        except ValueError:
+            return None
+        return (
+            vec if vec and any(vec)
+            and all(math.isfinite(v) for v in vec) else None
+        )
+
+    @staticmethod
+    def _split_command(cmd, platform=None):
+        try:
+            parts = shlex.split(cmd, posix=(platform or os.name) != "nt")
+        except ValueError:
+            return None
+        if (platform or os.name) == "nt":
+            parts = [
+                p[1:-1] if len(p) >= 2 and p[0] == p[-1] and p[0] in ("'", '"') else p
+                for p in parts
+            ]
+        return parts
+
+    def _command_embed(self, text):
+        if not self.cmd:
+            return None
+        argv = self._split_command(self.cmd)
+        if not argv:
+            return None
+        try:
+            # Keep unexpectedly large command output off the Python heap. The
+            # command is user-configured and trusted, but accidental output
+            # floods should still degrade to HashEmbed instead of exhausting
+            # agent memory.
+            with tempfile.TemporaryFile() as stdout:
+                proc = subprocess.run(
+                    argv,
+                    input=(text or "").encode("utf-8"),
+                    stdout=stdout,
+                    stderr=subprocess.DEVNULL,
+                    timeout=self.timeout,
+                    check=False,
+                )
+                size = stdout.tell()
+                if proc.returncode != 0 or size > self.MAX_OUTPUT_BYTES:
+                    return None
+                stdout.seek(0)
+                payload = stdout.read(self.MAX_OUTPUT_BYTES + 1)
+        except (OSError, subprocess.SubprocessError):
+            return None
+        return self._parse_vector(payload)
+
+    def _embed_with_source(self, text):
+        text = text or ""
+        if not self.cmd:
+            return "fallback", self.fallback.embed(text)
+
+        now = time.monotonic()
+        cached = self._cache.get(text)
+        if cached is not None:
+            source, vec, retry_at = cached
+            if source == "command" or now < retry_at:
+                return source, vec
+
+        vec = self._command_embed(text)
+        if vec is None:
+            source = "fallback"
+            vec = self.fallback.embed(text)
+            retry_at = now + self.FAILURE_CACHE_SECONDS
+        else:
+            source = "command"
+            retry_at = float("inf")
+        if text in self._cache or len(self._cache) < 4096:
+            self._cache[text] = (source, vec, retry_at)
+        return source, vec
+
+    def embed(self, text):
+        return self._embed_with_source(text)[1]
+
+    def similarity(self, a, b):
+        source_a, va = self._embed_with_source(a)
+        source_b, vb = self._embed_with_source(b)
+        if (
+            source_a != "command"
+            or source_b != "command"
+            or not va
+            or not vb
+            or len(va) != len(vb)
+        ):
+            return self.fallback.similarity(a, b)
+        dot = sum(x * y for x, y in zip(va, vb))
+        na = math.sqrt(sum(x * x for x in va)) or 1.0
+        nb = math.sqrt(sum(y * y for y in vb)) or 1.0
+        return max(0.0, dot / (na * nb))
+
+
 # ────────────────────────────────────────────────────────────────
 # Layer 2: Hippocampus — the weighted concept graph
 # ────────────────────────────────────────────────────────────────
@@ -1091,6 +1236,7 @@ def __init__(self, path):
         self.meta = {}    # small persisted strings (e.g. last_edge_decay)
         self.related = None
         self.embedder = HashEmbed()
+        self.reranker = CommandEmbed(fallback=self.embedder)
         self._thread_lock = threading.RLock()
         self._transaction_state = threading.local()
         self._deleted = set()
@@ -2143,7 +2289,7 @@ def recall(self, query, top_k=RECALL_TOP_K, max_hops=RECALL_RADIUS,
         if len(ranked) > 1 and not identity_q:
             reranked = []
             for nid, base in ranked[:top_k * 3]:
-                sim = self.embedder.similarity(query, self.nodes[nid]["text"])
+                sim = self.reranker.similarity(query, self.nodes[nid]["text"])
                 reranked.append((nid, base * (1.0 + sim)))
             reranked.sort(key=lambda x: (-x[1], x[0]))
             ranked = reranked