perf(retrieval): fuse normalized backend scores

app/mneme/memoria/server/eval/README.md

@@ -112,7 +112,7 @@ the same section.
 the official archive linked by the Hugging Face dataset card, verifies the
 published MD5, deterministically samples test queries, and compares the
 configured dense model with a TF-IDF lexical baseline, RRF hybrid retrieval,
-and an optional configured CrossEncoder. Generated data and reports stay under
+normalized score fusion, and an optional configured CrossEncoder. Generated data and reports stay under
 the git-ignored `storage/eval/` directory.
 
 The application image already contains the model dependencies, so no host
@@ -134,7 +134,7 @@ DOCKER=/Applications/Docker.app/Contents/Resources/bin/docker
   --sample-size 100 \
   --corpus-size 1000 \
   --top-k 10 \
-  --candidate-k 50 \
+  --candidate-k 200 \
   --rerank-k 20 \
   --reranker \
   --output storage/eval/beir/scifact-report.json

app/mneme/memoria/server/eval/beir.py

@@ -35,10 +35,13 @@
     calculate_ndcg,
     calculate_recall_at_k,
 )
+from app.mneme.memoria.server.retrieval.contracts import DocumentSearchHit
 from app.mneme.memoria.server.retrieval.fusion import (
+    DENSE_SCORE_WEIGHT,
     FUSION_CANDIDATE_K,
     LEXICAL_RRF_WEIGHT,
     RRF_CONSTANT,
+    normalized_score_fusion,
 )
 
 DATASET_NAME = "BEIR SciFact"
@@ -388,8 +391,9 @@ def run(args: argparse.Namespace) -> dict[str, Any]:
     lexical_queries = vectorizer.transform(query_texts)
     timings["lexical_fit_seconds"] = round(time.perf_counter() - started, 3)
     started = time.perf_counter()
+    lexical_scores = (lexical_queries @ lexical_corpus.T).toarray()
     lexical = {
-        query_id: _top_ids((lexical_queries[index] @ lexical_corpus.T).toarray().ravel(), doc_ids, args.candidate_k)
+        query_id: _top_ids(lexical_scores[index], doc_ids, args.candidate_k)
         for index, query_id in enumerate(query_ids)
     }
     timings["lexical_search_seconds"] = round(time.perf_counter() - started, 3)
@@ -405,12 +409,46 @@ def run(args: argparse.Namespace) -> dict[str, Any]:
         for query_id in query_ids
     }
     timings["hybrid_rrf_seconds"] = round(time.perf_counter() - started, 3)
+    started = time.perf_counter()
+    doc_index = {doc_id: index for index, doc_id in enumerate(doc_ids)}
+    score_fusion = {}
+    for query_index, query_id in enumerate(query_ids):
+        dense_hits = [
+            DocumentSearchHit(
+                doc_id,
+                doc_id,
+                doc_texts[doc_index[doc_id]],
+                {},
+                float(dense_scores[query_index, doc_index[doc_id]]),
+            )
+            for doc_id in dense[query_id]
+        ]
+        lexical_hits = [
+            DocumentSearchHit(
+                doc_id,
+                doc_id,
+                doc_texts[doc_index[doc_id]],
+                {},
+                float(lexical_scores[query_index, doc_index[doc_id]]),
+            )
+            for doc_id in lexical[query_id]
+        ]
+        score_fusion[query_id] = [
+            item.evidence_id
+            for item in normalized_score_fusion(
+                (dense_hits, lexical_hits),
+                top_k=args.candidate_k,
+                weights=(DENSE_SCORE_WEIGHT, 1.0 - DENSE_SCORE_WEIGHT),
+            )
+        ]
+    timings["score_fusion_seconds"] = round(time.perf_counter() - started, 3)
 
     rankings = {
         "dense": dense,
         "lexical_tfidf": lexical,
         "hybrid_rrf": hybrid,
         "dense_weighted_rrf": weighted_hybrid,
+        "normalized_score_fusion": score_fusion,
     }
     reranker_model = None
     if args.reranker:
@@ -494,6 +532,7 @@ def run(args: argparse.Namespace) -> dict[str, Any]:
             "production_candidate_k": FUSION_CANDIDATE_K,
             "dense_rrf_weight": 1.0,
             "lexical_rrf_weight": LEXICAL_RRF_WEIGHT,
+            "dense_score_weight": DENSE_SCORE_WEIGHT,
         },
         "metrics": metrics,
         "timings": timings,

app/mneme/memoria/server/retrieval/contracts.py

@@ -29,3 +29,4 @@ class DocumentSearchHit:
     document_id: str
     content: str
     metadata: dict[str, Any]
+    score: float = 0.0

app/mneme/memoria/server/retrieval/documents.py

@@ -6,8 +6,8 @@
 from app.mneme.memoria.server.database import open_read_session
 from app.mneme.memoria.server.retrieval.contracts import RetrievalScope, RetrievedEvidence
 from app.mneme.memoria.server.retrieval.fusion import (
-    LEXICAL_RRF_WEIGHT,
-    reciprocal_rank_fusion,
+    DENSE_SCORE_WEIGHT,
+    normalized_score_fusion,
 )
 from app.mneme.memoria.server.retrieval.keyword import search_keyword
 from app.mneme.memoria.server.retrieval.vector import search_vector
@@ -25,7 +25,7 @@ async def search(
         query: str,
         top_k: int,
     ) -> list[RetrievedEvidence]:
-        """Search vector and keyword indexes and combine their ranks with RRF.
+        """Search vector and keyword indexes and combine normalized raw scores.
 
         Both queries apply the same scope and active-projection constraints.
         Empty or non-positive limits return no evidence without touching storage.
