feat(embeddings): rerank, HyDE, dedup, scorer, model-identity migration

src/vouch/embeddings/dedup.py

@@ -0,0 +1,90 @@
+"""Ingest-time duplicate detection via embedding cosine similarity."""
+
+from __future__ import annotations
+
+import datetime as dt
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from .. import index_db
+
+DEFAULT_THRESHOLD = 0.95
+
+
+def check_and_log(
+    kb_dir: Path, *,
+    kind: str, id: str, vec: np.ndarray,
+    threshold: float = DEFAULT_THRESHOLD,
+) -> tuple[str, float] | None:
+    """Find nearest-neighbor cosine >= threshold; log if found."""
+    candidates = index_db.search_embedding(
+        kb_dir, query_vec=vec, kinds=(kind,), limit=2,
+    )
+    for k, near_id, _snip, cos in candidates:
+        if k != kind or near_id == id:
+            continue
+        if cos >= threshold:
+            _log(kb_dir, kind=kind, id=id, near_id=near_id, cosine=cos)
+            return near_id, cos
+    return None
+
+
+def _log(kb_dir: Path, *, kind: str, id: str, near_id: str, cosine: float) -> None:
+    with index_db.open_db(kb_dir) as conn:
+        conn.execute(
+            "INSERT INTO embedding_dupes (kind, id, near_id, cosine, detected_at) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (kind, id, near_id, float(cosine),
+             dt.datetime.now(dt.UTC).isoformat(timespec="seconds")),
+        )
+
+
+def list_duplicates(kb_dir: Path) -> list[dict[str, Any]]:
+    with index_db.open_db(kb_dir) as conn:
+        rows = conn.execute(
+            "SELECT kind, id, near_id, cosine, detected_at FROM embedding_dupes "
+            "ORDER BY detected_at DESC"
+        ).fetchall()
+    return [
+        {"kind": k, "id": i, "near_id": n, "cosine": float(c), "detected_at": d}
+        for k, i, n, c, d in rows
+    ]
+
+
+def scan_all(
+    kb_dir: Path, *, threshold: float = DEFAULT_THRESHOLD, dry_run: bool = False,
+) -> list[dict[str, Any]]:
+    """Cross-artifact scan within same kind. For ad-hoc audit."""
+    found: list[dict[str, Any]] = []
+    with index_db.open_db(kb_dir) as conn:
+        rows = conn.execute(
+            "SELECT kind, id, vec, dim FROM embedding_index"
+        ).fetchall()
+    if not rows:
+        return found
+    vecs: dict[tuple[str, str], np.ndarray] = {}
+    for kind, id_, blob, dim in rows:
+        vecs[(kind, id_)] = np.frombuffer(blob, dtype=np.float32, count=dim).copy()
+    seen: set[frozenset[tuple[str, str]]] = set()
+    keys = list(vecs.keys())
+    for i, k1 in enumerate(keys):
+        for k2 in keys[i + 1:]:
+            if k1[0] != k2[0]:
+                continue
+            if vecs[k1].shape != vecs[k2].shape:
+                # Different embedding dims (e.g. mid-migration) aren't comparable.
+                continue
+            cos = float(vecs[k1] @ vecs[k2])
+            if cos >= threshold:
+                pair = frozenset([k1, k2])
+                if pair in seen:
+                    continue
+                seen.add(pair)
+                if not dry_run:
+                    _log(kb_dir, kind=k1[0], id=k2[1], near_id=k1[1], cosine=cos)
+                found.append({
+                    "kind": k1[0], "id": k2[1], "near_id": k1[1], "cosine": cos,
+                })
+    return found

src/vouch/embeddings/hyde.py

