Chris CoutinhoandClaude Opus 4.7
1ed8362f78
Three minor fixes from the round-11 review on PR #750: - bm25_hybrid.py:209 — Comment said `doc_id` is `int (notes) or str (files)`, which is backwards. Notes, news_items, and deck_cards are stored as `str` (scanner.py:241, 666, 867); files are stored as `int` (scanner.py:425). Updated to point readers at scanner.py as the source of truth. - verification.py:338 — Lowered the News-API 403/404 log line from `info` to `debug`. The News app being uninstalled or disabled is a predictable operational state (matching the other verifiers' debug-on-not-found paths), so this should not generate operator-dashboard noise. Transient errors immediately below stay at `warning` because they're unexpected. - semantic.py:809 — `nc_get_vector_sync_status` was reading `document_receive_stream` via `getattr(..., None)`, but the attribute is guaranteed-defined on both `AppContext` and `OAuthAppContext` (as a field with `None` default). The defensive `getattr` masked typos that the eviction_task_group access at semantic.py:197-199 deliberately surfaces. Switched to direct access; the `if … is None:` value-check below is preserved (the attribute can legitimately be None before sync starts). Items deliberately deferred (with rationale in the plan file): - News verifier semaphore-hold during get_items (reviewer: "not required here, just worth tracking"; ADR already lists follow-ups). - Hardcoded 2× over-fetch / VERIFICATION_OVERFETCH (TODO already in code). - Integration test for the real Qdrant eviction filter (reviewer marked low-priority; type-preservation chain is unit-tested). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
276 lines
11 KiB
Python
276 lines
11 KiB
Python
"""BM25 hybrid search algorithm using Qdrant native RRF fusion."""
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from qdrant_client import models
|
||
from qdrant_client.models import FieldCondition, Filter, MatchValue
|
||
|
||
from nextcloud_mcp_server.config import get_settings
|
||
from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service
|
||
from nextcloud_mcp_server.observability.metrics import record_qdrant_operation
|
||
from nextcloud_mcp_server.observability.tracing import trace_operation
|
||
from nextcloud_mcp_server.search.algorithms import SearchAlgorithm, SearchResult
|
||
from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
|
||
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class BM25HybridSearchAlgorithm(SearchAlgorithm):
|
||
"""
|
||
Hybrid search combining dense semantic vectors with BM25 sparse vectors.
|
||
|
||
Uses Qdrant's native Reciprocal Rank Fusion (RRF) to automatically merge
|
||
results from both dense (semantic) and sparse (BM25 keyword) searches.
|
||
This provides the best of both worlds: semantic understanding for conceptual
|
||
queries and precise keyword matching for specific terms, acronyms, and codes.
|
||
|
||
The fusion happens efficiently in the database using the prefetch mechanism,
|
||
eliminating the need for application-layer result merging.
|
||
"""
|
||
|
||
def __init__(self, score_threshold: float = 0.0, fusion: str = "rrf"):
|
||
"""
|
||
Initialize BM25 hybrid search algorithm.
|
||
|
||
Args:
|
||
score_threshold: Minimum fusion score (0-1, default: 0.0 to allow fusion scoring)
|
||
Note: Both RRF and DBSF produce normalized scores
|
||
fusion: Fusion algorithm to use: "rrf" (Reciprocal Rank Fusion, default)
|
||
or "dbsf" (Distribution-Based Score Fusion)
|
||
|
||
Raises:
|
||
ValueError: If fusion is not "rrf" or "dbsf"
|
||
"""
|
||
if fusion not in ("rrf", "dbsf"):
|
||
raise ValueError(
|
||
f"Invalid fusion algorithm '{fusion}'. Must be 'rrf' or 'dbsf'"
|
||
)
|
||
|
||
self.score_threshold = score_threshold
|
||
self.fusion = models.Fusion.RRF if fusion == "rrf" else models.Fusion.DBSF
|
||
self.fusion_name = fusion
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return "bm25_hybrid"
|
||
|
||
@property
|
||
def requires_vector_db(self) -> bool:
|
||
return True
|
||
|
||
async def search(
|
||
self,
|
||
query: str,
|
||
user_id: str,
|
||
limit: int = 10,
|
||
doc_type: str | None = None,
|
||
**kwargs: Any,
|
||
) -> list[SearchResult]:
|
||
"""
|
||
Execute hybrid search using dense + sparse vectors with native RRF fusion.
|
||
|
||
Returns unverified results from Qdrant. Access verification is
|
||
performed separately at the server tool layer via
|
||
``nextcloud_mcp_server.search.verification.verify_search_results``
|
||
(see ADR-019).
|
||
|
||
Deduplicates by (doc_id, doc_type, chunk_start_offset, chunk_end_offset)
|
||
to show multiple chunks from the same document while avoiding duplicate chunks.
|
||
|
||
Args:
|
||
query: Natural language or keyword search query
|
||
user_id: User ID for filtering
|
||
limit: Maximum results to return
|
||
doc_type: Optional document type filter
|
||
**kwargs: Additional parameters (score_threshold override)
|
||
|
||
Returns:
|
||
List of unverified SearchResult objects ranked by RRF fusion score
|
||
|
||
Raises:
|
||
McpError: If vector sync is not enabled or search fails
|
||
"""
|
||
settings = get_settings()
|
||
score_threshold = kwargs.get("score_threshold", self.score_threshold)
|
||
|
||
logger.info(
|
||
f"BM25 hybrid search: query='{query}', user={user_id}, "
|
||
f"limit={limit}, score_threshold={score_threshold}, doc_type={doc_type}, "
|
||
f"fusion={self.fusion_name}"
|
||
)
|
||
|
||
# Generate dense embedding for semantic search
|
||
with trace_operation("search.get_embedding_service"):
|
||
embedding_service = get_embedding_service()
|
||
with trace_operation("search.dense_embedding"):
|
||
dense_embedding = await embedding_service.embed(query)
|
||
# Store for reuse by callers (e.g., viz_routes PCA visualization)
|
||
self.query_embedding = dense_embedding
|
||
logger.debug(f"Generated dense embedding (dimension={len(dense_embedding)})")
|
||
|
||
# Generate sparse embedding for BM25 keyword search
|
||
with trace_operation("search.get_bm25_service"):
|
||
bm25_service = get_bm25_service()
|
||
with trace_operation("search.sparse_embedding_bm25"):
|
||
sparse_embedding = await bm25_service.encode_async(query)
|
||
logger.debug(
|
||
f"Generated sparse embedding "
|
||
f"({len(sparse_embedding['indices'])} non-zero terms)"
|
||
)
|
||
|
||
# Build Qdrant filter
|
||
filter_conditions = [
|
||
get_placeholder_filter(), # Always exclude placeholders from user-facing queries
|
||
FieldCondition(
|
||
key="user_id",
|
||
match=MatchValue(value=user_id),
|
||
),
|
||
]
|
||
|
||
# Add doc_type filter if specified
|
||
if doc_type:
|
||
filter_conditions.append(
|
||
FieldCondition(
|
||
key="doc_type",
|
||
match=MatchValue(value=doc_type),
|
||
)
|
||
)
|
||
|
||
query_filter = Filter(must=filter_conditions)
|
||
|
||
# Execute hybrid search with Qdrant native RRF fusion
|
||
with trace_operation("search.get_qdrant_client"):
|
||
qdrant_client = await get_qdrant_client()
|
||
|
||
try:
|
||
# Use prefetch to run both dense and sparse searches
|
||
# Qdrant will automatically merge results using RRF
|
||
with trace_operation(
|
||
"search.qdrant_query",
|
||
attributes={"query.limit": limit * 2, "query.fusion": self.fusion_name},
|
||
):
|
||
search_response = await qdrant_client.query_points(
|
||
collection_name=settings.get_collection_name(),
|
||
prefetch=[
|
||
# Dense semantic search
|
||
models.Prefetch(
|
||
query=dense_embedding,
|
||
using="dense",
|
||
limit=limit * 2, # Get extra for deduplication
|
||
filter=query_filter,
|
||
),
|
||
# Sparse BM25 search
|
||
models.Prefetch(
|
||
query=models.SparseVector(
|
||
indices=sparse_embedding["indices"],
|
||
values=sparse_embedding["values"],
|
||
),
|
||
using="sparse",
|
||
limit=limit * 2, # Get extra for deduplication
|
||
filter=query_filter,
|
||
),
|
||
],
|
||
# Fusion query (RRF or DBSF based on initialization)
|
||
query=models.FusionQuery(fusion=self.fusion),
|
||
limit=limit * 2, # Get extra for deduplication
|
||
score_threshold=score_threshold,
|
||
with_payload=True,
|
||
with_vectors=False, # Don't return vectors to save bandwidth
|
||
)
|
||
record_qdrant_operation("search", "success")
|
||
except Exception:
|
||
record_qdrant_operation("search", "error")
|
||
raise
|
||
|
||
logger.info(
|
||
f"Qdrant {self.fusion_name.upper()} fusion returned {len(search_response.points)} results "
|
||
f"(before deduplication)"
|
||
)
|
||
|
||
if search_response.points:
|
||
# Log top 3 fusion scores to help with threshold tuning
|
||
top_scores = [p.score for p in search_response.points[:3]]
|
||
logger.debug(
|
||
f"Top 3 {self.fusion_name.upper()} fusion scores: {top_scores}"
|
||
)
|
||
|
||
# Deduplicate by (doc_id, doc_type, chunk_start, chunk_end)
|
||
# This allows multiple chunks from same doc, but removes duplicate chunks
|
||
with trace_operation(
|
||
"search.deduplicate",
|
||
attributes={"dedupe.num_points": len(search_response.points)},
|
||
):
|
||
seen_chunks = set()
|
||
results = []
|
||
|
||
for result in search_response.points:
|
||
if result.payload is None:
|
||
continue
|
||
# doc_id can be int (files) or str (notes/news_items/deck_cards) — see scanner.py
|
||
doc_id = result.payload["doc_id"]
|
||
doc_type = result.payload.get("doc_type", "note")
|
||
chunk_start = result.payload.get("chunk_start_offset")
|
||
chunk_end = result.payload.get("chunk_end_offset")
|
||
chunk_key = (doc_id, doc_type, chunk_start, chunk_end)
|
||
|
||
# Skip if we've already seen this exact chunk
|
||
if chunk_key in seen_chunks:
|
||
continue
|
||
|
||
seen_chunks.add(chunk_key)
|
||
|
||
# Build metadata dict with common fields
|
||
metadata = {
|
||
"chunk_index": result.payload.get("chunk_index"),
|
||
"total_chunks": result.payload.get("total_chunks"),
|
||
"search_method": f"bm25_hybrid_{self.fusion_name}",
|
||
}
|
||
|
||
# Add file-specific metadata for PDF viewer
|
||
if doc_type == "file" and (path := result.payload.get("file_path")):
|
||
metadata["path"] = path
|
||
|
||
# Add deck_card-specific metadata for frontend URL construction
|
||
# and verify-on-read (ADR-019) — both board_id and stack_id are
|
||
# required to call deck.get_card without an O(boards × stacks)
|
||
# iteration fallback.
|
||
if doc_type == "deck_card":
|
||
if board_id := result.payload.get("board_id"):
|
||
metadata["board_id"] = board_id
|
||
if stack_id := result.payload.get("stack_id"):
|
||
metadata["stack_id"] = stack_id
|
||
|
||
# Return unverified results (verification happens at output stage)
|
||
results.append(
|
||
SearchResult(
|
||
id=doc_id,
|
||
doc_type=doc_type,
|
||
title=result.payload.get("title", "Untitled"),
|
||
excerpt=result.payload.get("excerpt", ""),
|
||
score=result.score, # Fusion score (RRF or DBSF)
|
||
metadata=metadata,
|
||
chunk_start_offset=result.payload.get("chunk_start_offset"),
|
||
chunk_end_offset=result.payload.get("chunk_end_offset"),
|
||
page_number=result.payload.get("page_number"),
|
||
page_count=result.payload.get("page_count"),
|
||
chunk_index=result.payload.get("chunk_index", 0),
|
||
total_chunks=result.payload.get("total_chunks", 1),
|
||
point_id=str(result.id), # Qdrant point ID for batch retrieval
|
||
)
|
||
)
|
||
|
||
if len(results) >= limit:
|
||
break
|
||
|
||
logger.info(f"Returning {len(results)} unverified results after deduplication")
|
||
if results:
|
||
result_details = [
|
||
f"{r.doc_type}_{r.id} (score={r.score:.3f}, title='{r.title}')"
|
||
for r in results[:5] # Show top 5
|
||
]
|
||
logger.debug(f"Top results: {', '.join(result_details)}")
|
||
|
||
return results
|