@@ -36,8 +36,8 @@ async def search(
         async with open_read_session() as db:
             vector_hits = await search_vector(db, scope=scope, query=query, limit=top_k)
             keyword_hits = await search_keyword(db, scope=scope, query=query, limit=top_k)
-        return reciprocal_rank_fusion(
+        return normalized_score_fusion(
             (vector_hits, keyword_hits),
             top_k=top_k,
-            weights=(1.0, LEXICAL_RRF_WEIGHT),
+            weights=(DENSE_SCORE_WEIGHT, 1.0 - DENSE_SCORE_WEIGHT),
         )

app/mneme/memoria/server/retrieval/fusion.py

@@ -1,15 +1,16 @@
-"""Fuse document rankings with deterministic reciprocal-rank fusion.
+"""Fuse document rankings with deterministic rank and normalized-score fusion.
 
-Fusion uses rank positions instead of incomparable raw vector and lexical score scales.
+Backend scores are normalized per query before comparison across retrieval methods.
 """
 
 from collections.abc import Sequence
 
 from app.mneme.memoria.server.retrieval.contracts import DocumentSearchHit, RetrievedEvidence
 
 RRF_CONSTANT = 60
-FUSION_CANDIDATE_K = 100
+FUSION_CANDIDATE_K = 200
 LEXICAL_RRF_WEIGHT = 0.55
+DENSE_SCORE_WEIGHT = 0.55
 
 
 def reciprocal_rank_fusion(
@@ -59,3 +60,48 @@ def reciprocal_rank_fusion(
         )
         for chunk_id in ranked_chunk_ids
     ]
+
+
+def normalized_score_fusion(
+    rankings: Sequence[Sequence[DocumentSearchHit]],
+    *,
+    top_k: int,
+    weights: Sequence[float],
+) -> list[RetrievedEvidence]:
+    """Min-max normalize each backend's scores, then combine weighted candidates."""
+    if top_k <= 0:
+        return []
+    if len(weights) != len(rankings):
+        raise ValueError("weights must match the number of rankings")
+
+    hits_by_chunk_id: dict[str, DocumentSearchHit] = {}
+    scores_by_chunk_id: dict[str, float] = {}
+    for ranking, weight in zip(rankings, weights, strict=True):
+        unique_hits = list({hit.chunk_id: hit for hit in ranking}.values())
+        if not unique_hits:
+            continue
+        minimum = min(hit.score for hit in unique_hits)
+        maximum = max(hit.score for hit in unique_hits)
+        spread = maximum - minimum
+        for hit in unique_hits:
+            hits_by_chunk_id.setdefault(hit.chunk_id, hit)
+            normalized = (hit.score - minimum) / spread if spread > 0 else 1.0
+            scores_by_chunk_id[hit.chunk_id] = (
+                scores_by_chunk_id.get(hit.chunk_id, 0.0) + weight * normalized
+            )
+
+    ranked_chunk_ids = sorted(
+        scores_by_chunk_id,
+        key=lambda chunk_id: (-scores_by_chunk_id[chunk_id], chunk_id),
+    )[:top_k]
+    return [
+        RetrievedEvidence(
+            evidence_id=chunk_id,
+            source_type="document",
+            source_id=hits_by_chunk_id[chunk_id].document_id,
+            content=hits_by_chunk_id[chunk_id].content,
+            score=scores_by_chunk_id[chunk_id],
+            metadata=hits_by_chunk_id[chunk_id].metadata,
+        )
+        for chunk_id in ranked_chunk_ids
+    ]

app/mneme/memoria/server/retrieval/keyword.py

@@ -28,7 +28,7 @@ async def search_keyword(
         return []
 
     text_query = func.websearch_to_tsquery(literal_column("'simple'::regconfig"), query)
-    rank = func.ts_rank_cd(DocumentChunk.search_vector, text_query)
+    rank = func.ts_rank_cd(DocumentChunk.search_vector, text_query).label("score")
     statement = (
         select(
             DocumentChunk.chunk_id,
@@ -39,6 +39,7 @@ async def search_keyword(
             DocumentChunk.page_no,
             DocumentChunk.section_path,
             DocumentProjection.file_name,
+            rank,
         )
         .join(
             DocumentProjection,
@@ -62,6 +63,7 @@ async def search_keyword(
             chunk_id=row["chunk_id"],
             document_id=row["document_id"],
             content=row["content"],
+            score=float(row["score"]),
             metadata={
                 "document_version": row["document_version"],
                 "file_name": row["file_name"],

app/mneme/memoria/server/retrieval/vector.py

@@ -34,6 +34,7 @@ async def search_vector(
     await db.execute(text("SET LOCAL hnsw.ef_search = 100"))
     await db.execute(text("SET LOCAL hnsw.iterative_scan = strict_order"))
     distance = DocumentChunk.embedding.cosine_distance(query_embedding)
+    similarity = (1 - distance).label("score")
     statement = (
         select(
             DocumentChunk.chunk_id,
@@ -44,6 +45,7 @@ async def search_vector(
             DocumentChunk.page_no,
             DocumentChunk.section_path,
             DocumentProjection.file_name,
+            similarity,
         )
         .join(
             DocumentProjection,
@@ -66,6 +68,7 @@ async def search_vector(
             chunk_id=row["chunk_id"],
             document_id=row["document_id"],
             content=row["content"],
+            score=float(row["score"]),
             metadata={
                 "document_version": row["document_version"],
                 "file_name": row["file_name"],