@@ -0,0 +1,27 @@
+"""HyDE -- Hypothetical Document Embedding expansion."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+DEFAULT_TEMPLATE = (
+    "The following is a document that answers the question: '{query}'. "
+    "It contains relevant facts, claims, and supporting evidence about {query}."
+)
+
+
+def expand_query_template(query: str, *, min_chars: int = 20) -> str:
+    """Pad short queries with HyDE template; pass through long ones."""
+    if len(query.strip()) >= min_chars:
+        return query
+    return DEFAULT_TEMPLATE.format(query=query.strip())
+
+
+def expand_query_with_llm(query: str, *, llm: Callable[[str], str]) -> str:
+    """Use an LLM to draft a hypothetical answer; embed that instead."""
+    prompt = (
+        "Write a short, factual paragraph that would answer this question. "
+        "Stay neutral; don't ask follow-ups.\n\n"
+        f"Question: {query}\n\nAnswer:"
+    )
+    return llm(prompt).strip() or query

src/vouch/embeddings/migration.py

@@ -0,0 +1,61 @@
+"""Model-identity migration and backfill."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+from .. import index_db
+from .base import get_embedder
+
+
+def detect_mismatch(kb_dir: Path) -> dict[str, Any] | None:
+    """Return mismatch info or None if no mismatch."""
+    meta = index_db.get_embedding_meta(kb_dir)
+    stored = meta.get("embedding_model")
+    if not stored:
+        return None
+    try:
+        current = get_embedder()
+    except KeyError:
+        return None
+    if current.name == stored:
+        return None
+    return {
+        "stored_model": stored,
+        "stored_version": meta.get("embedding_model_version"),
+        "stored_dim": meta.get("embedding_dim"),
+        "current_model": current.name,
+        "current_version": current.version,
+        "current_dim": current.dim,
+    }
+
+
+def backfill_embeddings(store: Any) -> int:
+    """Re-encode every artifact under the current adapter. Returns count touched."""
+    embedder = get_embedder()
+    touched = 0
+    # (list-method, kind, text-extractor). Missing list methods are skipped;
+    # errors raised inside the loop body propagate so a partial backfill never
+    # silently updates embedding_meta as if it had succeeded.
+    plan: list[tuple[str, str, Callable[[Any], str]]] = [
+        ("list_claims", "claim", lambda c: c.text),
+        ("list_pages", "page", lambda p: f"{p.title}\n\n{p.body}"),
+        ("list_sources", "source", lambda s: s.title or s.locator or ""),
+        ("list_entities", "entity", lambda e: f"{e.name}\n\n{e.description or ''}"),
+        ("list_relations", "relation",
+         lambda r: f"{r.source} {r.relation.value} {r.target}"),
+        ("list_evidence", "evidence", lambda ev: ev.quote or ""),
+    ]
+    for list_name, kind, text_of in plan:
+        lister = getattr(store, list_name, None)
+        if lister is None:
+            continue
+        for obj in lister():
+            store._embed_and_store(kind=kind, id=obj.id, text=text_of(obj))
+            touched += 1
+    index_db.set_embedding_meta(
+        store.kb_dir, model=embedder.name, version=embedder.version, dim=embedder.dim,
+    )
+    return touched

src/vouch/embeddings/rerank.py

