admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feature/bm25'

Browse Source
This commit is contained in:
Chris Coutinho
2025-11-16 07:57:36 +01:00
parent d1fb7eb633 fc6a2f14e4
commit 2aa82d849c
18 changed files with 1033 additions and 1099 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/auth/userinfo_routes.py
+2 -8
View File
@@ -676,14 +676,11 @@ async def user_info_html(request: Request) -> HTMLResponse:
function vizApp() {{
return {{
query: '',
algorithm: 'hybrid',
algorithm: 'bm25_hybrid',
showAdvanced: false,
docTypes: [''], // Default to "All Types"
limit: 50,
scoreThreshold: 0.7,
semanticWeight: 0.5,
keywordWeight: 0.3,
fuzzyWeight: 0.2,
scoreThreshold: 0.0,
loading: false,
results: [],
@@ -697,9 +694,6 @@ async def user_info_html(request: Request) -> HTMLResponse:
algorithm: this.algorithm,
limit: this.limit,
score_threshold: this.scoreThreshold,
semantic_weight: this.semanticWeight,
keyword_weight: this.keywordWeight,
fuzzy_weight: this.fuzzyWeight,
}});
// Add doc_types parameter (filter out empty string for "All Types")
nextcloud_mcp_server/auth/viz_routes.py
+14 -41
View File
@@ -20,9 +20,7 @@ from starlette.responses import HTMLResponse, JSONResponse
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.search import (
FuzzySearchAlgorithm,
HybridSearchAlgorithm,
KeywordSearchAlgorithm,
BM25HybridSearchAlgorithm,
SemanticSearchAlgorithm,
)
from nextcloud_mcp_server.vector.pca import PCA
@@ -209,10 +207,8 @@ async def vector_visualization_html(request: Request) -> HTMLResponse:
<div class="viz-control-group" style="margin-bottom: 0;">
<label>Algorithm</label>
<select x-model="algorithm">
<option value="semantic">Semantic (Vector Similarity)</option>
<option value="keyword">Keyword (Token Matching)</option>
<option value="fuzzy">Fuzzy (Character Overlap)</option>
<option value="hybrid" selected>Hybrid (RRF Fusion)</option>
<option value="semantic">Semantic (Dense Vectors)</option>
<option value="bm25_hybrid" selected>BM25 Hybrid (Dense + Sparse RRF)</option>
</select>
</div>
@@ -260,25 +256,13 @@ async def vector_visualization_html(request: Request) -> HTMLResponse:
</div>
</div>
<!-- Hybrid Weights (only when hybrid selected) -->
<div x-show="algorithm === 'hybrid'" style="margin-top: 16px; padding: 12px; background: #e9ecef; border-radius: 4px;">
<label style="margin-bottom: 12px; display: block;">Hybrid Algorithm Weights</label>
<div style="margin-bottom: 8px;">
<label style="display: inline-block; width: 100px; font-weight: normal;">Semantic:</label>
<input type="range" x-model.number="semanticWeight" min="0" max="1" step="0.1" style="width: 200px; display: inline-block;">
<span class="viz-weight-display" x-text="semanticWeight.toFixed(1)"></span>
</div>
<div style="margin-bottom: 8px;">
<label style="display: inline-block; width: 100px; font-weight: normal;">Keyword:</label>
<input type="range" x-model.number="keywordWeight" min="0" max="1" step="0.1" style="width: 200px; display: inline-block;">
<span class="viz-weight-display" x-text="keywordWeight.toFixed(1)"></span>
</div>
<div>
<label style="display: inline-block; width: 100px; font-weight: normal;">Fuzzy:</label>
<input type="range" x-model.number="fuzzyWeight" min="0" max="1" step="0.1" style="width: 200px; display: inline-block;">
<span class="viz-weight-display" x-text="fuzzyWeight.toFixed(1)"></span>
</div>
<!-- Info: BM25 Hybrid uses native RRF fusion (no manual weights) -->
<div x-show="algorithm === 'bm25_hybrid'" style="margin-top: 16px; padding: 12px; background: #e9ecef; border-radius: 4px;">
<p style="margin: 0; font-size: 14px; color: #666;">
<strong>BM25 Hybrid Search:</strong> Uses Qdrant's native Reciprocal Rank Fusion (RRF)
to automatically combine dense semantic vectors with sparse BM25 keyword vectors.
No manual weight tuning required.
</p>
</div>
</div>
</div>
@@ -365,12 +349,9 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
# Parse query parameters
query = request.query_params.get("query", "")
algorithm = request.query_params.get("algorithm", "hybrid")
algorithm = request.query_params.get("algorithm", "bm25_hybrid")
limit = int(request.query_params.get("limit", "50"))
score_threshold = float(request.query_params.get("score_threshold", "0.7"))
semantic_weight = float(request.query_params.get("semantic_weight", "0.5"))
keyword_weight = float(request.query_params.get("keyword_weight", "0.3"))
fuzzy_weight = float(request.query_params.get("fuzzy_weight", "0.2"))
score_threshold = float(request.query_params.get("score_threshold", "0.0"))
# Parse doc_types (comma-separated list, None = all types)
doc_types_param = request.query_params.get("doc_types", "")
@@ -395,16 +376,8 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
# Create search algorithm (no client needed - verification removed)
if algorithm == "semantic":
search_algo = SemanticSearchAlgorithm(score_threshold=score_threshold)
elif algorithm == "keyword":
search_algo = KeywordSearchAlgorithm()
elif algorithm == "fuzzy":
search_algo = FuzzySearchAlgorithm()
elif algorithm == "hybrid":
search_algo = HybridSearchAlgorithm(
semantic_weight=semantic_weight,
keyword_weight=keyword_weight,
fuzzy_weight=fuzzy_weight,
)
elif algorithm == "bm25_hybrid":
search_algo = BM25HybridSearchAlgorithm(score_threshold=score_threshold)
else:
return JSONResponse(
{"success": False, "error": f"Unknown algorithm: {algorithm}"},
nextcloud_mcp_server/embedding/__init__.py
+9 -2
View File
@@ -1,6 +1,13 @@
"""Embedding service package for generating vector embeddings."""
from .service import EmbeddingService, get_embedding_service
from .bm25_provider import BM25SparseEmbeddingProvider
from .service import EmbeddingService, get_bm25_service, get_embedding_service
from .simple_provider import SimpleEmbeddingProvider
__all__ = ["EmbeddingService", "get_embedding_service", "SimpleEmbeddingProvider"]
__all__ = [
"EmbeddingService",
"get_embedding_service",
"BM25SparseEmbeddingProvider",
"get_bm25_service",
"SimpleEmbeddingProvider",
]
nextcloud_mcp_server/embedding/bm25_provider.py
+74
View File
@@ -0,0 +1,74 @@
"""BM25 sparse embedding provider using FastEmbed."""
import logging
from typing import Any
from fastembed import SparseTextEmbedding
logger = logging.getLogger(__name__)
class BM25SparseEmbeddingProvider:
"""
BM25 sparse embedding provider for hybrid search.
Uses FastEmbed's BM25 model to generate sparse vectors for keyword-based
retrieval. These sparse vectors are combined with dense semantic vectors
in Qdrant using Reciprocal Rank Fusion (RRF) for hybrid search.
Unlike dense embeddings which have fixed dimensions, sparse embeddings
have variable-length vectors with (index, value) pairs representing
term frequencies in the BM25 vocabulary.
"""
def __init__(self, model_name: str = "Qdrant/bm25"):
"""
Initialize BM25 sparse embedding provider.
Args:
model_name: FastEmbed BM25 model name (default: Qdrant/bm25)
"""
self.model_name = model_name
logger.info(f"Initializing BM25 sparse embedding provider: {model_name}")
# Initialize FastEmbed sparse embedding model
self.model = SparseTextEmbedding(model_name=model_name)
logger.info(f"BM25 sparse embedding model loaded: {model_name}")
def encode(self, text: str) -> dict[str, Any]:
"""
Generate BM25 sparse embedding for a single text.
Args:
text: Input text to encode
Returns:
Dictionary with 'indices' and 'values' keys for Qdrant sparse vector
"""
# FastEmbed returns a generator, take first result
sparse_embedding = next(iter(self.model.embed([text])))
return {
"indices": sparse_embedding.indices.tolist(),
"values": sparse_embedding.values.tolist(),
}
def encode_batch(self, texts: list[str]) -> list[dict[str, Any]]:
"""
Generate BM25 sparse embeddings for multiple texts (batched).
Args:
texts: List of texts to encode
Returns:
List of dictionaries with 'indices' and 'values' for each text
"""
sparse_embeddings = list(self.model.embed(texts))
return [
{
"indices": emb.indices.tolist(),
"values": emb.values.tolist(),
}
for emb in sparse_embeddings
]
nextcloud_mcp_server/embedding/service.py
+18
View File
@@ -4,6 +4,7 @@ import logging
import os
from .base import EmbeddingProvider
from .bm25_provider import BM25SparseEmbeddingProvider
from .ollama_provider import OllamaEmbeddingProvider
from .simple_provider import SimpleEmbeddingProvider
@@ -109,3 +110,20 @@ def get_embedding_service() -> EmbeddingService:
if _embedding_service is None:
_embedding_service = EmbeddingService()
return _embedding_service
# BM25 sparse embedding singleton
_bm25_service: BM25SparseEmbeddingProvider | None = None
def get_bm25_service() -> BM25SparseEmbeddingProvider:
"""
Get singleton BM25 sparse embedding service instance.
Returns:
Global BM25SparseEmbeddingProvider instance
"""
global _bm25_service
if _bm25_service is None:
_bm25_service = BM25SparseEmbeddingProvider()
return _bm25_service
nextcloud_mcp_server/search/__init__.py
+8 -14
View File
@@ -1,13 +1,11 @@
"""Search algorithms module for unified multi-algorithm search.
"""Search algorithms module for BM25 hybrid search.
This module provides a unified interface for different search algorithms:
- Semantic search (vector similarity)
- Keyword search (token-based matching)
- Fuzzy search (character overlap)
- Hybrid search (RRF fusion of multiple algorithms)
This module provides BM25 hybrid search combining:
- Dense semantic vectors (vector similarity via embeddings)
- Sparse BM25 vectors (keyword-based retrieval)
All algorithms share the same interface and can be used interchangeably by both
MCP tools and the visualization pane.
Results are fused using Qdrant's native Reciprocal Rank Fusion (RRF) for
optimal relevance across both semantic and keyword queries.
"""
from nextcloud_mcp_server.search.algorithms import (
@@ -16,9 +14,7 @@ from nextcloud_mcp_server.search.algorithms import (
SearchResult,
get_indexed_doc_types,
)
from nextcloud_mcp_server.search.fuzzy import FuzzySearchAlgorithm
from nextcloud_mcp_server.search.hybrid import HybridSearchAlgorithm
from nextcloud_mcp_server.search.keyword import KeywordSearchAlgorithm
from nextcloud_mcp_server.search.bm25_hybrid import BM25HybridSearchAlgorithm
from nextcloud_mcp_server.search.semantic import SemanticSearchAlgorithm
__all__ = [
@@ -27,7 +23,5 @@ __all__ = [
"SearchResult",
"get_indexed_doc_types",
"SemanticSearchAlgorithm",
"KeywordSearchAlgorithm",
"FuzzySearchAlgorithm",
"HybridSearchAlgorithm",
"BM25HybridSearchAlgorithm",
]
nextcloud_mcp_server/search/bm25_hybrid.py
+206
View File
@@ -0,0 +1,206 @@
"""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.search.algorithms import SearchAlgorithm, SearchResult
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):
"""
Initialize BM25 hybrid search algorithm.
Args:
score_threshold: Minimum RRF score (0-1, default: 0.0 to allow RRF scoring)
Note: RRF produces normalized scores, so threshold is typically lower
"""
self.score_threshold = score_threshold
@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 should be
performed separately at the final output stage using verify_search_results().
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}"
)
# Generate dense embedding for semantic search
embedding_service = get_embedding_service()
dense_embedding = await embedding_service.embed(query)
logger.debug(f"Generated dense embedding (dimension={len(dense_embedding)})")
# Generate sparse embedding for BM25 keyword search
bm25_service = get_bm25_service()
sparse_embedding = bm25_service.encode(query)
logger.debug(
f"Generated sparse embedding "
f"({len(sparse_embedding['indices'])} non-zero terms)"
)
# Build Qdrant filter
filter_conditions = [
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
qdrant_client = await get_qdrant_client()
try:
# Use prefetch to run both dense and sparse searches
# Qdrant will automatically merge results using RRF
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,
),
],
# RRF fusion query (no additional query needed, just fusion)
query=models.FusionQuery(fusion=models.Fusion.RRF),
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 RRF fusion returned {len(search_response.points)} results "
f"(before deduplication)"
)
if search_response.points:
# Log top 3 RRF scores to help with threshold tuning
top_scores = [p.score for p in search_response.points[:3]]
logger.debug(f"Top 3 RRF fusion scores: {top_scores}")
# Deduplicate by (doc_id, doc_type) - multiple chunks per document
seen_docs = set()
results = []
for result in search_response.points:
doc_id = int(result.payload["doc_id"])
doc_type = result.payload.get("doc_type", "note")
doc_key = (doc_id, doc_type)
# Skip if we've already seen this document
if doc_key in seen_docs:
continue
seen_docs.add(doc_key)
# 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, # RRF fusion score
metadata={
"chunk_index": result.payload.get("chunk_index"),
"total_chunks": result.payload.get("total_chunks"),
"search_method": "bm25_hybrid_rrf",
},
)
)
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
nextcloud_mcp_server/search/fuzzy.py
-219
View File
@@ -1,219 +0,0 @@
"""Fuzzy search algorithm using character overlap matching on Qdrant payload."""
import logging
from typing import Any
from qdrant_client.models import FieldCondition, Filter, MatchValue
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.search.algorithms import SearchAlgorithm, SearchResult
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
logger = logging.getLogger(__name__)
class FuzzySearchAlgorithm(SearchAlgorithm):
"""Fuzzy search using simple character-based similarity.
Implements character overlap matching with configurable threshold:
- Compares character sets between query and text
- Requires configurable % character overlap to match (default: 70%)
- Tolerant to typos and minor variations
"""
def __init__(self, threshold: float = 0.7):
"""Initialize fuzzy search algorithm.
Args:
threshold: Minimum character overlap ratio (0-1, default: 0.7)
"""
if not 0.0 <= threshold <= 1.0:
raise ValueError(f"Threshold must be between 0.0 and 1.0, got {threshold}")
self.threshold = threshold
@property
def name(self) -> str:
return "fuzzy"
async def search(
self,
query: str,
user_id: str,
limit: int = 10,
doc_type: str | None = None,
**kwargs: Any,
) -> list[SearchResult]:
"""Execute fuzzy search using character overlap on Qdrant payload.
Queries Qdrant for all indexed documents, then scores based on character
overlap in title and excerpt fields. Returns unverified results - access
verification should be performed separately at the final output stage.
Args:
query: Search query
user_id: User ID for filtering
limit: Maximum results to return
doc_type: Optional document type filter (None = all types)
**kwargs: Additional parameters (threshold override)
Returns:
List of unverified SearchResult objects ranked by character overlap score
"""
settings = get_settings()
threshold = kwargs.get("threshold", self.threshold)
logger.info(
f"Fuzzy search: query='{query}', user={user_id}, "
f"limit={limit}, threshold={threshold}, doc_type={doc_type}"
)
# Build Qdrant filter
filter_conditions = [
FieldCondition(key="user_id", match=MatchValue(value=user_id))
]
if doc_type:
filter_conditions.append(
FieldCondition(key="doc_type", match=MatchValue(value=doc_type))
)
# Scroll through Qdrant to get all matching documents
qdrant_client = await get_qdrant_client()
collection = settings.get_collection_name()
all_points = []
offset = None
# Scroll through all points matching filter
while True:
scroll_result, next_offset = await qdrant_client.scroll(
collection_name=collection,
scroll_filter=Filter(must=filter_conditions),
limit=100, # Batch size
offset=offset,
with_payload=["doc_id", "doc_type", "title", "excerpt", "chunk_index"],
with_vectors=False, # Don't need vectors
)
all_points.extend(scroll_result)
if next_offset is None:
break
offset = next_offset
logger.debug(f"Retrieved {len(all_points)} points from Qdrant for fuzzy search")
# Deduplicate by (doc_id, doc_type) - keep first chunk
seen_docs = {}
for point in all_points:
doc_id = int(point.payload["doc_id"])
dtype = point.payload.get("doc_type", "note")
doc_key = (doc_id, dtype)
chunk_idx = point.payload.get("chunk_index", 0)
if doc_key not in seen_docs or chunk_idx == 0:
seen_docs[doc_key] = point
logger.debug(f"Deduplicated to {len(seen_docs)} unique documents")
# Score each document based on fuzzy matches
scored_results = []
query_lower = query.lower()
for doc_key, point in seen_docs.items():
doc_id, dtype = doc_key
title = point.payload.get("title", "")
excerpt = point.payload.get("excerpt", "")
# Check title match
title_score = self._calculate_char_overlap(query_lower, title.lower())
# Check excerpt match
excerpt_score = self._calculate_char_overlap(query_lower, excerpt.lower())
# Use best score
best_score = max(title_score, excerpt_score)
if best_score >= threshold:
match_location = "title" if title_score >= excerpt_score else "excerpt"
scored_results.append(
{
"doc_id": doc_id,
"doc_type": dtype,
"title": title,
"excerpt": excerpt
if excerpt_score >= title_score
else f"Title match: {title}",
"score": best_score,
"match_location": match_location,
}
)
# Sort by score (descending) and limit
scored_results.sort(key=lambda x: x["score"], reverse=True)
top_results = scored_results[:limit]
# Return unverified results (verification happens at output stage)
final_results = []
for result in top_results:
final_results.append(
SearchResult(
id=result["doc_id"],
doc_type=result["doc_type"],
title=result["title"],
excerpt=result["excerpt"],
score=result["score"],
metadata={"match_location": result["match_location"]},
)
)
logger.info(f"Fuzzy search returned {len(final_results)} unverified results")
if final_results:
result_details = [
f"{r.doc_type}_{r.id} (score={r.score:.3f}, title='{r.title}')"
for r in final_results[:5]
]
logger.debug(f"Top fuzzy results: {', '.join(result_details)}")
return final_results
def _calculate_char_overlap(self, query: str, text: str) -> float:
"""Calculate character overlap ratio between query and text.
Args:
query: Query string (normalized)
text: Text to compare (normalized)
Returns:
Overlap ratio (0.0-1.0)
"""
if not query or not text:
return 0.0
# Convert to character sets
query_chars = set(query)
text_chars = set(text)
# Calculate overlap
overlap = query_chars & text_chars
overlap_ratio = len(overlap) / len(query_chars)
return overlap_ratio
def _extract_excerpt(self, content: str, max_length: int = 200) -> str:
"""Extract excerpt from content.
Args:
content: Full document content
max_length: Maximum excerpt length
Returns:
Excerpt string
"""
if not content:
return ""
excerpt = content[:max_length].strip()
if len(content) > max_length:
excerpt += "..."
return excerpt
nextcloud_mcp_server/search/hybrid.py
-278
View File
@@ -1,278 +0,0 @@
"""Hybrid search algorithm using Reciprocal Rank Fusion (RRF)."""
import logging
from collections import defaultdict
from typing import Any
import anyio
from nextcloud_mcp_server.search.algorithms import SearchAlgorithm, SearchResult
from nextcloud_mcp_server.search.fuzzy import FuzzySearchAlgorithm
from nextcloud_mcp_server.search.keyword import KeywordSearchAlgorithm
from nextcloud_mcp_server.search.semantic import SemanticSearchAlgorithm
logger = logging.getLogger(__name__)
class HybridSearchAlgorithm(SearchAlgorithm):
"""Hybrid search combining multiple algorithms using Reciprocal Rank Fusion.
Implements RRF from ADR-003 to combine results from:
- Semantic search (vector similarity)
- Keyword search (token matching)
- Fuzzy search (character overlap)
RRF formula: score = weight / (k + rank)
where k=60 (standard value) and rank is 1-indexed position.
"""
DEFAULT_RRF_K = 60 # Standard RRF constant
def __init__(
self,
semantic_weight: float = 0.5,
keyword_weight: float = 0.3,
fuzzy_weight: float = 0.2,
rrf_k: int = DEFAULT_RRF_K,
):
"""Initialize hybrid search with algorithm weights.
Args:
semantic_weight: Weight for semantic results (default: 0.5)
keyword_weight: Weight for keyword results (default: 0.3)
fuzzy_weight: Weight for fuzzy results (default: 0.2)
rrf_k: RRF constant for rank decay (default: 60)
Raises:
ValueError: If weights are invalid
"""
# Validate weights
if semantic_weight < 0 or keyword_weight < 0 or fuzzy_weight < 0:
raise ValueError("Weights must be non-negative")
total_weight = semantic_weight + keyword_weight + fuzzy_weight
if total_weight > 1.0:
raise ValueError(f"Weights sum to {total_weight:.2f}, must be ≤1.0")
if total_weight == 0.0:
raise ValueError("At least one weight must be > 0")
self.semantic_weight = semantic_weight
self.keyword_weight = keyword_weight
self.fuzzy_weight = fuzzy_weight
self.rrf_k = rrf_k
self.total_weight = total_weight
# Initialize sub-algorithms
self.semantic = SemanticSearchAlgorithm()
self.keyword = KeywordSearchAlgorithm()
self.fuzzy = FuzzySearchAlgorithm()
@property
def name(self) -> str:
return "hybrid"
@property
def requires_vector_db(self) -> bool:
# Requires vector DB if semantic search has non-zero weight
return self.semantic_weight > 0
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 RRF to combine algorithms.
Returns unverified results from combined algorithms. Access verification
should be performed separately at the final output stage.
Args:
query: Search query
user_id: User ID for filtering
limit: Maximum results to return
doc_type: Optional document type filter
**kwargs: Additional parameters passed to sub-algorithms
Returns:
List of unverified SearchResult objects ranked by RRF combined score
"""
logger.info(
f"Hybrid search: query='{query}', user={user_id}, limit={limit}, "
f"weights=(semantic={self.semantic_weight}, keyword={self.keyword_weight}, "
f"fuzzy={self.fuzzy_weight})"
)
# Prepare algorithm configurations for parallel execution
algo_configs = []
if self.semantic_weight > 0:
algo_configs.append(
(
"semantic",
self.semantic.search,
query,
user_id,
limit * 2,
doc_type,
kwargs,
)
)
if self.keyword_weight > 0:
algo_configs.append(
(
"keyword",
self.keyword.search,
query,
user_id,
limit * 2,
doc_type,
kwargs,
)
)
if self.fuzzy_weight > 0:
algo_configs.append(
(
"fuzzy",
self.fuzzy.search,
query,
user_id,
limit * 2,
doc_type,
kwargs,
)
)
# Pre-allocate results list and extract algorithm names
results_list = [None] * len(algo_configs)
algo_names = [name for name, *_ in algo_configs]
async def search_one(
index: int,
search_func,
query_arg: str,
user_id_arg: str,
limit_arg: int,
doc_type_arg: str | None,
kwargs_arg: dict,
):
"""Execute one search algorithm and store result at index."""
result = await search_func(
query_arg, user_id_arg, limit_arg, doc_type_arg, **kwargs_arg
)
results_list[index] = result
# Execute searches in parallel using anyio task group
async with anyio.create_task_group() as tg:
for idx, (name, search_func, q, uid, lim, dt, kw) in enumerate(
algo_configs
):
tg.start_soon(search_one, idx, search_func, q, uid, lim, dt, kw)
# Build results dict
algo_results = {}
for algo_name, results in zip(algo_names, results_list):
algo_results[algo_name] = results
logger.debug(f"{algo_name} returned {len(results)} results")
# Combine using RRF
combined_results = self._reciprocal_rank_fusion(
algo_results,
{
"semantic": self.semantic_weight,
"keyword": self.keyword_weight,
"fuzzy": self.fuzzy_weight,
},
limit,
)
logger.info(f"Hybrid search returned {len(combined_results)} combined results")
if combined_results:
result_details = [
f"{r.doc_type}_{r.id} (score={r.score:.3f}, title='{r.title}')"
for r in combined_results[:5]
]
logger.debug(f"Top hybrid results: {', '.join(result_details)}")
return combined_results
def _reciprocal_rank_fusion(
self,
algo_results: dict[str, list[SearchResult]],
weights: dict[str, float],
limit: int,
) -> list[SearchResult]:
"""Combine multiple ranked result lists using RRF.
Args:
algo_results: Dict of algorithm_name -> ranked results
weights: Dict of algorithm_name -> weight (0-1)
limit: Maximum results to return
Returns:
Combined and re-ranked results
"""
# Track RRF scores per document
rrf_scores: dict[tuple[int, str], float] = defaultdict(float)
# Track best result object for each document
best_results: dict[tuple[int, str], SearchResult] = {}
for algo_name, results in algo_results.items():
weight = weights.get(algo_name, 0.0)
if weight == 0:
continue
for rank, result in enumerate(results, start=1):
doc_key = (result.id, result.doc_type)
# RRF formula: weight / (k + rank)
rrf_score = weight / (self.rrf_k + rank)
rrf_scores[doc_key] += rrf_score
# Track best result object (prefer higher original scores)
if doc_key not in best_results:
best_results[doc_key] = result
elif result.score > best_results[doc_key].score:
best_results[doc_key] = result
# Sort by combined RRF score
sorted_docs = sorted(
rrf_scores.items(),
key=lambda x: x[1],
reverse=True,
)[:limit]
# Calculate normalization factor to scale RRF scores to 0-1 range
# Theoretical max RRF score = total_weight / (rrf_k + 1)
# Normalization factor = (rrf_k + 1) / total_weight
normalization_factor = (self.rrf_k + 1) / self.total_weight
# Build final results with normalized RRF scores
final_results = []
for doc_key, rrf_score in sorted_docs:
result = best_results[doc_key]
# Normalize RRF score to 0-1 range for better user comprehension
normalized_score = rrf_score * normalization_factor
# Create new result with normalized score
# Keep original metadata but add RRF details
metadata = result.metadata or {}
metadata["rrf_score_raw"] = rrf_score # Original RRF score
metadata["original_score"] = result.score # Original algorithm score
metadata["normalization_factor"] = normalization_factor
final_results.append(
SearchResult(
id=result.id,
doc_type=result.doc_type,
title=result.title,
excerpt=result.excerpt,
score=normalized_score, # Use normalized score (0-1 range)
metadata=metadata,
)
)
return final_results
nextcloud_mcp_server/search/keyword.py
-277
View File
@@ -1,277 +0,0 @@
"""Keyword search algorithm using token-based matching on Qdrant payload (ADR-001)."""
import logging
from typing import Any
from qdrant_client.models import FieldCondition, Filter, MatchValue
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.search.algorithms import SearchAlgorithm, SearchResult
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
logger = logging.getLogger(__name__)
class KeywordSearchAlgorithm(SearchAlgorithm):
"""Keyword search using token-based matching with weighted scoring.
Implements token-based search from ADR-001:
- Title matches weighted 3x higher than content matches
- Case-insensitive token matching
- Relevance scoring based on match frequency and location
"""
# Weighting constants from ADR-001
TITLE_WEIGHT = 3.0
CONTENT_WEIGHT = 1.0
@property
def name(self) -> str:
return "keyword"
async def search(
self,
query: str,
user_id: str,
limit: int = 10,
doc_type: str | None = None,
**kwargs: Any,
) -> list[SearchResult]:
"""Execute keyword search using token matching on Qdrant payload.
Queries Qdrant for all indexed documents, then scores based on token
matches in title and excerpt fields. Returns unverified results - access
verification should be performed separately at the final output stage.
Args:
query: Search query to tokenize and match
user_id: User ID for filtering
limit: Maximum results to return
doc_type: Optional document type filter (None = all types)
**kwargs: Additional parameters (unused)
Returns:
List of unverified SearchResult objects ranked by keyword match score
"""
settings = get_settings()
logger.info(
f"Keyword search: query='{query}', user={user_id}, "
f"limit={limit}, doc_type={doc_type}"
)
# Tokenize query
query_tokens = self._process_query(query)
logger.debug(f"Query tokens: {query_tokens}")
# Build Qdrant filter
filter_conditions = [
FieldCondition(key="user_id", match=MatchValue(value=user_id))
]
if doc_type:
filter_conditions.append(
FieldCondition(key="doc_type", match=MatchValue(value=doc_type))
)
# Scroll through Qdrant to get all matching documents
# We need title and excerpt from payload for token matching
qdrant_client = await get_qdrant_client()
collection = settings.get_collection_name()
all_points = []
offset = None
# Scroll through all points matching filter
while True:
scroll_result, next_offset = await qdrant_client.scroll(
collection_name=collection,
scroll_filter=Filter(must=filter_conditions),
limit=100, # Batch size
offset=offset,
with_payload=[
"doc_id",
"doc_type",
"title",
"excerpt",
"chunk_index",
"total_chunks",
],
with_vectors=False, # Don't need vectors for keyword search
)
all_points.extend(scroll_result)
if next_offset is None:
break
offset = next_offset
logger.debug(
f"Retrieved {len(all_points)} points from Qdrant for keyword search"
)
# Deduplicate by (doc_id, doc_type) - keep best chunk per document
seen_docs = {}
for point in all_points:
doc_id = int(point.payload["doc_id"])
dtype = point.payload.get("doc_type", "note")
doc_key = (doc_id, dtype)
# Keep first chunk (chunk_index=0) as it has the most relevant content
chunk_idx = point.payload.get("chunk_index", 0)
if doc_key not in seen_docs or chunk_idx == 0:
seen_docs[doc_key] = point
logger.debug(f"Deduplicated to {len(seen_docs)} unique documents")
# Score each document based on keyword matches
scored_results = []
for doc_key, point in seen_docs.items():
doc_id, dtype = doc_key
title = point.payload.get("title", "")
excerpt = point.payload.get("excerpt", "")
# Calculate keyword match score
score = self._calculate_score(query_tokens, title, excerpt)
if score > 0: # Only include matches
scored_results.append(
{
"doc_id": doc_id,
"doc_type": dtype,
"title": title,
"excerpt": excerpt,
"score": score,
}
)
# Sort by score (descending) and limit
scored_results.sort(key=lambda x: x["score"], reverse=True)
top_results = scored_results[:limit]
# Return unverified results (verification happens at output stage)
final_results = []
for result in top_results:
final_results.append(
SearchResult(
id=result["doc_id"],
doc_type=result["doc_type"],
title=result["title"],
excerpt=result["excerpt"],
score=result["score"],
metadata={},
)
)
logger.info(f"Keyword search returned {len(final_results)} unverified results")
if final_results:
result_details = [
f"{r.doc_type}_{r.id} (score={r.score:.3f}, title='{r.title}')"
for r in final_results[:5]
]
logger.debug(f"Top keyword results: {', '.join(result_details)}")
return final_results
def _process_query(self, query: str) -> list[str]:
"""Tokenize and normalize query.
Args:
query: Raw query string
Returns:
List of normalized tokens
"""
# Convert to lowercase and split into tokens
tokens = query.lower().split()
# Filter out very short tokens (optional)
tokens = [token for token in tokens if len(token) > 1]
return tokens
def _calculate_score(
self, query_tokens: list[str], title: str, content: str
) -> float:
"""Calculate relevance score based on token matches.
Args:
query_tokens: List of query tokens
title: Document title
content: Document content
Returns:
Relevance score (0.0-1.0)
"""
if not query_tokens:
return 0.0
# Process title and content
title_tokens = title.lower().split()
content_tokens = content.lower().split()
score = 0.0
# Count matches in title
title_matches = sum(1 for qt in query_tokens if qt in title_tokens)
if query_tokens: # Avoid division by zero
title_match_ratio = title_matches / len(query_tokens)
score += self.TITLE_WEIGHT * title_match_ratio
# Count matches in content
content_matches = sum(1 for qt in query_tokens if qt in content_tokens)
if query_tokens:
content_match_ratio = content_matches / len(query_tokens)
score += self.CONTENT_WEIGHT * content_match_ratio
# Normalize score to 0-1 range
# Max score would be TITLE_WEIGHT + CONTENT_WEIGHT if all tokens match everywhere
max_score = self.TITLE_WEIGHT + self.CONTENT_WEIGHT
normalized_score = min(score / max_score, 1.0)
return normalized_score
def _extract_excerpt(
self, content: str, query_tokens: list[str], max_length: int = 200
) -> str:
"""Extract excerpt showing match context.
Args:
content: Full document content
query_tokens: Query tokens to find
max_length: Maximum excerpt length in characters
Returns:
Excerpt string with context around matches
"""
if not content:
return ""
content_lower = content.lower()
# Find first occurrence of any query token
first_match_pos = -1
for token in query_tokens:
pos = content_lower.find(token)
if pos != -1:
if first_match_pos == -1 or pos < first_match_pos:
first_match_pos = pos
if first_match_pos == -1:
# No matches found, return beginning
return content[:max_length].strip() + (
"..." if len(content) > max_length else ""
)
# Extract context around match
start = max(0, first_match_pos - max_length // 2)
end = min(len(content), first_match_pos + max_length // 2)
excerpt = content[start:end].strip()
# Add ellipsis if truncated
if start > 0:
excerpt = "..." + excerpt
if end < len(content):
excerpt = excerpt + "..."
return excerpt
nextcloud_mcp_server/server/semantic.py
+29 -61
View File
@@ -1,7 +1,6 @@
"""Semantic search MCP tools using vector database."""
import logging
from typing import Literal
from httpx import RequestError
from mcp.server.fastmcp import Context, FastMCP
@@ -25,12 +24,7 @@ from nextcloud_mcp_server.models.semantic import (
from nextcloud_mcp_server.observability.metrics import (
instrument_tool,
)
from nextcloud_mcp_server.search import (
FuzzySearchAlgorithm,
HybridSearchAlgorithm,
KeywordSearchAlgorithm,
SemanticSearchAlgorithm,
)
from nextcloud_mcp_server.search.bm25_hybrid import BM25HybridSearchAlgorithm
logger = logging.getLogger(__name__)
@@ -46,36 +40,30 @@ def configure_semantic_tools(mcp: FastMCP):
ctx: Context,
limit: int = 10,
doc_types: list[str] | None = None,
score_threshold: float = 0.7,
algorithm: Literal["semantic", "keyword", "fuzzy", "hybrid"] = "hybrid",
semantic_weight: float = 0.5,
keyword_weight: float = 0.3,
fuzzy_weight: float = 0.2,
score_threshold: float = 0.0,
) -> SemanticSearchResponse:
"""
Search Nextcloud content using configurable algorithms with cross-app support.
Search Nextcloud content using BM25 hybrid search with cross-app support.
Supports multiple search algorithms with client-configurable weighting:
- semantic: Vector similarity search (requires VECTOR_SYNC_ENABLED=true)
- keyword: Token-based matching (title matches weighted 3x)
- fuzzy: Character overlap matching (typo-tolerant)
- hybrid: Combines all algorithms using Reciprocal Rank Fusion (default)
Uses Qdrant's native hybrid search combining:
- Dense semantic vectors: For conceptual similarity and natural language queries
- BM25 sparse vectors: For precise keyword matching, acronyms, and specific terms
Document types are queried from the vector database to determine what's
actually indexed. Currently only "note" documents are fully supported.
Results are automatically fused using Reciprocal Rank Fusion (RRF) in the
database for optimal relevance. This provides the best of both semantic
understanding and keyword precision.
Requires VECTOR_SYNC_ENABLED=true. Currently only "note" documents are
fully supported for indexing.
Args:
query: Natural language search query
query: Natural language or keyword search query
limit: Maximum number of results to return (default: 10)
doc_types: Document types to search (e.g., ["note", "file"]). None = search all indexed types (default)
score_threshold: Minimum similarity score for semantic/hybrid (0-1, default: 0.7)
algorithm: Search algorithm to use (default: "hybrid")
semantic_weight: Weight for semantic results in hybrid mode (default: 0.5)
keyword_weight: Weight for keyword results in hybrid mode (default: 0.3)
fuzzy_weight: Weight for fuzzy results in hybrid mode (default: 0.2)
score_threshold: Minimum RRF fusion score (0-1, default: 0.0 for RRF scoring)
Returns:
SemanticSearchResponse with matching documents and relevance scores
SemanticSearchResponse with matching documents ranked by RRF fusion scores
"""
from nextcloud_mcp_server.config import get_settings
@@ -84,42 +72,22 @@ def configure_semantic_tools(mcp: FastMCP):
username = client.username
logger.info(
f"Search: query='{query}', user={username}, algorithm={algorithm}, "
f"BM25 hybrid search: query='{query}', user={username}, "
f"limit={limit}, score_threshold={score_threshold}"
)
# Check that vector sync is enabled
if not settings.vector_sync_enabled:
raise McpError(
ErrorData(
code=-1,
message="BM25 hybrid search requires VECTOR_SYNC_ENABLED=true",
)
)
try:
# Create appropriate algorithm instance
if algorithm == "semantic":
if not settings.vector_sync_enabled:
raise McpError(
ErrorData(
code=-1,
message="Semantic search requires VECTOR_SYNC_ENABLED=true",
)
)
search_algo = SemanticSearchAlgorithm(score_threshold=score_threshold)
elif algorithm == "keyword":
search_algo = KeywordSearchAlgorithm()
elif algorithm == "fuzzy":
search_algo = FuzzySearchAlgorithm()
elif algorithm == "hybrid":
if semantic_weight > 0 and not settings.vector_sync_enabled:
raise McpError(
ErrorData(
code=-1,
message="Hybrid search with semantic component requires VECTOR_SYNC_ENABLED=true",
)
)
search_algo = HybridSearchAlgorithm(
semantic_weight=semantic_weight,
keyword_weight=keyword_weight,
fuzzy_weight=fuzzy_weight,
)
else:
raise McpError(
ErrorData(code=-1, message=f"Unknown algorithm: {algorithm}")
)
# Create BM25 hybrid search algorithm
search_algo = BM25HybridSearchAlgorithm(score_threshold=score_threshold)
# Execute search across requested document types
# If doc_types is None, search all indexed types (cross-app search)
@@ -179,13 +147,13 @@ def configure_semantic_tools(mcp: FastMCP):
)
)
logger.info(f"Returning {len(results)} results from {algorithm} search")
logger.info(f"Returning {len(results)} results from BM25 hybrid search")
return SemanticSearchResponse(
results=results,
query=query,
total_found=len(results),
search_method=algorithm,
search_method="bm25_hybrid",
)
except ValueError as e:
nextcloud_mcp_server/vector/processor.py
+14 -5
View File
@@ -15,7 +15,7 @@ from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.embedding import get_embedding_service
from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service
from nextcloud_mcp_server.observability.metrics import (
record_qdrant_operation,
record_vector_sync_processing,
@@ -233,15 +233,21 @@ async def _index_document(
)
chunks = chunker.chunk_text(content)
# Generate embeddings (I/O bound - external API call)
# Generate dense embeddings (I/O bound - external API call)
embedding_service = get_embedding_service()
embeddings = await embedding_service.embed_batch(chunks)
dense_embeddings = await embedding_service.embed_batch(chunks)
# Generate sparse embeddings (BM25 for keyword matching)
bm25_service = get_bm25_service()
sparse_embeddings = bm25_service.encode_batch(chunks)
# Prepare Qdrant points
indexed_at = int(time.time())
points = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
for i, (chunk, dense_emb, sparse_emb) in enumerate(
zip(chunks, dense_embeddings, sparse_embeddings)
):
# Generate deterministic UUID for point ID
# Using uuid5 with DNS namespace and combining doc info
point_name = f"{doc_task.doc_type}:{doc_task.doc_id}:chunk:{i}"
@@ -250,7 +256,10 @@ async def _index_document(
points.append(
PointStruct(
id=point_id,
vector=embedding,
vector={
"dense": dense_emb,
"sparse": sparse_emb,
},
payload={
"user_id": doc_task.user_id,
"doc_id": doc_task.doc_id,
nextcloud_mcp_server/vector/qdrant_client.py
+24 -9
View File
@@ -2,7 +2,7 @@
import logging
from qdrant_client import AsyncQdrantClient
from qdrant_client import AsyncQdrantClient, models
from qdrant_client.models import Distance, VectorParams
from nextcloud_mcp_server.config import get_settings
@@ -84,7 +84,12 @@ async def get_qdrant_client() -> AsyncQdrantClient:
f"Collection '{collection_name}' found, validating dimensions..."
)
collection_info = await _qdrant_client.get_collection(collection_name)
actual_dimension = collection_info.config.params.vectors.size
# Handle both named vectors (dict) and legacy single vector
vectors = collection_info.config.params.vectors
if isinstance(vectors, dict):
actual_dimension = vectors["dense"].size
else:
actual_dimension = vectors.size
# Validate dimension matches
if actual_dimension != expected_dimension:
@@ -112,17 +117,27 @@ async def get_qdrant_client() -> AsyncQdrantClient:
)
await _qdrant_client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(
size=expected_dimension,
distance=Distance.COSINE,
),
vectors_config={
"dense": VectorParams(
size=expected_dimension,
distance=Distance.COSINE,
),
},
sparse_vectors_config={
"sparse": models.SparseVectorParams(
index=models.SparseIndexParams(
on_disk=False,
)
),
},
)
logger.info(
f"Created Qdrant collection: {collection_name}\n"
f" Dimension: {expected_dimension}\n"
f" Model: {settings.ollama_embedding_model}\n"
f" Dense vector dimension: {expected_dimension}\n"
f" Dense embedding model: {settings.ollama_embedding_model}\n"
f" Sparse vectors: BM25 (for hybrid search)\n"
f" Distance: COSINE\n"
f"Background sync will index all documents with this embedding model."
f"Background sync will index all documents with dense + sparse vectors."
)
return _qdrant_client