wehub-skill-sync
14 KiB
14 KiB
Qdrant 高级使用指南
分布式部署
集群搭建
Qdrant 使用 Raft 共识算法进行分布式协调。
# docker-compose.yml for 3-node cluster
version: '3.8'
services:
qdrant-node-1:
image: qdrant/qdrant:latest
ports:
- "6333:6333"
- "6334:6334"
- "6335:6335"
volumes:
- ./node1_storage:/qdrant/storage
environment:
- QDRANT__CLUSTER__ENABLED=true
- QDRANT__CLUSTER__P2P__PORT=6335
- QDRANT__SERVICE__HTTP_PORT=6333
- QDRANT__SERVICE__GRPC_PORT=6334
qdrant-node-2:
image: qdrant/qdrant:latest
ports:
- "6343:6333"
- "6344:6334"
- "6345:6335"
volumes:
- ./node2_storage:/qdrant/storage
environment:
- QDRANT__CLUSTER__ENABLED=true
- QDRANT__CLUSTER__P2P__PORT=6335
- QDRANT__CLUSTER__BOOTSTRAP=http://qdrant-node-1:6335
depends_on:
- qdrant-node-1
qdrant-node-3:
image: qdrant/qdrant:latest
ports:
- "6353:6333"
- "6354:6334"
- "6355:6335"
volumes:
- ./node3_storage:/qdrant/storage
environment:
- QDRANT__CLUSTER__ENABLED=true
- QDRANT__CLUSTER__P2P__PORT=6335
- QDRANT__CLUSTER__BOOTSTRAP=http://qdrant-node-1:6335
depends_on:
- qdrant-node-1
分片配置
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, ShardingMethod
client = QdrantClient(host="localhost", port=6333)
# 创建分片集合
client.create_collection(
collection_name="large_collection",
vectors_config=VectorParams(size=384, distance=Distance.COSINE),
shard_number=6, # 分片数量
replication_factor=2, # 每个分片的副本数
write_consistency_factor=1 # 写入所需确认数
)
# 检查集群状态
cluster_info = client.get_cluster_info()
print(f"Peers: {cluster_info.peers}")
print(f"Raft state: {cluster_info.raft_info}")
复制与一致性
from qdrant_client.models import WriteOrdering
# 强一致性写入
client.upsert(
collection_name="critical_data",
points=points,
ordering=WriteOrdering.STRONG # 等待所有副本
)
# 最终一致性（更快）
client.upsert(
collection_name="logs",
points=points,
ordering=WriteOrdering.WEAK # 主节点确认后返回
)
# 从特定分片读取
results = client.search(
collection_name="documents",
query_vector=query,
consistency="majority" # 从多数副本读取
)
混合搜索
稠密 + 稀疏向量
结合语义（稠密）与关键词（稀疏）搜索：
from qdrant_client.models import (
VectorParams, SparseVectorParams, SparseIndexParams,
Distance, PointStruct, SparseVector, Prefetch, Query
)
# 创建混合集合
client.create_collection(
collection_name="hybrid",
vectors_config={
"dense": VectorParams(size=384, distance=Distance.COSINE)
},
sparse_vectors_config={
"sparse": SparseVectorParams(
index=SparseIndexParams(on_disk=False)
)
}
)
# 插入两种向量类型
def encode_sparse(text: str) -> SparseVector:
"""简单的类 BM25 稀疏编码"""
from collections import Counter
tokens = text.lower().split()
counts = Counter(tokens)
# 将 token 映射到索引（生产环境请使用词表）
indices = [hash(t) % 30000 for t in counts.keys()]
values = list(counts.values())
return SparseVector(indices=indices, values=values)
client.upsert(
collection_name="hybrid",
points=[
PointStruct(
id=1,
vector={
"dense": dense_encoder.encode("Python programming").tolist(),
"sparse": encode_sparse("Python programming language code")
},
payload={"text": "Python programming language code"}
)
]
)
# 使用互惠排名融合（RRF）进行混合搜索
from qdrant_client.models import FusionQuery
results = client.query_points(
collection_name="hybrid",
prefetch=[
Prefetch(query=dense_query, using="dense", limit=20),
Prefetch(query=sparse_query, using="sparse", limit=20)
],
query=FusionQuery(fusion="rrf"), # 合并结果
limit=10
)
多阶段搜索
from qdrant_client.models import Prefetch, Query
# 两阶段检索：粗筛后精排
results = client.query_points(
collection_name="documents",
prefetch=[
Prefetch(
query=query_vector,
limit=100, # 宽筛第一阶段
params={"quantization": {"rescore": False}} # 快速近似
)
],
query=Query(nearest=query_vector),
limit=10,
params={"quantization": {"rescore": True}} # 精确重排序
)
推荐
物品到物品推荐
# 查找相似物品
recommendations = client.recommend(
collection_name="products",
positive=[1, 2, 3], # 用户喜欢的 ID
negative=[4], # 用户不喜欢的 ID
limit=10
)
# 带过滤条件的推荐
recommendations = client.recommend(
collection_name="products",
positive=[1, 2],
query_filter={
"must": [
{"key": "category", "match": {"value": "electronics"}},
{"key": "in_stock", "match": {"value": True}}
]
},
limit=10
)
从其他集合查找
from qdrant_client.models import RecommendStrategy, LookupLocation
# 使用另一个集合中的向量进行推荐
results = client.recommend(
collection_name="products",
positive=[
LookupLocation(
collection_name="user_history",
id="user_123"
)
],
strategy=RecommendStrategy.AVERAGE_VECTOR,
limit=10
)
高级过滤
嵌套负载过滤
from qdrant_client.models import Filter, FieldCondition, MatchValue, NestedCondition
# 对嵌套对象进行过滤
results = client.search(
collection_name="documents",
query_vector=query,
query_filter=Filter(
must=[
NestedCondition(
key="metadata",
filter=Filter(
must=[
FieldCondition(
key="author.name",
match=MatchValue(value="John")
)
]
)
)
]
),
limit=10
)
地理过滤
from qdrant_client.models import FieldCondition, GeoRadius, GeoPoint
# 查找半径范围内的结果
results = client.search(
collection_name="locations",
query_vector=query,
query_filter=Filter(
must=[
FieldCondition(
key="location",
geo_radius=GeoRadius(
center=GeoPoint(lat=40.7128, lon=-74.0060),
radius=5000 # 米
)
)
]
),
limit=10
)
# 地理边界框
from qdrant_client.models import GeoBoundingBox
results = client.search(
collection_name="locations",
query_vector=query,
query_filter=Filter(
must=[
FieldCondition(
key="location",
geo_bounding_box=GeoBoundingBox(
top_left=GeoPoint(lat=40.8, lon=-74.1),
bottom_right=GeoPoint(lat=40.6, lon=-73.9)
)
)
]
),
limit=10
)
全文搜索
from qdrant_client.models import TextIndexParams, TokenizerType
# 创建文本索引
client.create_payload_index(
collection_name="documents",
field_name="content",
field_schema=TextIndexParams(
type="text",
tokenizer=TokenizerType.WORD,
min_token_len=2,
max_token_len=15,
lowercase=True
)
)
# 全文过滤
from qdrant_client.models import MatchText
results = client.search(
collection_name="documents",
query_vector=query,
query_filter=Filter(
must=[
FieldCondition(
key="content",
match=MatchText(text="machine learning")
)
]
),
limit=10
)
量化策略
标量量化（INT8）
from qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig, ScalarType
# 内存减少约 4 倍，精度损失极小
client.create_collection(
collection_name="scalar_quantized",
vectors_config=VectorParams(size=384, distance=Distance.COSINE),
quantization_config=ScalarQuantization(
scalar=ScalarQuantizationConfig(
type=ScalarType.INT8,
quantile=0.99, # 裁剪极端值
always_ram=True # 将量化后的向量保留在 RAM 中
)
)
)
乘积量化
from qdrant_client.models import ProductQuantization, ProductQuantizationConfig, CompressionRatio
# 内存减少约 16 倍，有一定精度损失
client.create_collection(
collection_name="product_quantized",
vectors_config=VectorParams(size=384, distance=Distance.COSINE),
quantization_config=ProductQuantization(
product=ProductQuantizationConfig(
compression=CompressionRatio.X16,
always_ram=True
)
)
)
二值量化
from qdrant_client.models import BinaryQuantization, BinaryQuantizationConfig
# 内存减少约 32 倍，需要过采样
client.create_collection(
collection_name="binary_quantized",
vectors_config=VectorParams(size=384, distance=Distance.COSINE),
quantization_config=BinaryQuantization(
binary=BinaryQuantizationConfig(always_ram=True)
)
)
# 带过采样的搜索
results = client.search(
collection_name="binary_quantized",
query_vector=query,
search_params={
"quantization": {
"rescore": True,
"oversampling": 2.0 # 检索 2 倍候选数量后重排序
}
},
limit=10
)
快照与备份
创建快照
# 创建集合快照
snapshot_info = client.create_snapshot(collection_name="documents")
print(f"Snapshot: {snapshot_info.name}")
# 列出快照
snapshots = client.list_snapshots(collection_name="documents")
for s in snapshots:
print(f"{s.name}: {s.size} bytes")
# 完整存储快照
full_snapshot = client.create_full_snapshot()
从快照恢复
# 下载快照
client.download_snapshot(
collection_name="documents",
snapshot_name="documents-2024-01-01.snapshot",
target_path="./backup/"
)
# 恢复（通过 REST API）
import requests
response = requests.put(
"http://localhost:6333/collections/documents/snapshots/recover",
json={"location": "file:///backup/documents-2024-01-01.snapshot"}
)
集合别名
# 创建别名
client.update_collection_aliases(
change_aliases_operations=[
{"create_alias": {"alias_name": "production", "collection_name": "documents_v2"}}
]
)
# 蓝绿部署
# 1. 创建带更新的新集合
client.create_collection(collection_name="documents_v3", ...)
# 2. 填充新集合
client.upsert(collection_name="documents_v3", points=new_points)
# 3. 原子切换
client.update_collection_aliases(
change_aliases_operations=[
{"delete_alias": {"alias_name": "production"}},
{"create_alias": {"alias_name": "production", "collection_name": "documents_v3"}}
]
)
# 通过别名搜索
results = client.search(collection_name="production", query_vector=query, limit=10)
滚动与迭代
滚动遍历所有点
# 分页迭代
offset = None
all_points = []
while True:
results, offset = client.scroll(
collection_name="documents",
limit=100,
offset=offset,
with_payload=True,
with_vectors=False
)
all_points.extend(results)
if offset is None:
break
print(f"Total points: {len(all_points)}")
带过滤的滚动
# 带过滤条件的滚动
results, _ = client.scroll(
collection_name="documents",
scroll_filter=Filter(
must=[
FieldCondition(key="status", match=MatchValue(value="active"))
]
),
limit=1000
)
异步客户端
import asyncio
from qdrant_client import AsyncQdrantClient
async def main():
client = AsyncQdrantClient(host="localhost", port=6333)
# 异步操作
await client.create_collection(
collection_name="async_docs",
vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)
await client.upsert(
collection_name="async_docs",
points=points
)
results = await client.search(
collection_name="async_docs",
query_vector=query,
limit=10
)
return results
results = asyncio.run(main())
gRPC 客户端
from qdrant_client import QdrantClient
# 优先使用 gRPC 以获得更佳性能
client = QdrantClient(
host="localhost",
port=6333,
grpc_port=6334,
prefer_grpc=True # 可用时使用 gRPC
)
# 纯 gRPC 客户端
from qdrant_client import QdrantClient
client = QdrantClient(
host="localhost",
grpc_port=6334,
prefer_grpc=True,
https=False
)
多租户
基于负载的隔离
# 单个集合，按租户过滤
client.upsert(
collection_name="multi_tenant",
points=[
PointStruct(
id=1,
vector=embedding,
payload={"tenant_id": "tenant_a", "text": "..."}
)
]
)
# 在租户范围内搜索
results = client.search(
collection_name="multi_tenant",
query_vector=query,
query_filter=Filter(
must=[FieldCondition(key="tenant_id", match=MatchValue(value="tenant_a"))]
),
limit=10
)
每租户独立集合
# 创建租户集合
def create_tenant_collection(tenant_id: str):
client.create_collection(
collection_name=f"tenant_{tenant_id}",
vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)
# 搜索租户集合
def search_tenant(tenant_id: str, query_vector: list, limit: int = 10):
return client.search(
collection_name=f"tenant_{tenant_id}",
query_vector=query_vector,
limit=limit
)
性能监控
集合统计信息
# 集合信息
info = client.get_collection("documents")
print(f"Points: {info.points_count}")
print(f"Indexed vectors: {info.indexed_vectors_count}")
print(f"Segments: {len(info.segments)}")
print(f"Status: {info.status}")
# 详细段信息
for i, segment in enumerate(info.segments):
print(f"Segment {i}: {segment}")
遥测
# 获取遥测数据
telemetry = client.get_telemetry()
print(f"Collections: {telemetry.collections}")
print(f"Operations: {telemetry.operations}")