向量資料庫混合檢索：Milvus/Qdrant/Weaviate對比完整指南 2026
数据库
向量資料庫混合檢索：Milvus/Qdrant/Weaviate對比完整指南 2026
RAG（檢索增強產生）應用的核心瓶頸不在產生，而在檢索。純向量檢索擅长語義匹配，卻無法精確過濾；純關鍵詞檢索擅长精確匹配，卻丟失語義。混合檢索将兩者融合，在2026年已成為向量資料庫的標配能力。但Milvus、Qdrant、Weaviate三者的混合檢索實作差異巨大，選錯資料庫可能讓你在性能和功能上付出沉重代價。
核心概念速覽
| 概念 | 說明 | 適用場景 |
|---|---|---|
| 向量檢索 | 基於嵌入向量的相似度搜尋 | 語義匹配 |
| 關鍵詞檢索 | 基於BM25/TF-IDF的文本搜尋 | 精確匹配 |
| 混合檢索 | 向量+關鍵詞融合檢索 | 生產級RAG |
| 稠密向量 | 神經網路產生的嵌入向量 | 語義理解 |
| 稀疏向量 | BM25/SPLADE產生的稀疏表示 | 關鍵詞匹配 |
| 重排序 | 對檢索結果二次排序 | 提升精度 |
| 過濾搜尋 | 在元資料上添加過濾條件 | 條件篩選 |
| 多模態檢索 | 跨文本/圖像/音訊的檢索 | 跨模態搜尋 |
五大痛點分析
- 純向量檢索精度不足：語義相似但内容无關的結果混入Top-K，例如搜尋"蘋果手機"返回"蘋果水果"的文件
- 關鍵詞檢索丟失語義：無法理解同義詞和上下文，"AI"搜不到"人工智能"的文件
- 過濾條件與向量檢索脫節：先過濾再檢索導致召回不足，先檢索再過濾導致性能浪費
- 多模態資料難以統一檢索：文本、圖像、表格混存，缺乏統一的檢索介面
- 生產環境性能瓶頸：億級向量+複雜過濾條件下，P99延遲從毫秒級飆升到秒級
分步實操：5個核心模式
模式一：向量資料庫選型
執行環境：Python 3.12+ / Docker 27+
# 選型對比腳本 - 自動化基准測試
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+ 基准測試"""
from pymilvus import MilvusClient, DataType
client = MilvusClient(uri="http://localhost:19530")
# 創建集合
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
)
# 插入測試
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
# 向量搜尋測試
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
# 混合搜尋測試（向量+過濾）
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
)
# 執行基准測試
# 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")
模式二：Milvus混合檢索
Milvus 2.5+支援稠密+稀疏向量的原生混合檢索：
# milvus_hybrid_search.py
# 執行環境：Milvus 2.5+ / pymilvus 2.5+
from pymilvus import (
MilvusClient, DataType,
AnnSearchRequest, WeightedRanker
)
import numpy as np
class MilvusHybridSearch:
"""Milvus混合檢索 - 稠密向量 + 稀疏向量(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):
"""創建支援混合檢索的集合"""
schema = self.client.create_schema(auto_id=True)
# 主鍵
schema.add_field("id", DataType.INT64, is_primary=True)
# 稠密向量欄位 - 語義搜尋
schema.add_field("dense_vector", DataType.FLOAT_VECTOR, dim=self.dim)
# 稀疏向量欄位 - 關鍵詞搜尋（BM25/SPLADE）
schema.add_field("sparse_vector", DataType.SPARSE_FLOAT_VECTOR)
# 文本和元資料
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)
# 創建索引
index_params = self.client.prepare_index_params()
# 稠密向量索引 - HNSW
index_params.add_index(
field_name="dense_vector",
index_type="HNSW",
metric_type="COSINE",
params={"M": 16, "efConstruction": 256}
)
# 稀疏向量索引 - SPARSE_INVERTED_INDEX
index_params.add_index(
field_name="sparse_vector",
index_type="SPARSE_INVERTED_INDEX",
metric_type="IP",
)
# 標量索引
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]
):
"""插入文件"""
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]:
"""混合檢索 - 稠密+稀疏加權融合"""
# 稠密向量搜尋请求
dense_req = AnnSearchRequest(
data=[query_dense],
anns_field="dense_vector",
param={
"metric_type": "COSINE",
"params": {"ef": 128}
},
limit=limit * 2 # 過采样
)
# 稀疏向量搜尋请求
sparse_req = AnnSearchRequest(
data=[query_sparse],
anns_field="sparse_vector",
param={
"metric_type": "IP",
},
limit=limit * 2
)
# 加權融合排序
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]:
"""混合檢索 + 重排序"""
# 第一階段：混合檢索過采样
candidates = self.hybrid_search(
query_dense=query_dense,
query_sparse=query_sparse,
limit=limit * 4, # 過采样4倍
)
# 第二階段：Cross-Encoder重排序
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
pairs = [[query, c["text"]] for c in candidates]
scores = reranker.predict(pairs)
# 合并分數并排序
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]
# 使用示例
if __name__ == "__main__":
searcher = MilvusHybridSearch()
searcher.create_collection()
# 模擬插入資料
from sklearn.feature_extraction.text import TfidfVectorizer
docs = [
"Rust語言在嵌入式系統中的應用越来越廣泛",
"向量資料庫混合檢索技術詳解",
"深度學習模型部署最佳實踐",
"Kubernetes集群運維自動化方案",
"大語言模型RAG架構設計",
]
# 產生稀疏向量（BM25風格）
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)
# 產生稠密向量
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嵌入式", "category": "programming", "tags": ["rust", "embedded"]},
{"title": "向量資料庫", "category": "database", "tags": ["vector", "search"]},
{"title": "模型部署", "category": "ai", "tags": ["ml", "deployment"]},
{"title": "K8s運維", "category": "devops", "tags": ["k8s", "sre"]},
{"title": "RAG架構", "category": "ai", "tags": ["llm", "rag"]},
]
searcher.insert_documents(docs, dense_vectors.tolist(), sparse_vectors, metadata)
print("✅ Documents inserted successfully")
模式三：Qdrant過濾搜尋
Qdrant的過濾搜尋在性能和靈活性上表现優異：
# qdrant_filtered_search.py
# 執行環境：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混合檢索 - 向量搜尋 + 精確過濾 + 稀疏向量"""
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):
"""創建支援混合檢索的集合"""
self.client.create_collection(
collection_name=self.collection_name,
vectors_config={
"dense": VectorParams(
size=self.dim,
distance=Distance.COSINE,
on_disk=True, # 大規模資料啟用磁碟存儲
)
},
sparse_vectors_config={
"sparse": SparseVectorParams(
index=SparseIndexParams(on_disk=False)
)
},
# 啟用Wal和優化器
optimizers_config={
"indexing_threshold": 20000,
"memmap_threshold": 50000,
}
)
# 創建負載索引（加速過濾）
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]
):
"""插入文件"""
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]:
"""過濾搜尋 - 向量搜尋 + 精確過濾條件"""
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]:
"""混合檢索 - 稠密+稀疏融合"""
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]:
"""多租戶隔離搜尋"""
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
]
# 使用示例
if __name__ == "__main__":
searcher = QdrantHybridSearch()
searcher.create_collection()
# 過濾搜尋
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")
模式四：Weaviate多模態檢索
Weaviate原生支援多模態混合檢索：
# weaviate_multimodal_search.py
# 執行環境：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多模態混合檢索"""
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):
"""創建支援多模態的集合"""
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",
# 多模態向量化：文本+圖像
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],
):
"""插入多模態文件"""
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]:
"""文本語義搜尋 + 過濾"""
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]:
"""多模態檢索 - 文本+圖像聯合查詢"""
collection = self.client.collections.get(self.collection_name)
if query and image_path:
# 文本+圖像聯合查詢
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()
# 使用示例
if __name__ == "__main__":
searcher = WeaviateMultimodalSearch()
searcher.create_collection()
results = searcher.text_search(
query="向量資料庫混合檢索",
category="database",
year_min=2024,
)
for r in results:
print(f"[{r['score']:.3f}] {r['title']}: {r['text'][:50]}...")
searcher.close()
模式五：生產級混合檢索架構
構建支援億級向量的生產級混合檢索系統：
# production_hybrid_retrieval.py
# 執行環境：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:
"""嵌入服務 - 統一向量化介面"""
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]:
"""產生稠密向量"""
embedding = self.dense_model.encode(text, normalize_embeddings=True)
return embedding.tolist()
def encode_sparse(self, text: str) -> dict[int, float]:
"""產生稀疏向量（SPLADE風格）"""
# 簡化實作，生產環境使用SPLADE模型
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:
"""生產級混合檢索服務"""
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:
"""執行混合檢索"""
start_time = time.time()
# 1. 向量化查詢
dense_vector = self.embedding.encode_dense(request.query)
sparse_vector = self.embedding.encode_sparse(request.query)
# 2. 構建過濾條件
filter_expr = self._build_filter(request)
# 3. 執行混合檢索
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. 重排序（可選）
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:
"""構建過濾表達式"""
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重排序"""
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
# 全局服務實例
retrieval_service = HybridRetrievalService(backend="milvus")
@app.post("/search", response_model=SearchResponse)
async def search(request: SearchRequest):
"""混合檢索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}
避坑指南
坑1：忽略向量歸一化
# ❌ 錯誤：未歸一化的向量導致餘弦相似度計算偏差
vectors = model.encode(texts) # 可能未歸一化
# ✅ 正確：始终歸一化向量
vectors = model.encode(texts, normalize_embeddings=True)
# 或手動歸一化
vectors = vectors / np.linalg.norm(vectors, axis=1, keepdims=True)
坑2：過濾條件過於嚴格
# ❌ 錯誤：先過濾再檢索，召回不足
filter = 'category == "tech" and year == 2026 and author == "張三"'
# 可能過濾掉大部分文件，向量檢索空间太小
# ✅ 正確：寬鬆過濾 + 重排序
filter = 'category == "tech" and year >= 2024' # 放寬條件
# 然后用重排序模型精排
坑3：HNSW參數設定不當
# ❌ 錯誤：M和ef參數過小，召回率低
index_params = {"M": 4, "efConstruction": 32} # 召回率可能低於80%
# ✅ 正確：根據資料規模調整參數
# 小資料集（<100萬）
index_params = {"M": 16, "efConstruction": 256}
# 大資料集（>1000萬）
index_params = {"M": 32, "efConstruction": 512}
# 搜尋時ef >= limit * 2
search_params = {"ef": 256}
坑4：稀疏向量化方法選择錯誤
# ❌ 錯誤：使用TF-IDF作為稀疏向量，缺乏語義資訊
from sklearn.feature_extraction.text import TfidfVectorizer
# ✅ 正確：使用SPLADE或BM25+語義擴展
# SPLADE: 學習稀疏表示，同時保留語義
# BM25: 傳統關鍵詞匹配，適合精確過濾
# 推薦：SPLADE用於稀疏向量欄位，BM25用於輔助過濾
坑5：忽略索引預熱
# ❌ 錯誤：冷啟動時搜尋延遲極高
# 首次搜尋需要載入索引到記憶體，P99延遲可能>10秒
# ✅ 正確：服務啟動時預熱索引
@app.on_event("startup")
async def warmup():
# 執行幾次空搜尋預熱索引
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")
報錯排查表
| 報錯資訊 | 原因 | 解决方案 |
|---|---|---|
Collection not found |
集合未創建 | 先呼叫create_collection() |
Dimension mismatch |
向量維度與集合設定不一致 | 檢查嵌入模型輸出維度 |
Index not ready |
索引構建未完成 | 等待索引構建，或檢查indexing_threshold |
Memory limit exceeded |
資料量超出記憶體限制 | 啟用on_disk模式或擴容 |
Timeout on hybrid search |
搜尋超時 | 减小limit、降低ef、優化過濾條件 |
Sparse vector format error |
稀疏向量格式不正確 | 確保使用{dim_id: float}格式 |
Filter syntax error |
過濾表達式語法錯誤 | 檢查欄位名和运算符 |
Connection refused |
資料庫未啟動 | 檢查Docker容器状態 |
Rate limit exceeded |
请求頻率過高 | 添加请求限流或批量介面 |
Reranker OOM |
重排序模型記憶體不足 | 减小過采样倍數或使用更小的重排序模型 |
进階優化
1. 自適应權重調整
def adaptive_weights(query: str, dense_weight: float = 0.7) -> tuple[float, float]:
"""根據查詢特征自適应調整稠密/稀疏權重"""
# 長查詢偏向語義（稠密），短查詢偏向關鍵詞（稀疏）
if len(query) > 50:
return (0.8, 0.2) # 長查詢：語義優先
elif len(query) < 10:
return (0.4, 0.6) # 短查詢：關鍵詞優先
else:
return (dense_weight, 1.0 - dense_weight)
2. 分級檢索策略
def tiered_search(query: str, limit: int = 10):
"""分級檢索：先快後準"""
# 第一級：低精度快速檢索
fast_results = searcher.hybrid_search(
query_dense=encode(query),
query_sparse=encode_sparse(query),
limit=limit * 2,
search_params={"ef": 32}, # 低精度
)
# 第二級：高精度精排
if len(fast_results) > limit:
reranked = reranker.rank(query, fast_results)
return reranked[:limit]
return fast_results
3. 快取熱门查詢
from functools import lru_cache
import hashlib
@lru_cache(maxsize=10000)
def cached_search(query_hash: str, limit: int, filter_hash: str):
"""快取熱门查詢結果"""
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. 資料分片策略
# 按类别分片，減少單集合資料量
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(...)
# 全局搜尋：並行查詢所有分片
import asyncio
results = asyncio.gather(*[
shard.hybrid_search(...) for shard in shards.values()
])
return merge_and_rank(results)
5. 監控與告警
# Prometheus指標
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
對比分析
| 特性 | Milvus | Qdrant | Weaviate |
|---|---|---|---|
| 混合檢索 | ✅ 稠密+稀疏 | ✅ 稠密+稀疏+RRF | ✅ 原生混合 |
| 過濾搜尋 | ✅ 標量過濾 | ✅ 高級過濾 | ✅ GraphQL過濾 |
| 多模態 | ❌ 需外部模型 | ❌ 需外部模型 | ✅ 原生多模態 |
| 稀疏向量 | ✅ SPLADE/BM25 | ✅ SPLADE/BM25 | ✅ 内置BM25 |
| 分布式 | ✅ 原生分布式 | ✅ 分片 | ✅ 多節點 |
| 性能(100萬) | ~5ms | ~3ms | ~8ms |
| 性能(1億) | ~15ms | ~20ms | ~50ms |
| 記憶體效率 | ★★★★ | ★★★★★ | ★★★ |
| 生態成熟度 | ★★★★★ | ★★★★ | ★★★★ |
| 運維複雜度 | 高 | 中 | 中 |
| 適用場景 | 大規模生產 | 中小規模/過濾重 | 多模態/RAG |
總結
向量資料庫混合檢索在2026年已成為RAG應用的標配：
- Milvus：大規模生產首選，原生分布式，稀疏向量支援完善，適合億級資料
- Qdrant：過濾搜尋性能最優，Rust實作記憶體效率高，適合中小規模+複雜過濾場景
- Weaviate：多模態檢索最便捷，原生支援文本+圖像聯合查詢，適合RAG快速落地
選择建議：資料量>1億選Milvus；過濾條件複雜選Qdrant；多模態需求選Weaviate；三者都支援混合檢索，關鍵是根據業務場景選择。
在線工具推薦
- /zh-TW/json/format - JSON格式化工具，除錯向量資料庫API回應
- /zh-TW/dev/curl-to-code - cURL轉程式碼，快速產生向量資料庫API呼叫
- /zh-TW/encode/hash - 哈希計算工具，產生文件唯一ID
- /zh-TW/text/diff - 文字對比工具，對比不同檢索策略的結果差異
本站提供瀏覽器本地工具，免註冊即可試用 →
#向量数据库#混合检索#Milvus#Qdrant#Weaviate#2026#数据库