feat(embeddings): hybrid fusion strategies (RRF, weighted, normalized)

src/vouch/embeddings/__init__.py

@@ -8,7 +8,7 @@
 modules are only imported when an embedding code path executes.
 """
 
-import contextlib
+from contextlib import suppress
 
 from .base import (
     DEFAULT_MODEL_NAME,
@@ -21,13 +21,13 @@
 # Auto-register the default adapter if sentence-transformers is installed.
 # Each adapter import is best-effort -- failure means that adapter's optional
 # dependency isn't installed, which is fine (a different adapter may be).
-with contextlib.suppress(ImportError):
+with suppress(ImportError):
     from . import st_mpnet  # noqa: F401
 
-with contextlib.suppress(ImportError):
+with suppress(ImportError):
     from . import st_minilm  # noqa: F401
 
-with contextlib.suppress(ImportError):
+with suppress(ImportError):
     from . import fastembed_bge  # noqa: F401
 
 __all__ = [

src/vouch/embeddings/cache.py

@@ -0,0 +1,83 @@
+"""Query embedding LRU cache backed by `query_embedding_cache` table."""
+
+from __future__ import annotations
+
+import datetime as dt
+import hashlib
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from .. import index_db
+
+
+def _query_key(query: str) -> str:
+    return hashlib.sha256(query.encode("utf-8")).hexdigest()
+
+
+def cache_query_vec(
+    kb_dir: Path, *, query: str, vec: np.ndarray, max_entries: int = 1024,
+) -> None:
+    h = _query_key(query)
+    blob = np.asarray(vec, dtype=np.float32).tobytes()
+    now = dt.datetime.now(dt.UTC).isoformat(timespec="seconds")
+    with index_db.open_db(kb_dir) as conn:
+        conn.execute(
+            "INSERT OR REPLACE INTO query_embedding_cache "
+            "(query_hash, vec, hit_count, last_used_at) "
+            "VALUES (?, ?, COALESCE("
+            "  (SELECT hit_count FROM query_embedding_cache WHERE query_hash=?), 0"
+            ") + 1, ?)",
+            (h, blob, h, now),
+        )
+        n = conn.execute(
+            "SELECT COUNT(*) FROM query_embedding_cache"
+        ).fetchone()[0]
+        if n > max_entries:
+            conn.execute(
+                "DELETE FROM query_embedding_cache WHERE query_hash IN ("
+                " SELECT query_hash FROM query_embedding_cache "
+                " ORDER BY last_used_at ASC LIMIT ?)",
+                (n - max_entries,),
+            )
+
+
+def lookup_query_vec(kb_dir: Path, *, query: str) -> np.ndarray | None:
+    h = _query_key(query)
+    now = dt.datetime.now(dt.UTC).isoformat(timespec="seconds")
+    with index_db.open_db(kb_dir) as conn:
+        row = conn.execute(
+            "SELECT vec FROM query_embedding_cache WHERE query_hash = ?",
+            (h,),
+        ).fetchone()
+        if not row:
+            return None
+        conn.execute(
+            "UPDATE query_embedding_cache SET hit_count = hit_count + 1, "
+            "last_used_at = ? WHERE query_hash = ?",
+            (now, h),
+        )
+    return np.frombuffer(row[0], dtype=np.float32).copy()
+
+
+def query_cache_size(kb_dir: Path) -> int:
+    with index_db.open_db(kb_dir) as conn:
+        return int(conn.execute(
+            "SELECT COUNT(*) FROM query_embedding_cache"
+        ).fetchone()[0])
+
+
+def query_cache_clear(kb_dir: Path) -> None:
+    with index_db.open_db(kb_dir) as conn:
+        conn.execute("DELETE FROM query_embedding_cache")
+
+
+def query_cache_stats(kb_dir: Path) -> dict[str, Any]:
+    with index_db.open_db(kb_dir) as conn:
+        row = conn.execute(
+            "SELECT COUNT(*), COALESCE(SUM(hit_count), 0), "
+            "COALESCE(MAX(hit_count), 0) FROM query_embedding_cache"
+        ).fetchone()
+    return {"entries": int(row[0]), "hits": int(row[1]),
+            "max_hits_per_entry": int(row[2])}

src/vouch/embeddings/fusion.py

@@ -0,0 +1,86 @@
+"""Result-list fusion strategies for hybrid retrieval."""
+
+from __future__ import annotations
+
+Hit = tuple[str, str, str, float]
+Hits = list[Hit]
+
+
+def _key(h: Hit) -> tuple[str, str]:
+    return (h[0], h[1])
+
+
+def _coalesce_snippet(h: Hit, other: Hits) -> str:
+    if h[2]:
+        return h[2]
+    for o in other:
+        if _key(o) == _key(h) and o[2]:
+            return o[2]
+    return ""
+
+
+def rrf_fuse(a: Hits, b: Hits, *, limit: int = 10, k: int = 60) -> Hits:
+    """Reciprocal Rank Fusion: score = sum(1 / (k + rank)) across lists."""
+    if limit < 0:
+        raise ValueError("limit must be >= 0")
+    if k < 0:
+        # rank starts at 1, so k must be > -1, but disallow negative
+        # entirely to match the canonical RRF definition (Cormack et al.)
+        # and to avoid division by zero when k == -rank.
+        raise ValueError("k must be >= 0")
+    scores: dict[tuple[str, str], float] = {}
+    for lst in (a, b):
+        for rank, h in enumerate(lst, start=1):
+            scores[_key(h)] = scores.get(_key(h), 0.0) + 1.0 / (k + rank)
+    out: Hits = []
+    seen: set[tuple[str, str]] = set()
+    for h in a + b:
+        if _key(h) in seen:
+            continue
+        seen.add(_key(h))
+        out.append((h[0], h[1], _coalesce_snippet(h, a + b), scores[_key(h)]))
+    out.sort(key=lambda h: h[3], reverse=True)
+    return out[:limit]
+
+
+def _minmax(xs: list[float]) -> list[float]:
+    if not xs:
+        return xs
+    lo, hi = min(xs), max(xs)
+    if hi - lo < 1e-9:
+        return [0.5] * len(xs)
+    return [(x - lo) / (hi - lo) for x in xs]
+
+
+def weighted_fuse(
+    a: Hits,
+    b: Hits,
+    *,
+    w_a: float = 0.5,
+    w_b: float = 0.5,
+    limit: int = 10,
+) -> Hits:
+    """Min-max normalize each list, then score = w_a * a_score + w_b * b_score."""
+    if limit < 0:
+        raise ValueError("limit must be >= 0")
+    a_norm = _minmax([h[3] for h in a])
+    b_norm = _minmax([h[3] for h in b])
+    a_score = {_key(h): a_norm[i] for i, h in enumerate(a)}
+    b_score = {_key(h): b_norm[i] for i, h in enumerate(b)}
+    merged: dict[tuple[str, str], float] = {}
+    for key in {*a_score, *b_score}:
+        merged[key] = w_a * a_score.get(key, 0.0) + w_b * b_score.get(key, 0.0)
+    out: Hits = []
+    seen: set[tuple[str, str]] = set()
+    for h in a + b:
+        if _key(h) in seen:
+            continue
+        seen.add(_key(h))
+        out.append((h[0], h[1], _coalesce_snippet(h, a + b), merged[_key(h)]))
+    out.sort(key=lambda h: h[3], reverse=True)
+    return out[:limit]
+
+
+def normalized_fuse(a: Hits, b: Hits, *, limit: int = 10) -> Hits:
+    """Equal-weight weighted_fuse (0.5/0.5)."""
+    return weighted_fuse(a, b, w_a=0.5, w_b=0.5, limit=limit)