@@ -0,0 +1,60 @@
+"""Cross-encoder reranking over candidate hits."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Sequence
+from typing import Any, ClassVar
+
+Hit = tuple[str, str, str, float]
+
+
+class Reranker(ABC):
+    name: ClassVar[str] = ""
+    version: ClassVar[str] = ""
+
+    @abstractmethod
+    def score(self, query: str, candidates: Sequence[str]) -> list[float]:
+        """Return one score per candidate. Higher = more relevant."""
+
+
+class CrossEncoderReranker(Reranker):
+    name = "cross-encoder/ms-marco-MiniLM-L6-v2"
+    version = "v1"
+
+    def __init__(self) -> None:
+        from sentence_transformers import CrossEncoder  # type: ignore[import-not-found]
+        self._model: Any = CrossEncoder(self.name)
+
+    def score(self, query: str, candidates: Sequence[str]) -> list[float]:
+        if not candidates:
+            return []
+        pairs = [(query, c) for c in candidates]
+        scores = self._model.predict(pairs)
+        return [float(s) for s in scores]
+
+
+def rerank(
+    *, query: str, hits: list[Hit], reranker: Reranker, top_k: int = 50,
+) -> list[Hit]:
+    if not hits:
+        return []
+    if top_k < 0:
+        raise ValueError(f"top_k must be >= 0, got {top_k}")
+    candidates = [h[2] or h[1] for h in hits]
+    scores = reranker.score(query, candidates)
+    if len(scores) != len(candidates):
+        raise ValueError(
+            f"reranker returned {len(scores)} scores for {len(candidates)} candidates"
+        )
+    reranked = [
+        (kind, id_, snip, score)
+        for (kind, id_, snip, _orig), score in zip(hits, scores, strict=True)
+    ]
+    reranked.sort(key=lambda h: h[3], reverse=True)
+    return reranked[:top_k]
+
+
+def default_reranker() -> Reranker:
+    """Return a CrossEncoderReranker; raise ImportError if extras not installed."""
+    return CrossEncoderReranker()

src/vouch/embeddings/scorer.py

@@ -0,0 +1,81 @@
+"""Retrieval evaluation metrics: recall@k, MRR, nDCG.
+
+Ground truth format: a set of "kind:id" strings (e.g. {"claim:c1"}).
+Hits format matches index_db: list of (kind, id, snippet, score).
+"""
+
+from __future__ import annotations
+
+import json
+import math
+from pathlib import Path
+
+Hit = tuple[str, str, str, float]
+
+
+def _key(h: Hit) -> str:
+    return f"{h[0]}:{h[1]}"
+
+
+def recall_at_k(hits: list[Hit], relevant: set[str], *, k: int = 10) -> float:
+    if not relevant:
+        return 0.0
+    top = {_key(h) for h in hits[:k]}
+    return len(top & relevant) / len(relevant)
+
+
+def mrr(hits: list[Hit], relevant: set[str]) -> float:
+    for i, h in enumerate(hits, start=1):
+        if _key(h) in relevant:
+            return 1.0 / i
+    return 0.0
+
+
+def ndcg(hits: list[Hit], relevant: set[str], *, k: int = 10) -> float:
+    dcg = 0.0
+    for i, h in enumerate(hits[:k], start=1):
+        if _key(h) in relevant:
+            dcg += 1.0 / math.log2(i + 1)
+    ideal = sum(1.0 / math.log2(i + 1) for i in range(1, min(len(relevant), k) + 1))
+    if ideal <= 0:
+        return 0.0
+    return dcg / ideal
+
+
+def evaluate(
+    *,
+    kb_dir: Path,
+    queries_file: Path,
+    k: int = 10,
+    metrics: tuple[str, ...] = ("recall@k", "mrr", "ndcg"),
+) -> dict[str, float]:
+    """Run a metric sweep over a JSONL queries file."""
+    from .. import index_db
+    known = {"recall@k", "mrr", "ndcg"}
+    unknown = set(metrics) - known
+    if unknown:
+        raise ValueError(f"unknown metric(s): {sorted(unknown)}; known: {sorted(known)}")
+    totals = {m: 0.0 for m in metrics}
+    n = 0
+    with queries_file.open() as f:
+        for line in f:
+            if not line.strip():
+                continue
+            row = json.loads(line)
+            q = row["query"]
+            rel = set(row["relevant"])
+            hits = index_db.search_semantic(kb_dir, q, limit=k)
+            if "recall@k" in metrics:
+                totals["recall@k"] += recall_at_k(hits, rel, k=k)
+            if "mrr" in metrics:
+                totals["mrr"] += mrr(hits, rel)
+            if "ndcg" in metrics:
+                totals["ndcg"] += ndcg(hits, rel, k=k)
+            n += 1
+    if n == 0:
+        return {m: 0.0 for m in metrics}
+    return {m: totals[m] / n for m in metrics}
+
+
+def write_report(out: dict[str, float], path: Path) -> None:
+    path.write_text(json.dumps(out, indent=2))

tests/embeddings/test_dedup.py

@@ -0,0 +1,58 @@
+"""Ingest-time duplicate detection."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from vouch.embeddings import register
+from vouch.embeddings.base import DEFAULT_MODEL_NAME, Embedder
+from vouch.embeddings.dedup import list_duplicates
+from vouch.models import Claim
+from vouch.storage import KBStore
+
+
+class _HashEmbedder(Embedder):
+    """Deterministic embedder using text length + hash byte — avoids float32 overflow."""
+
+    name = "mock"
+    version = "1"
+    dim = 8
+
+    def encode(self, text: str) -> np.ndarray:
+        import hashlib
+        h = hashlib.sha256(text.encode()).digest()
+        out = np.array([h[i] / 255.0 for i in range(self.dim)], dtype=np.float32)
+        norm = float(np.linalg.norm(out))
+        if norm > 0:
+            out /= norm
+        return out
+
+
+@pytest.fixture(autouse=True)
+def _register_default() -> None:
+    register(DEFAULT_MODEL_NAME, _HashEmbedder)
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> KBStore:
+    return KBStore.init(tmp_path)
+
+
+def test_identical_text_flagged_as_duplicate(store: KBStore) -> None:
+    src = store.put_source(b"e")
+    store.put_claim(Claim(id="c1", text="exact same text", evidence=[src.id]))
+    store.put_claim(Claim(id="c2", text="exact same text", evidence=[src.id]))
+    dupes = list_duplicates(store.kb_dir)
+    pairs = {(d["id"], d["near_id"]) for d in dupes}
+    assert ("c2", "c1") in pairs or ("c1", "c2") in pairs
+
+
+def test_disjoint_text_not_flagged(store: KBStore) -> None:
+    src = store.put_source(b"e")
+    store.put_claim(Claim(id="c1", text="apples", evidence=[src.id]))
+    store.put_claim(Claim(id="c2", text="zebras", evidence=[src.id]))
+    dupes = list_duplicates(store.kb_dir)
+    assert dupes == []