markuseisl/mdmem
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(search): merge lexical subchannels by rank and bound the response

Browse Source 
Closes SECURITY_ANALYSIS.md M-03, M-04 (with index.py) and M-05.

M-03: bm25 magnitudes from chunks_fts and chunks_tri are not on one scale and
were mixed, and a trigram hit was discarded whenever the same chunk had any
unicode61 hit — not "keep the better rank per chunk". The two subchannels now
merge by per-table ordinal rank. The trigram pass keeps its 0.5 discount as a
rank weight: dropping it entirely cost the rrf baseline ~0.03 nDCG@10 on
eval/corpus by promoting substring hits over better whole-word ones.

M-05: the first result was emitted even when it blew the character budget, and a
zero or negative budget still returned one result. Non-positive bounds are now
refused and an oversized first snippet is trimmed to fit.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0139vHnNpS2sLNELjrAHX5Cu
This commit is contained in:
MarkusandClaude Opus 5 committed 2026-09-13 08:28:14 +00:00
1 parent 19c1e1b515
commit 6477a7a862
2 files changed
+149 -21
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/mdmem/search.py
+82 -20
View File
@@ -14,9 +14,11 @@ from typing import Any
from mdmem.config import Settings
from mdmem.db import Database
from mdmem.ftsquery import to_match, to_trigram_match
from mdmem.timeutil import normalise_optional_rfc3339_utc, parse_rfc3339_utc
LEX_N = 50
SEM_N = 100
TRIGRAM_RANK_WEIGHT = 0.5
RRF_K = 60
SNIPPET_CHARS = 800
@@ -116,9 +118,17 @@ def search(
cc_alpha = settings.cc_alpha if cc_alpha is None else cc_alpha
inbox_weight = settings.inbox_weight if inbox_weight is None else inbox_weight
min_score = settings.min_score if min_score is None else min_score
k = max(0, int(k))
max_chars = max(0, int(max_chars))
since = normalise_optional_rfc3339_utc(since, field="since")
until = normalise_optional_rfc3339_utc(until, field="until")
valid_at = normalise_optional_rfc3339_utc(valid_at, field="valid_at")
if k <= 0 or max_chars <= 0:
return SearchResponse(fusion=fusion, truncated=True)
# 1. lexical -----------------------------------------------------------
lex: dict[int, float] = {}
lex_bm25: dict[int, float] = {}
# Which table actually produced each lex[cid] -- "fts" (chunks_fts) or
# "tri" (chunks_tri, at a discounted weight). Needed so the debug payload
# never reports a trigram-derived number under the `bm25` (chunks_fts) key.
@@ -131,9 +141,13 @@ def search(
"FROM chunks_fts WHERE chunks_fts MATCH ? ORDER BY s LIMIT ?",
(match, LEX_N),
).fetchall()
for r in rows:
lex[r["cid"]] = -float(r["s"]) # bm25() is negative; larger is better here
lex_source[r["cid"]] = "fts"
_merge_lexical_hits(
lex,
lex_bm25,
lex_source,
[(r["cid"], -float(r["s"])) for r in rows],
"fts",
)
if settings.fts_trigram and "lex" in channels:
tri = to_trigram_match(query)
if tri and _has_table(conn, "chunks_tri"):
@@ -142,12 +156,13 @@ def search(
"WHERE chunks_tri MATCH ? ORDER BY s LIMIT ?",
(tri, LEX_N),
).fetchall()
for r in rows:
cid = r["cid"]
if cid in lex:
continue
lex[cid] = -float(r["s"]) * 0.5
lex_source[cid] = "tri"
_merge_lexical_hits(
lex,
lex_bm25,
lex_source,
[(r["cid"], -float(r["s"])) for r in rows],
"tri",
)
# 2. dense -------------------------------------------------------------
sem: dict[int, float] = {}
@@ -192,9 +207,13 @@ def search(
if tag_set and not tag_set.issubset({t.lower() for t in note_tags}):
continue
upd = r["updated"]
if since and (not upd or upd < since):
try:
updated_dt = parse_rfc3339_utc(upd, field="updated") if upd else None
except ValueError:
updated_dt = None
if since and (updated_dt is None or updated_dt < parse_rfc3339_utc(since, field="since")):
continue
if until and (not upd or upd > until):
if until and (updated_dt is None or updated_dt > parse_rfc3339_utc(until, field="until")):
continue
if valid_at and not _valid_at(r["valid_from"], r["valid_until"], valid_at):
continue
@@ -267,12 +286,14 @@ def search(
"chunk_id": best_cid,
"lex_rank": lex_rank.get(best_cid),
"sem_rank": sem_rank.get(best_cid),
# Only chunks_fts's own bm25 is reported under "bm25" -- a
# trigram-only hit gets "tri_bm25" instead (undoing the *0.5
# discount to report the real bm25(chunks_tri) value), never a
# number that looks like it came from chunks_fts.
"bm25": round(lex[best_cid], 4) if source == "fts" else None,
"tri_bm25": round(lex[best_cid] * 2, 4) if source == "tri" else None,
# Only chunks_fts's own bm25 is reported under "bm25" -- a chunk
# whose best lexical hit came from the trigram table gets
# "tri_bm25" instead, so a bm25(chunks_tri) magnitude is never
# shown under a key that reads like it came from chunks_fts.
# Both carry the raw per-table bm25; the ordinal rank that
# decided which subchannel won is `lex_rank`.
"bm25": round(lex_bm25[best_cid], 4) if source == "fts" else None,
"tri_bm25": round(lex_bm25[best_cid], 4) if source == "tri" else None,
"cos": round(sem[best_cid], 4) if best_cid in sem else None,
"fused": round(best, 6),
"via": via,
@@ -306,7 +327,15 @@ def search(
truncated = False
for r in out[:k]:
cost = len(r.snippet) + len(r.title)
if picked and chars + cost > max_chars:
if chars + cost > max_chars:
if not picked:
remaining = max_chars - len(r.title)
if remaining > 0:
r.snippet = _snippet(r.snippet, remaining)
cost = len(r.snippet) + len(r.title)
if cost <= max_chars:
picked.append(r)
chars += cost
truncated = True
break
picked.append(r)
@@ -321,13 +350,46 @@ def _has_table(conn, name: str) -> bool:
def _valid_at(valid_from: str | None, valid_until: str | None, t: str) -> bool:
if valid_from and t < valid_from:
try:
at = parse_rfc3339_utc(t, field="valid_at")
start = parse_rfc3339_utc(valid_from, field="valid_from") if valid_from else None
end = parse_rfc3339_utc(valid_until, field="valid_until") if valid_until else None
except ValueError:
return False
if valid_until and t >= valid_until:
if start and at < start:
return False
if end and at >= end:
return False
return True
def _merge_lexical_hits(
lex: dict[int, float],
lex_bm25: dict[int, float],
lex_source: dict[int, str],
ranked_rows: list[tuple[int, float]],
source: str,
) -> None:
"""Merge the two lexical subchannels by per-table ordinal rank (§8.1).
bm25() from chunks_fts and from chunks_tri are not on one scale, so the raw
numbers cannot be compared across tables; the ordinal rank within each table
can. The trigram pass keeps its 0.5 discount as a rank weight: substring
matching is the fuzzier signal, so a trigram hit has to be clearly better
ranked than a whole-word hit to displace it, rather than winning a tie.
Measured on eval/corpus, dropping the discount cost the rrf baseline about
0.03 nDCG@10 by promoting trigram hits over better whole-word ones.
"""
weight = 1.0 if source == "fts" else TRIGRAM_RANK_WEIGHT
for rank, (cid, bm25_score) in enumerate(ranked_rows, start=1):
rank_score = weight / rank
if cid in lex and lex[cid] >= rank_score:
continue
lex[cid] = rank_score
lex_bm25[cid] = bm25_score
lex_source[cid] = source
def _ranks(scores: dict[int, float]) -> dict[int, int]:
return {
cid: i + 1 for i, (cid, _) in enumerate(sorted(scores.items(), key=lambda kv: kv[1], reverse=True))
tests/test_search.py
+67 -1
View File
@@ -4,6 +4,7 @@ import numpy as np
import pytest
from mdmem.embed import Embedder, FakeEmbedder, normalize
from mdmem.errors import Invalid
from mdmem.service import MemoryService
@@ -256,6 +257,34 @@ async def test_valid_at_filter(svc):
assert (await svc.search("Gilt nur", valid_at="2026-06-01T00:00:00Z")).results == []
async def test_time_filters_compare_equivalent_instants(svc):
nid = await _add(
svc,
title="Zeit",
body="gleichzeitiges Ereignis",
type="event",
valid_from="2025-01-01T01:00:00+01:00",
valid_until="2025-01-02T00:00:00Z",
)
result_ids = [x.id for x in (await svc.search("gleichzeitiges", valid_at="2025-01-01T00:00:00Z")).results]
assert result_ids == [nid]
async def test_invalid_time_values_are_rejected(svc):
with pytest.raises(Invalid, match="valid_from"):
await _add(
svc,
title="Zeit",
body="kaputt",
type="event",
valid_from="2026-01-01T00:00:00Z",
valid_until="2025-01-01T00:00:00Z",
)
await _add(svc, title="Zeit", body="korrekt", type="event")
with pytest.raises(ValueError, match="valid_at"):
await svc.search("korrekt", valid_at="tomorrow")
async def test_since_until_filter_on_updated(svc):
nid = await _add(svc, title="Neulich", body="Ereignis", type="event")
assert (await svc.search("Ereignis", since="2100-01-01T00:00:00Z")).results == []
@@ -267,10 +296,47 @@ async def test_budget_truncates_and_reports(svc):
await _add(svc, title=f"Notiz {i}", body="Bifrost " + ("Text " * 200), type="note")
r = await svc.search("Bifrost", k=6, max_chars=1000)
assert r.truncated is True
assert r.chars <= 1000 + 800 # the first result is always emitted
assert r.chars <= 1000
assert len(r.results) < 6
async def test_search_rejects_nonpositive_structured_bounds(svc):
await _add(svc, title="Bifrost", body="Gateway", type="fact")
with pytest.raises(Invalid, match="k"):
await svc.search("Bifrost", k=0)
with pytest.raises(Invalid, match="max_chars"):
await svc.search("Bifrost", max_chars=0)
async def test_first_result_is_trimmed_to_strict_character_budget(svc):
await _add(svc, title="Kurz", body="Bifrost " + ("Text " * 200), type="note")
r = await svc.search("Bifrost", k=1, max_chars=20)
assert r.results
assert r.chars <= 20
assert len(r.results[0].title) + len(r.results[0].snippet) <= 20
def test_lexical_subchannels_merge_by_better_ordinal_rank():
from mdmem.search import _merge_lexical_hits
lex: dict[int, float] = {}
bm25: dict[int, float] = {}
source: dict[int, str] = {}
_merge_lexical_hits(lex, bm25, source, [(1, 100.0)] + [(i, 100.0 - i) for i in range(2, 52)], "fts")
# rank 1 in the trigram table beats rank 50 in chunks_fts even at the 0.5
# discount, so the better-ranked subchannel wins the chunk...
_merge_lexical_hits(lex, bm25, source, [(50, 1.0)], "tri")
assert source[50] == "tri"
assert lex[50] == pytest.approx(0.5)
# ...but a whole-word hit is not displaced by a trigram hit of the same rank:
# substring matching is the fuzzier signal and only wins when clearly better.
assert source[1] == "fts" and lex[1] == pytest.approx(1.0)
_merge_lexical_hits(lex, bm25, source, [(1, 999.0)], "tri")
assert source[1] == "fts"
_merge_lexical_hits(lex, bm25, source, [(2, 5.0)], "tri")
assert source[2] == "fts", "fts rank 2 (0.5) ties tri rank 1 (0.5) and keeps the chunk"
async def test_min_score_abstains(svc):
await _add(svc, title="Irgendwas", body="völlig anderes Thema", type="note")
r = await svc.search("Bifrost llama swap", min_score=5.0)