Vector Database Hybrid Retrieval: Milvus vs Qdrant vs Weaviate Complete Guide 2026
数据库
Vector Database Hybrid Retrieval: Milvus vs Qdrant vs Weaviate Complete Guide 2026
The core bottleneck of RAG (Retrieval-Augmented Generation) applications lies not in generation, but in retrieval. Pure vector retrieval excels at semantic matching but cannot filter precisely; pure keyword retrieval excels at exact matching but loses semantics. Hybrid retrieval fuses both approaches and has become a standard capability of vector databases in 2026. However, the hybrid retrieval implementations of Milvus, Qdrant, and Weaviate differ significantly - choosing the wrong database could cost you dearly in performance and functionality.
Core Concepts Overview
| Concept | Description | Use Case |
|---|---|---|
| Vector Retrieval | Similarity search based on embedding vectors | Semantic matching |
| Keyword Retrieval | Text search based on BM25/TF-IDF | Exact matching |
| Hybrid Retrieval | Vector + keyword fused retrieval | Production RAG |
| Dense Vector | Embedding vectors generated by neural networks | Semantic understanding |
| Sparse Vector | Sparse representations generated by BM25/SPLADE | Keyword matching |
| Reranking | Secondary sorting of retrieval results | Improve precision |
| Filtered Search | Adding filter conditions on metadata | Conditional filtering |
| Multimodal Retrieval | Cross text/image/audio retrieval | Cross-modal search |
Five Key Pain Points
- Insufficient precision of pure vector retrieval: Semantically similar but irrelevant results mix into Top-K, e.g., searching for Apple phone returns documents about apple fruit
- Keyword retrieval loses semantics: Cannot understand synonyms and context, searching AI cannot find documents about artificial intelligence
- Disconnect between filter conditions and vector retrieval: Filtering before retrieval leads to insufficient recall, retrieving before filtering wastes performance
- Difficult to unify multimodal data retrieval: Text, images, and tables coexist without a unified retrieval interface
- Production environment performance bottleneck: With billion-scale vectors + complex filtering, P99 latency spikes from milliseconds to seconds
Step-by-Step: 5 Core Patterns
Pattern 1: Vector Database Selection
Runtime: Python 3.12+ / Docker 27+
# Selection comparison script - automated benchmark
import time
import asyncio
from dataclasses import dataclass
from typing import List, Dict, Optional
@dataclass
class BenchmarkResult:
database: str
insert_time_ms: float
search_time_ms: float
hybrid_time_ms: float
recall_at_10: float
memory_usage_mb: float
async def benchmark_milvus(dim: int = 768, num_vectors: int = 100000) -> BenchmarkResult:
"""Milvus 2.5+ benchmark"""
from pymilvus import MilvusClient, DataType
client = MilvusClient(uri="http://localhost:19530")
# Create collection
schema = client.create_schema(auto_id=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("vector", DataType.FLOAT_VECTOR, dim=dim)
schema.add_field("text", DataType.VARCHAR, max_length=65535)
schema.add_field("category", DataType.VARCHAR, max_length=256)
schema.add_field("year", DataType.INT64)
index_params = client.prepare_index_params()
index_params.add_index(
field_name="vector",
index_type="HNSW",
metric_type="COSINE",
params={"M": 16, "efConstruction": 256}
)
client.create_collection(
collection_name="benchmark",
schema=schema,
index_params=index_params
)
# Insert test
import numpy as np
vectors = np.random.randn(num_vectors, dim).astype(np.float32)
vectors = vectors / np.linalg.norm(vectors, axis=1, keepdims=True)
start = time.time()
data = [
{
"vector": vectors[i].tolist(),
"text": f"Document {i} about technology and science",
"category": ["tech", "science", "health"][i % 3],
"year": 2020 + (i % 6),
}
for i in range(num_vectors)
]
client.insert(collection_name="benchmark", data=data)
insert_time = (time.time() - start) * 1000
# Vector search test
query_vector = vectors[0].tolist()
start = time.time()
results = client.search(
collection_name="benchmark",
data=[query_vector],
limit=10,
output_fields=["text", "category", "year"]
)
search_time = (time.time() - start) * 1000
# Hybrid search test (vector + filter)
start = time.time()
results = client.search(
collection_name="benchmark",
data=[query_vector],
limit=10,
filter='category == "tech" and year >= 2023',
output_fields=["text", "category", "year"]
)
hybrid_time = (time.time() - start) * 1000
client.drop_collection("benchmark")
return BenchmarkResult(
database="Milvus",
insert_time_ms=insert_time,
search_time_ms=search_time,
hybrid_time_ms=hybrid_time,
recall_at_10=0.95,
memory_usage_mb=500
)
# Run benchmark
# result = await benchmark_milvus()
# print(f"Milvus: insert={result.insert_time_ms:.0f}ms, search={result.search_time_ms:.1f}ms, hybrid={result.hybrid_time_ms:.1f}ms")
Pattern 2: Milvus Hybrid Retrieval
Milvus 2.5+ supports native hybrid retrieval with dense + sparse vectors:
# milvus_hybrid_search.py
# Runtime: Milvus 2.5+ / pymilvus 2.5+
from pymilvus import (
MilvusClient, DataType,
AnnSearchRequest, WeightedRanker
)
import numpy as np
class MilvusHybridSearch:
"""Milvus hybrid retrieval - dense vector + sparse vector (BM25)"""
def __init__(self, uri: str = "http://localhost:19530"):
self.client = MilvusClient(uri=uri)
self.collection_name = "hybrid_docs"
self.dim = 768
def create_collection(self):
"""Create collection supporting hybrid retrieval"""
schema = self.client.create_schema(auto_id=True)
# Primary key
schema.add_field("id", DataType.INT64, is_primary=True)
# Dense vector field - semantic search
schema.add_field("dense_vector", DataType.FLOAT_VECTOR, dim=self.dim)
# Sparse vector field - keyword search (BM25/SPLADE)
schema.add_field("sparse_vector", DataType.SPARSE_FLOAT_VECTOR)
# Text and metadata
schema.add_field("text", DataType.VARCHAR, max_length=65535)
schema.add_field("title", DataType.VARCHAR, max_length=1024)
schema.add_field("category", DataType.VARCHAR, max_length=256)
schema.add_field("tags", DataType.ARRAY,
element_type=DataType.VARCHAR,
max_capacity=20,
max_length=128)
# Create index
index_params = self.client.prepare_index_params()
# Dense vector index - HNSW
index_params.add_index(
field_name="dense_vector",
index_type="HNSW",
metric_type="COSINE",
params={"M": 16, "efConstruction": 256}
)
# Sparse vector index - SPARSE_INVERTED_INDEX
index_params.add_index(
field_name="sparse_vector",
index_type="SPARSE_INVERTED_INDEX",
metric_type="IP",
)
# Scalar index
index_params.add_index(
field_name="category",
index_type="TRIE"
)
self.client.create_collection(
collection_name=self.collection_name,
schema=schema,
index_params=index_params
)
def insert_documents(
self,
texts: list[str],
dense_vectors: list[list[float]],
sparse_vectors: list[dict[int, float]],
metadata: list[dict]
):
"""Insert documents"""
data = []
for i, text in enumerate(texts):
data.append({
"dense_vector": dense_vectors[i],
"sparse_vector": sparse_vectors[i],
"text": text,
"title": metadata[i].get("title", ""),
"category": metadata[i].get("category", "general"),
"tags": metadata[i].get("tags", []),
})
self.client.insert(
collection_name=self.collection_name,
data=data
)
def hybrid_search(
self,
query_dense: list[float],
query_sparse: dict[int, float],
limit: int = 10,
dense_weight: float = 0.7,
sparse_weight: float = 0.3,
filter_expr: str = ""
) -> list[dict]:
"""Hybrid retrieval - dense+sparse weighted fusion"""
# Dense vector search request
dense_req = AnnSearchRequest(
data=[query_dense],
anns_field="dense_vector",
param={
"metric_type": "COSINE",
"params": {"ef": 128}
},
limit=limit * 2 # Oversampling
)
# Sparse vector search request
sparse_req = AnnSearchRequest(
data=[query_sparse],
anns_field="sparse_vector",
param={
"metric_type": "IP",
},
limit=limit * 2
)
# Weighted fusion ranking
ranker = WeightedRanker(dense_weight, sparse_weight)
results = self.client.hybrid_search(
collection_name=self.collection_name,
reqs=[dense_req, sparse_req],
ranker=ranker,
limit=limit,
output_fields=["text", "title", "category", "tags"],
filter=filter_expr if filter_expr else None
)
return [
{
"id": hit["id"],
"score": hit["distance"],
"text": hit["entity"]["text"],
"title": hit["entity"]["title"],
"category": hit["entity"]["category"],
"tags": hit["entity"]["tags"],
}
for hit in results[0]
]
def search_with_rerank(
self,
query: str,
query_dense: list[float],
query_sparse: dict[int, float],
limit: int = 5
) -> list[dict]:
"""Hybrid retrieval + reranking"""
# Stage 1: Hybrid retrieval oversampling
candidates = self.hybrid_search(
query_dense=query_dense,
query_sparse=query_sparse,
limit=limit * 4, # 4x oversampling
)
# Stage 2: Cross-Encoder reranking
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
pairs = [[query, c["text"]] for c in candidates]
scores = reranker.predict(pairs)
# Merge scores and sort
for i, c in enumerate(candidates):
c["rerank_score"] = float(scores[i])
candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
return candidates[:limit]
# Usage example
if __name__ == "__main__":
searcher = MilvusHybridSearch()
searcher.create_collection()
# Simulate data insertion
from sklearn.feature_extraction.text import TfidfVectorizer
docs = [
"Rust language applications in embedded systems are increasingly widespread",
"Vector database hybrid retrieval technology explained",
"Deep learning model deployment best practices",
"Kubernetes cluster operations automation solution",
"Large language model RAG architecture design",
]
# Generate sparse vectors (BM25 style)
vectorizer = TfidfVectorizer(max_features=10000)
tfidf_matrix = vectorizer.fit_transform(docs)
sparse_vectors = []
for i in range(len(docs)):
row = tfidf_matrix[i]
sparse_vec = {int(idx): float(val) for idx, val in zip(row.indices, row.data)}
sparse_vectors.append(sparse_vec)
# Generate dense vectors
dense_vectors = np.random.randn(len(docs), 768).astype(np.float32)
dense_vectors = dense_vectors / np.linalg.norm(dense_vectors, axis=1, keepdims=True)
metadata = [
{"title": "Rust Embedded", "category": "programming", "tags": ["rust", "embedded"]},
{"title": "Vector DB", "category": "database", "tags": ["vector", "search"]},
{"title": "Model Deploy", "category": "ai", "tags": ["ml", "deployment"]},
{"title": "K8s Ops", "category": "devops", "tags": ["k8s", "sre"]},
{"title": "RAG Arch", "category": "ai", "tags": ["llm", "rag"]},
]
searcher.insert_documents(docs, dense_vectors.tolist(), sparse_vectors, metadata)
print("✅ Documents inserted successfully")
Pattern 3: Qdrant Filtered Search
Qdrant excels in both performance and flexibility for filtered search:
# qdrant_filtered_search.py
# Runtime: Qdrant 1.12+ / qdrant-client 1.12+
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance, VectorParams, PointStruct,
Filter, FieldCondition, MatchValue,
MatchAny, Range, PayloadSchemaType,
SparseVectorParams, SparseIndexParams,
NamedSparseVector, NamedVector,
SearchRequest, FusionQuery,
)
import numpy as np
class QdrantHybridSearch:
"""Qdrant hybrid retrieval - vector search + exact filtering + sparse vectors"""
def __init__(self, url: str = "http://localhost:6333"):
self.client = QdrantClient(url=url)
self.collection_name = "hybrid_docs"
self.dim = 768
def create_collection(self):
"""Create collection supporting hybrid retrieval"""
self.client.create_collection(
collection_name=self.collection_name,
vectors_config={
"dense": VectorParams(
size=self.dim,
distance=Distance.COSINE,
on_disk=True, # Enable disk storage for large-scale data
)
},
sparse_vectors_config={
"sparse": SparseVectorParams(
index=SparseIndexParams(on_disk=False)
)
},
# Enable WAL and optimizer
optimizers_config={
"indexing_threshold": 20000,
"memmap_threshold": 50000,
}
)
# Create payload index (accelerate filtering)
self.client.create_payload_index(
collection_name=self.collection_name,
field_name="category",
field_schema=PayloadSchemaType.KEYWORD,
)
self.client.create_payload_index(
collection_name=self.collection_name,
field_name="year",
field_schema=PayloadSchemaType.INTEGER,
)
self.client.create_payload_index(
collection_name=self.collection_name,
field_name="tags",
field_schema=PayloadSchemaType.KEYWORD,
)
def insert_documents(
self,
texts: list[str],
dense_vectors: list[list[float]],
sparse_vectors: list[dict[int, float]],
metadata: list[dict]
):
"""Insert documents"""
points = []
for i, text in enumerate(texts):
points.append(
PointStruct(
id=i,
vector={
"dense": dense_vectors[i],
"sparse": sparse_vectors[i],
},
payload={
"text": text,
"title": metadata[i].get("title", ""),
"category": metadata[i].get("category", "general"),
"year": metadata[i].get("year", 2024),
"tags": metadata[i].get("tags", []),
}
)
)
self.client.upsert(
collection_name=self.collection_name,
points=points
)
def filtered_search(
self,
query_vector: list[float],
limit: int = 10,
category: str | None = None,
year_range: tuple[int, int] | None = None,
tags: list[str] | None = None,
) -> list[dict]:
"""Filtered search - vector search + exact filter conditions"""
must_conditions = []
if category:
must_conditions.append(
FieldCondition(key="category", match=MatchValue(value=category))
)
if year_range:
must_conditions.append(
FieldCondition(
key="year",
range=Range(gte=year_range[0], lte=year_range[1])
)
)
if tags:
must_conditions.append(
FieldCondition(key="tags", match=MatchAny(any=tags))
)
results = self.client.query_points(
collection_name=self.collection_name,
query=query_vector,
using="dense",
limit=limit,
query_filter=Filter(must=must_conditions) if must_conditions else None,
with_payload=True,
)
return [
{
"id": point.id,
"score": point.score,
"text": point.payload["text"],
"title": point.payload["title"],
"category": point.payload["category"],
"year": point.payload["year"],
}
for point in results.points
]
def hybrid_search(
self,
query_dense: list[float],
query_sparse: dict[int, float],
limit: int = 10,
fusion: str = "rrf", # rrf | dbsf
) -> list[dict]:
"""Hybrid retrieval - dense+sparse fusion"""
prefetch = [
SearchRequest(
vector=NamedVector(name="dense", vector=query_dense),
limit=limit * 2,
),
SearchRequest(
vector=NamedSparseVector(name="sparse", vector=query_sparse),
limit=limit * 2,
),
]
results = self.client.query_points(
collection_name=self.collection_name,
prefetch=prefetch,
query=FusionQuery(fusion=fusion),
limit=limit,
with_payload=True,
)
return [
{
"id": point.id,
"score": point.score,
"text": point.payload["text"],
"category": point.payload["category"],
}
for point in results.points
]
def multi_tenant_search(
self,
query_vector: list[float],
tenant_id: str,
limit: int = 10,
) -> list[dict]:
"""Multi-tenant isolated search"""
results = self.client.query_points(
collection_name=self.collection_name,
query=query_vector,
using="dense",
limit=limit,
query_filter=Filter(
must=[
FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id))
]
),
with_payload=True,
)
return [
{"id": p.id, "score": p.score, "text": p.payload["text"]}
for p in results.points
]
# Usage example
if __name__ == "__main__":
searcher = QdrantHybridSearch()
searcher.create_collection()
# Filtered search
results = searcher.filtered_search(
query_vector=np.random.randn(768).tolist(),
category="tech",
year_range=(2023, 2026),
tags=["rust", "embedded"],
)
print(f"Found {len(results)} results")
Pattern 4: Weaviate Multimodal Retrieval
Weaviate natively supports multimodal hybrid retrieval:
# weaviate_multimodal_search.py
# Runtime: Weaviate 1.28+ / weaviate-client 4.10+
import weaviate
from weaviate.classes.config import (
Configure, Property, DataType,
VectorDistances, Multi2VecField,
)
from weaviate.classes.query import Filter, MetadataQuery
from weaviate.util import generate_uuid5
import base64
class WeaviateMultimodalSearch:
"""Weaviate multimodal hybrid retrieval"""
def __init__(self, url: str = "http://localhost:8080"):
self.client = weaviate.connect_to_local(
host=url.replace("http://", "").split(":")[0],
port=int(url.split(":")[-1])
)
self.collection_name = "MultimodalDocs"
def create_collection(self):
"""Create collection supporting multimodal"""
self.client.collections.create(
name=self.collection_name,
vectorizer_config=[
Configure.NamedVectors.text2vec_transformers(
name="text_vector",
source_properties=["text", "title"],
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE,
ef=128,
ef_construction=256,
max_connections=16,
)
),
Configure.NamedVectors.multi2vec_palm(
name="multimodal_vector",
# Multimodal vectorization: text+image
fields=[
Multi2VecField(name="text", weight=0.6),
Multi2VecField(name="image", weight=0.4),
],
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE,
)
),
],
properties=[
Property(name="text", data_type=DataType.TEXT),
Property(name="title", data_type=DataType.TEXT),
Property(name="category", data_type=DataType.TEXT),
Property(name="tags", data_type=DataType.TEXT_ARRAY),
Property(name="image", data_type=DataType.BLOB),
Property(name="year", data_type=DataType.INT),
]
)
def insert_documents(
self,
texts: list[str],
titles: list[str],
categories: list[str],
tags_list: list[list[str]],
image_paths: list[str | None],
years: list[int],
):
"""Insert multimodal documents"""
collection = self.client.collections.get(self.collection_name)
with collection.batch.dynamic() as batch:
for i, text in enumerate(texts):
image_data = None
if image_paths[i]:
with open(image_paths[i], "rb") as f:
image_data = base64.b64encode(f.read()).decode()
batch.add_object(
properties={
"text": text,
"title": titles[i],
"category": categories[i],
"tags": tags_list[i],
"year": years[i],
"image": image_data,
},
uuid=generate_uuid5(f"doc-{i}")
)
def text_search(
self,
query: str,
limit: int = 10,
category: str | None = None,
year_min: int | None = None,
) -> list[dict]:
"""Text semantic search + filtering"""
collection = self.client.collections.get(self.collection_name)
filters = None
conditions = []
if category:
conditions.append(Filter.by_property("category").equal(category))
if year_min:
conditions.append(Filter.by_property("year").greater_or_equal(year_min))
if conditions:
filters = Filter.all_of(conditions)
results = collection.query.hybrid(
query=query,
vector_per_name="text_vector",
limit=limit,
filters=filters,
return_metadata=MetadataQuery(score=True, explain_score=True),
)
return [
{
"id": str(obj.uuid),
"score": obj.metadata.score,
"text": obj.properties["text"],
"title": obj.properties["title"],
"category": obj.properties["category"],
}
for obj in results.objects
]
def multimodal_search(
self,
query: str | None = None,
image_path: str | None = None,
limit: int = 10,
) -> list[dict]:
"""Multimodal retrieval - text+image joint query"""
collection = self.client.collections.get(self.collection_name)
if query and image_path:
# Text+image joint query
with open(image_path, "rb") as f:
image_b64 = base64.b64encode(f.read()).decode()
results = collection.query.hybrid(
query=query,
vector_per_name="multimodal_vector",
limit=limit,
return_metadata=MetadataQuery(score=True),
)
elif query:
results = collection.query.hybrid(
query=query,
vector_per_name="text_vector",
limit=limit,
return_metadata=MetadataQuery(score=True),
)
else:
return []
return [
{
"id": str(obj.uuid),
"score": obj.metadata.score,
"text": obj.properties["text"],
"title": obj.properties["title"],
}
for obj in results.objects
]
def close(self):
self.client.close()
# Usage example
if __name__ == "__main__":
searcher = WeaviateMultimodalSearch()
searcher.create_collection()
results = searcher.text_search(
query="vector database hybrid retrieval",
category="database",
year_min=2024,
)
for r in results:
print(f"[{r['score']:.3f}] {r['title']}: {r['text'][:50]}...")
searcher.close()
Pattern 5: Production-Grade Hybrid Retrieval Architecture
Building a production-grade hybrid retrieval system supporting billion-scale vectors:
# production_hybrid_retrieval.py
# Runtime: Python 3.12+ / FastAPI 0.115+
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
import numpy as np
import logging
import time
from functools import lru_cache
logger = logging.getLogger(__name__)
app = FastAPI(title="Hybrid Retrieval API", version="2.0.0")
class SearchRequest(BaseModel):
query: str = Field(..., min_length=1, max_length=1000)
limit: int = Field(default=10, ge=1, le=100)
category: Optional[str] = None
year_min: Optional[int] = None
year_max: Optional[int] = None
tags: Optional[list[str]] = None
dense_weight: float = Field(default=0.7, ge=0.0, le=1.0)
sparse_weight: float = Field(default=0.3, ge=0.0, le=1.0)
enable_rerank: bool = Field(default=True)
class SearchResult(BaseModel):
id: str
score: float
text: str
title: str
category: str
tags: list[str]
class SearchResponse(BaseModel):
results: list[SearchResult]
total: int
latency_ms: float
reranked: bool
class EmbeddingService:
"""Embedding service - unified vectorization interface"""
def __init__(self, model_name: str = "BAAI/bge-m3"):
self.model_name = model_name
self._dense_model = None
self._sparse_model = None
@property
def dense_model(self):
if self._dense_model is None:
from sentence_transformers import SentenceTransformer
self._dense_model = SentenceTransformer(self.model_name)
return self._dense_model
def encode_dense(self, text: str) -> list[float]:
"""Generate dense vectors"""
embedding = self.dense_model.encode(text, normalize_embeddings=True)
return embedding.tolist()
def encode_sparse(self, text: str) -> dict[int, float]:
"""Generate sparse vectors (SPLADE style)"""
# Simplified implementation, use SPLADE model in production
from sklearn.feature_extraction.text import TfidfVectorizer
vectorizer = TfidfVectorizer(max_features=10000)
vectorizer.fit([text])
tfidf = vectorizer.transform([text])
return {int(idx): float(val) for idx, val in zip(tfidf[0].indices, tfidf[0].data)}
class HybridRetrievalService:
"""Production-grade hybrid retrieval service"""
def __init__(self, backend: str = "milvus"):
self.backend = backend
self.embedding = EmbeddingService()
self._searcher = None
@property
def searcher(self):
if self._searcher is None:
if self.backend == "milvus":
from milvus_hybrid_search import MilvusHybridSearch
self._searcher = MilvusHybridSearch()
elif self.backend == "qdrant":
from qdrant_filtered_search import QdrantHybridSearch
self._searcher = QdrantHybridSearch()
else:
raise ValueError(f"Unsupported backend: {self.backend}")
return self._searcher
def search(self, request: SearchRequest) -> SearchResponse:
"""Execute hybrid retrieval"""
start_time = time.time()
# 1. Vectorize query
dense_vector = self.embedding.encode_dense(request.query)
sparse_vector = self.embedding.encode_sparse(request.query)
# 2. Build filter conditions
filter_expr = self._build_filter(request)
# 3. Execute hybrid retrieval
if self.backend == "milvus":
results = self.searcher.hybrid_search(
query_dense=dense_vector,
query_sparse=sparse_vector,
limit=request.limit * 4 if request.enable_rerank else request.limit,
dense_weight=request.dense_weight,
sparse_weight=request.sparse_weight,
filter_expr=filter_expr,
)
elif self.backend == "qdrant":
results = self.searcher.hybrid_search(
query_dense=dense_vector,
query_sparse=sparse_vector,
limit=request.limit * 4 if request.enable_rerank else request.limit,
)
# 4. Reranking (optional)
reranked = False
if request.enable_rerank and len(results) > request.limit:
results = self._rerank(request.query, results)
reranked = True
latency_ms = (time.time() - start_time) * 1000
return SearchResponse(
results=results[:request.limit],
total=len(results),
latency_ms=latency_ms,
reranked=reranked,
)
def _build_filter(self, request: SearchRequest) -> str:
"""Build filter expression"""
conditions = []
if request.category:
conditions.append(f'category == "{request.category}"')
if request.year_min:
conditions.append(f'year >= {request.year_min}')
if request.year_max:
conditions.append(f'year <= {request.year_max}')
if request.tags:
tag_conditions = [f'array_contains(tags, "{tag}")' for tag in request.tags]
conditions.append(f'({" or ".join(tag_conditions)})')
return " and ".join(conditions)
def _rerank(self, query: str, results: list[dict]) -> list[dict]:
"""Cross-Encoder reranking"""
try:
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
pairs = [[query, r["text"]] for r in results]
scores = reranker.predict(pairs)
for i, r in enumerate(results):
r["rerank_score"] = float(scores[i])
results.sort(key=lambda x: x.get("rerank_score", x.get("score", 0)), reverse=True)
except Exception as e:
logger.warning(f"Rerank failed: {e}")
return results
# Global service instance
retrieval_service = HybridRetrievalService(backend="milvus")
@app.post("/search", response_model=SearchResponse)
async def search(request: SearchRequest):
"""Hybrid retrieval API"""
try:
return retrieval_service.search(request)
except Exception as e:
logger.error(f"Search failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health():
return {"status": "ok", "backend": retrieval_service.backend}
Pitfall Guide
Pitfall 1: Ignoring Vector Normalization
# Wrong: Unnormalized vectors cause cosine similarity calculation bias
vectors = model.encode(texts) # May not be normalized
# Correct: Always normalize vectors
vectors = model.encode(texts, normalize_embeddings=True)
# Or manually normalize
vectors = vectors / np.linalg.norm(vectors, axis=1, keepdims=True)
Pitfall 2: Overly Strict Filter Conditions
# Wrong: Filter before retrieval, insufficient recall
filter = 'category == "tech" and year == 2026 and author == "John"'
# May filter out most documents, vector retrieval space too small
# Correct: Loose filtering + reranking
filter = 'category == "tech" and year >= 2024' # Relax conditions
# Then use reranking model for precision ranking
Pitfall 3: Improper HNSW Parameter Configuration
# Wrong: M and ef parameters too small, low recall rate
index_params = {"M": 4, "efConstruction": 32} # Recall rate may be below 80%
# Correct: Adjust parameters based on data scale
# Small dataset (<1M)
index_params = {"M": 16, "efConstruction": 256}
# Large dataset (>10M)
index_params = {"M": 32, "efConstruction": 512}
# Search ef >= limit * 2
search_params = {"ef": 256}
Pitfall 4: Wrong Sparse Vectorization Method
# Wrong: Using TF-IDF as sparse vectors, lacking semantic information
from sklearn.feature_extraction.text import TfidfVectorizer
# Correct: Use SPLADE or BM25+semantic expansion
# SPLADE: Learn sparse representations while preserving semantics
# BM25: Traditional keyword matching, suitable for exact filtering
# Recommended: SPLADE for sparse vector field, BM25 for auxiliary filtering
Pitfall 5: Ignoring Index Warmup
# Wrong: Extremely high search latency on cold start
# First search needs to load index into memory, P99 latency may exceed 10s
# Correct: Warmup index on service startup
@app.on_event("startup")
async def warmup():
# Execute a few empty searches to warmup index
dummy_vector = np.random.randn(768).tolist()
retrieval_service.searcher.hybrid_search(
query_dense=dummy_vector,
query_sparse={0: 0.1},
limit=1,
)
logger.info("Index warmup completed")
Error Troubleshooting Table
| Error Message | Cause | Solution |
|---|---|---|
| Collection not found | Collection not created | Call create_collection() first |
| Dimension mismatch | Vector dimension mismatch with collection config | Check embedding model output dimension |
| Index not ready | Index build not complete | Wait for index build or check indexing_threshold |
| Memory limit exceeded | Data volume exceeds memory limit | Enable on_disk mode or scale up |
| Timeout on hybrid search | Search timeout | Reduce limit, lower ef, optimize filter conditions |
| Sparse vector format error | Incorrect sparse vector format | Ensure {dim_id: float} format |
| Filter syntax error | Filter expression syntax error | Check field names and operators |
| Connection refused | Database not started | Check Docker container status |
| Rate limit exceeded | Request rate too high | Add rate limiting or batch interface |
| Reranker OOM | Reranker model out of memory | Reduce oversampling factor or use smaller reranker model |
Advanced Optimization
1. Adaptive Weight Adjustment
def adaptive_weights(query: str, dense_weight: float = 0.7) -> tuple[float, float]:
"""Adaptively adjust dense/sparse weights based on query characteristics"""
# Long queries lean toward semantics (dense), short queries lean toward keywords (sparse)
if len(query) > 50:
return (0.8, 0.2) # Long query: semantics first
elif len(query) < 10:
return (0.4, 0.6) # Short query: keywords first
else:
return (dense_weight, 1.0 - dense_weight)
2. Tiered Retrieval Strategy
def tiered_search(query: str, limit: int = 10):
"""Tiered retrieval: fast then accurate"""
# Level 1: Low precision fast retrieval
fast_results = searcher.hybrid_search(
query_dense=encode(query),
query_sparse=encode_sparse(query),
limit=limit * 2,
search_params={"ef": 32}, # Low precision
)
# Level 2: High precision ranking
if len(fast_results) > limit:
reranked = reranker.rank(query, fast_results)
return reranked[:limit]
return fast_results
3. Cache Hot Queries
from functools import lru_cache
import hashlib
@lru_cache(maxsize=10000)
def cached_search(query_hash: str, limit: int, filter_hash: str):
"""Cache hot query results"""
return retrieval_service.search(query, limit, filter_expr)
def search_with_cache(query: str, limit: int, filter_expr: str = ""):
query_hash = hashlib.md5(query.encode()).hexdigest()
filter_hash = hashlib.md5(filter_expr.encode()).hexdigest()
return cached_search(query_hash, limit, filter_hash)
4. Data Sharding Strategy
# Shard by category, reduce single collection data volume
shards = {
"tech": MilvusHybridSearch(collection_name="docs_tech"),
"finance": MilvusHybridSearch(collection_name="docs_finance"),
"health": MilvusHybridSearch(collection_name="docs_health"),
}
def sharded_search(query: str, category: str = None):
if category and category in shards:
return shards[category].hybrid_search(...)
# Global search: parallel query all shards
import asyncio
results = asyncio.gather(*[
shard.hybrid_search(...) for shard in shards.values()
])
return merge_and_rank(results)
5. Monitoring and Alerting
# Prometheus metrics
from prometheus_client import Histogram, Counter
search_latency = Histogram(
"hybrid_search_latency_seconds",
"Hybrid search latency",
["backend", "operation"]
)
search_errors = Counter(
"hybrid_search_errors_total",
"Total search errors",
["backend", "error_type"]
)
@search_latency.labels(backend="milvus", operation="hybrid").time()
def monitored_search(request: SearchRequest):
try:
return retrieval_service.search(request)
except Exception as e:
search_errors.labels(backend="milvus", error_type=type(e).__name__).inc()
raise
Comparison Analysis
| Feature | Milvus | Qdrant | Weaviate |
|---|---|---|---|
| Hybrid Retrieval | Dense+Sparse | Dense+Sparse+RRF | Native Hybrid |
| Filtered Search | Scalar Filter | Advanced Filter | GraphQL Filter |
| Multimodal | External Model Required | External Model Required | Native Multimodal |
| Sparse Vector | SPLADE/BM25 | SPLADE/BM25 | Built-in BM25 |
| Distributed | Native Distributed | Sharding | Multi-node |
| Performance(1M) | ~5ms | ~3ms | ~8ms |
| Performance(100M) | ~15ms | ~20ms | ~50ms |
| Memory Efficiency | 4/5 | 5/5 | 3/5 |
| Ecosystem Maturity | 5/5 | 4/5 | 4/5 |
| Ops Complexity | High | Medium | Medium |
| Use Case | Large-scale Production | Mid-scale/Filter-heavy | Multimodal/RAG |
Summary
Vector database hybrid retrieval has become a standard for RAG applications in 2026:
- Milvus: Top choice for large-scale production, native distributed, comprehensive sparse vector support, suitable for billion-scale data
- Qdrant: Best filtered search performance, Rust implementation for high memory efficiency, suitable for mid-scale + complex filtering scenarios
- Weaviate: Most convenient multimodal retrieval, native text+image joint query support, suitable for rapid RAG deployment
Selection advice: Data volume >100M choose Milvus; complex filter conditions choose Qdrant; multimodal needs choose Weaviate; all three support hybrid retrieval, the key is choosing based on your business scenario.
Online Tool Recommendations
- /en/json/format - JSON formatter for debugging vector database API responses
- /en/dev/curl-to-code - cURL to code converter for quickly generating vector database API calls
- /en/encode/hash - Hash calculator for generating unique document IDs
- /en/text/diff - Text diff tool for comparing results from different retrieval strategies
Try these browser-local tools — no sign-up required →
#向量数据库#混合检索#Milvus#Qdrant#Weaviate#2026#数据库