qdrant_config.py•5.34 kB
#!/usr/bin/env python3
"""
Qdrant性能优化配置
专门用于调优HNSW参数，提升检索性能
"""
from typing import Dict, Any, Optional
from qdrant_client.models import SearchParams, HnswConfigDiff
class QdrantPerformanceConfig:
"""Qdrant性能配置管理器"""
def __init__(self):
# 基础性能配置
self.base_config = {
# HNSW参数优化 - 针对个人场景(<10万条数据)
"ef": 128, # 搜索时的候选数量，96-128适合个人场景
"ef_construct": 200, # 构建时的候选数量，影响索引质量
"m": 16, # 每个节点的连接数，16是平衡值
"max_indexing_threads": 4, # 索引线程数，适合个人机器
# 搜索参数
"exact": False, # 使用近似搜索，提升速度
"hnsw_ef": 128, # HNSW搜索参数
# 性能优化
"quantization": None, # 个人场景不需要量化
"on_disk": False, # 内存模式，提升速度
}
# 不同数据量的优化配置
self.size_configs = {
"small": { # <1000条记忆
"ef": 64,
"ef_construct": 100,
"m": 8,
"hnsw_ef": 64
},
"medium": { # 1000-10000条记忆
"ef": 96,
"ef_construct": 150,
"m": 12,
"hnsw_ef": 96
},
"large": { # 10000+条记忆
"ef": 128,
"ef_construct": 200,
"m": 16,
"hnsw_ef": 128
}
}
def get_search_params(self, memory_count: Optional[int] = None) -> SearchParams:
"""获取优化的搜索参数"""
config = self._get_config_for_size(memory_count)
return SearchParams(
hnsw_ef=config["hnsw_ef"],
exact=config.get("exact", False)
)
def get_hnsw_config(self, memory_count: Optional[int] = None) -> HnswConfigDiff:
"""获取HNSW配置"""
config = self._get_config_for_size(memory_count)
return HnswConfigDiff(
ef_construct=config["ef_construct"],
m=config["m"],
max_indexing_threads=config.get("max_indexing_threads", 4)
)
def _get_config_for_size(self, memory_count: Optional[int] = None) -> Dict[str, Any]:
"""根据数据量选择配置"""
if memory_count is None:
return self.base_config
if memory_count < 1000:
return {**self.base_config, **self.size_configs["small"]}
elif memory_count < 10000:
return {**self.base_config, **self.size_configs["medium"]}
else:
return {**self.base_config, **self.size_configs["large"]}
def get_collection_config(self, memory_count: Optional[int] = None) -> Dict[str, Any]:
"""获取集合创建配置"""
config = self._get_config_for_size(memory_count)
return {
"hnsw_config": self.get_hnsw_config(memory_count),
"optimizers_config": {
"deleted_threshold": 0.2,
"vacuum_min_vector_number": 1000,
"default_segment_number": 2,
"max_segment_size": 200000, # 20万向量per segment
"memmap_threshold": 50000, # 5万向量开始使用mmap
"indexing_threshold": 20000, # 2万向量开始索引
"flush_interval_sec": 5,
"max_optimization_threads": 2
}
}
def print_config_summary(self, memory_count: Optional[int] = None):
"""打印配置摘要"""
config = self._get_config_for_size(memory_count)
size_category = self._get_size_category(memory_count)
print(f"📊 Qdrant性能配置 ({size_category})")
print(f" 记忆数量: {memory_count or '未知'}")
print(f" HNSW ef: {config['hnsw_ef']}")
print(f" ef_construct: {config['ef_construct']}")
print(f" m (连接数): {config['m']}")
print(f" 精确搜索: {'否' if not config.get('exact', False) else '是'}")
def _get_size_category(self, memory_count: Optional[int] = None) -> str:
"""获取数据量分类"""
if memory_count is None:
return "默认"
elif memory_count < 1000:
return "小型"
elif memory_count < 10000:
return "中型"
else:
return "大型"
# 全局配置实例
qdrant_perf_config = QdrantPerformanceConfig()
def get_optimized_search_params(memory_count: Optional[int] = None) -> SearchParams:
"""获取优化的搜索参数 - 便捷函数"""
return qdrant_perf_config.get_search_params(memory_count)
def print_performance_tips():
"""打印性能优化建议"""
print("🚀 Qdrant性能优化建议:")
print(" 1. 使用HNSW近似搜索而非精确搜索")
print(" 2. 根据数据量调整ef参数(64-128)")
print(" 3. 适当的m值平衡精度和速度(8-16)")
print(" 4. 启用内存模式提升速度")
print(" 5. 合理设置索引线程数")