ベクトルデータベースハイブリッド検索：Milvus/Qdrant/Weaviate比較完全ガイド 2026
数据库
ベクトルデータベースハイブリッド検索：Milvus/Qdrant/Weaviate比較完全ガイド 2026
RAG（検索拡張生成）アプリケーションの核心的なボトルネックは生成ではなく検索にある。純粋なベクトル検索は意味マッチングに優れるが、精密なフィルタリングができない。純粋なキーワード検索は精密マッチングに優れるが、意味を失う。ハイブリッド検索は両者を融合し、2026年にはベクトルデータベースの標準機能となっている。しかし、Milvus、Qdrant、Weaviateのハイブリッド検索実装は大きく異なり、データベースの選択を誤るとパフォーマンスと機能で大きな代償を払うことになる。
コア概念一覧
| 概念 | 説明 | 適用シナリオ |
|---|---|---|
| ベクトル検索 | 埋め込みベクトルに基づく類似度検索 | 意味マッチング |
| キーワード検索 | BM25/TF-IDFに基づくテキスト検索 | 精密マッチング |
| ハイブリッド検索 | ベクトル+キーワード融合検索 | 本番級RAG |
| 密ベクトル | ニューラルネットワークが生成する埋め込みベクトル | 意味理解 |
| 疎ベクトル | BM25/SPLADEが生成する疎表現 | キーワードマッチング |
| リランキング | 検索結果の二次ソート | 精度向上 |
| フィルタ検索 | メタデータにフィルタ条件を追加 | 条件絞り込み |
| マルチモーダル検索 | テキスト/画像/音声をまたぐ検索 | クロスモーダル検索 |
5つの主要ペインポイント
- 純粋なベクトル検索の精度不足：意味は似ているが内容が無関係な結果がTop-Kに混入。例えば「Apple携帯」を検索すると「りんご果物」のドキュメントが返される
- キーワード検索で意味が失われる：同義語や文脈を理解できず、「AI」で「人工知能」のドキュメントが見つからない
- フィルタ条件とベクトル検索の乖離：先にフィルタしてから検索すると再現率が不足し、先に検索してからフィルタするとパフォーマンスが無駄になる
- マルチモーダルデータの統一検索が困難：テキスト、画像、表が混在し、統一された検索インターフェースがない
- 本番環境のパフォーマンスボトルネック：億規模のベクトル+複雑なフィルタ条件下で、P99レイテンシがミリ秒から秒単位に急増
ステップバイステップ：5つのコアパターン
パターン1：ベクトルデータベース選定
実行環境：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")
パターン2：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]:
"""ハイブリッド検索 + リランキング"""
# 第1段階：ハイブリッド検索オーバーサンプリング
candidates = self.hybrid_search(
query_dense=query_dense,
query_sparse=query_sparse,
limit=limit * 4, # 4倍オーバーサンプリング
)
# 第2段階：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": "ベクトルDB", "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")
パターン3：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")
パターン4：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()
パターン5：本番級ハイブリッド検索アーキテクチャ
億規模のベクトルをサポートする本番級ハイブリッド検索システムを構築：
# 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):
"""段階検索：速くから正確に"""
# 第1段階：低精度高速検索
fast_results = searcher.hybrid_search(
query_dense=encode(query),
query_sparse=encode_sparse(query),
limit=limit * 2,
search_params={"ef": 32}, # 低精度
)
# 第2段階：高精度ランキング
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 |
| メモリ効率 | 4/5 | 5/5 | 3/5 |
| エコシステム成熟度 | 5/5 | 4/5 | 4/5 |
| 運用複雑度 | 高 | 中 | 中 |
| 適用シナリオ | 大規模本番 | 中小規模/フィルタ重視 | マルチモーダル/RAG |
まとめ
ベクトルデータベースハイブリッド検索は2026年にRAGアプリケーションの標準機能となった：
- Milvus：大規模本番の第一選択、ネイティブ分散、疎ベクトルサポートが充実、億規模データに適している
- Qdrant：フィルタ検索のパフォーマンスが最も優秀、Rust実装でメモリ効率が高い、中小規模+複雑フィルタシナリオに適している
- Weaviate：マルチモーダル検索が最も便利、ネイティブでテキスト+画像联合クエリをサポート、RAGの迅速導入に適している
選定アドバイス：データ量>1億ならMilvus、フィルタ条件が複雑ならQdrant、マルチモーダルニーズならWeaviate。3つすべてハイブリッド検索をサポートしており、重要なのはビジネスシナリオに基づいて選択すること。
オンラインツール推奨
- /ja/json/format - JSONフォーマッター、ベクトルデータベースAPIレスポンスのデバッグに
- /ja/dev/curl-to-code - cURLコード変換、ベクトルデータベースAPI呼び出しを迅速生成
- /ja/encode/hash - ハッシュ計算ツール、ドキュメントの一意IDを生成
- /ja/text/diff - テキスト差分ツール、異なる検索戦略の結果差異を比較
ブラウザローカルツールを無料で試す →
#向量数据库#混合检索#Milvus#Qdrant#Weaviate#2026#数据库