数据分析agent三，qdrant(2)教程
·
import logging
from typing import Optional, List, Dict, Any, Union
from qdrant_client import QdrantClient
from qdrant_client.http import models
from qdrant_client.http.exceptions import UnexpectedResponse
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class QdrantClientManager:
"""
Qdrant 客户端管理器
封装了连接、Collection 管理、向量增删改查等常用操作
"""
def __init__(
self,
host: str = "localhost",
port: int = 6333,
api_key: Optional[str] = None,
prefer_grpc: bool = False,
timeout: int = 10
):
"""
初始化 Qdrant 客户端
Args:
host: Qdrant 服务地址
port: 端口号（HTTP: 6333, gRPC: 6334）
api_key: API 密钥（如果服务端设置了）
prefer_grpc: 是否优先使用 gRPC
timeout: 超时时间（秒）
"""
self.host = host
self.port = port
self.api_key = api_key
self.client = QdrantClient(
host=host,
port=port,
api_key=api_key,
prefer_grpc=prefer_grpc,
timeout=timeout
)
# 缓存已存在的 collection 名称
self._collections_cache = set()
self._refresh_collections_cache()
logger.info(f"✅ Qdrant 客户端初始化成功: {host}:{port}")
# ==================== 连接状态 ====================
def is_healthy(self) -> bool:
"""检查服务是否健康"""
try:
self.client.get_collections()
return True
except Exception as e:
logger.error(f"❌ 服务不健康: {e}")
return False
def get_version(self) -> str:
"""获取 Qdrant 服务版本"""
try:
response = self.client.get_collections()
return response.version if hasattr(response, 'version') else "unknown"
except Exception as e:
logger.error(f"获取版本失败: {e}")
return "unknown"
# ==================== Collection 管理 ====================
def _refresh_collections_cache(self):
"""刷新 collection 缓存"""
try:
result = self.client.get_collections()
self._collections_cache = {col.name for col in result.collections}
except Exception:
self._collections_cache = set()
def collection_exists(self, collection_name: str) -> bool:
"""判断 Collection 是否存在"""
return collection_name in self._collections_cache
def create_collection(
self,
collection_name: str,
vector_size: int,
distance: Union[str, models.Distance] = models.Distance.COSINE,
on_disk_payload: bool = True,
recreate: bool = False
) -> bool:
"""
创建 Collection
Args:
collection_name: Collection 名称
vector_size: 向量维度
distance: 距离度量方式 (COSINE / DOT / EUCLID)
on_disk_payload: 是否将 payload 存储在磁盘
recreate: 如果已存在是否重建
Returns:
创建成功返回 True，失败返回 False
"""
try:
if self.collection_exists(collection_name):
if not recreate:
logger.warning(f"⚠️ Collection '{collection_name}' 已存在，跳过创建")
return True
self.delete_collection(collection_name)
logger.info(f"🗑️ 已删除旧 Collection: {collection_name}")
self.client.create_collection(
collection_name=collection_name,
vectors_config=models.VectorParams(
size=vector_size,
distance=distance
),
on_disk_payload=on_disk_payload
)
self._refresh_collections_cache()
logger.info(f"✅ Collection 创建成功: {collection_name} (dim={vector_size})")
return True
except Exception as e:
logger.error(f"❌ 创建 Collection 失败: {e}")
return False
def delete_collection(self, collection_name: str) -> bool:
"""删除 Collection"""
try:
if not self.collection_exists(collection_name):
logger.warning(f"⚠️ Collection '{collection_name}' 不存在")
return True
self.client.delete_collection(collection_name)
self._refresh_collections_cache()
logger.info(f"🗑️ Collection 删除成功: {collection_name}")
return True
except Exception as e:
logger.error(f"❌ 删除 Collection 失败: {e}")
return False
def list_collections(self) -> List[str]:
"""列出所有 Collection 名称"""
self._refresh_collections_cache()
return list(self._collections_cache)
def get_collection_info(self, collection_name: str) -> Optional[Dict]:
"""获取 Collection 的详细信息"""
try:
if not self.collection_exists(collection_name):
logger.warning(f"⚠️ Collection '{collection_name}' 不存在")
return None
info = self.client.get_collection(collection_name)
return {
"name": collection_name,
"vectors_count": info.vectors_count if hasattr(info, 'vectors_count') else 0,
"points_count": info.points_count if hasattr(info, 'points_count') else 0,
"segments_count": info.segments_count if hasattr(info, 'segments_count') else 0,
"status": info.status if hasattr(info, 'status') else "unknown",
"vector_size": info.config.params.vectors.size if hasattr(info, 'config') else 0
}
except Exception as e:
logger.error(f"❌ 获取 Collection 信息失败: {e}")
return None
# ==================== 向量操作 ====================
def upsert(
self,
collection_name: str,
points: List[Dict[str, Any]]
) -> bool:
"""
插入或更新向量
Args:
collection_name: Collection 名称
points: 点数据列表，格式: [{"id": 1, "vector": [...], "payload": {...}}]
Returns:
操作成功返回 True
"""
try:
if not self.collection_exists(collection_name):
logger.error(f"❌ Collection '{collection_name}' 不存在")
return False
# 构建 PointStruct 列表
point_structs = []
for p in points:
point_structs.append(
models.PointStruct(
id=p["id"],
vector=p["vector"],
payload=p.get("payload", {})
)
)
self.client.upsert(
collection_name=collection_name,
points=point_structs
)
logger.info(f"📥 成功插入 {len(points)} 条数据到 '{collection_name}'")
return True
except Exception as e:
logger.error(f"❌ 插入数据失败: {e}")
return False
def search(
self,
collection_name: str,
query_vector: List[float],
limit: int = 10,
score_threshold: Optional[float] = None,
filter_conditions: Optional[Dict] = None,
with_payload: bool = True,
with_vectors: bool = False
) -> List[Dict]:
"""
向量相似度搜索
Args:
collection_name: Collection 名称
query_vector: 查询向量
limit: 返回结果数量
score_threshold: 相似度阈值（只返回分数 >= 此值的结果）
filter_conditions: 过滤条件
with_payload: 是否返回 payload
with_vectors: 是否返回向量
Returns:
搜索结果列表
"""
try:
if not self.collection_exists(collection_name):
logger.error(f"❌ Collection '{collection_name}' 不存在")
return []
# 构建过滤条件
q_filter = None
if filter_conditions:
q_filter = self._build_filter(filter_conditions)
# 执行搜索（适配新版 API）
try:
# 尝试使用新版 API (≥1.7.0)
response = self.client.query_points(
collection_name=collection_name,
query=query_vector,
limit=limit,
score_threshold=score_threshold,
query_filter=q_filter,
with_payload=with_payload,
with_vectors=with_vectors
)
results = response.points
except AttributeError:
# 降级到旧版 API
results = self.client.search(
collection_name=collection_name,
query_vector=query_vector,
limit=limit,
score_threshold=score_threshold,
query_filter=q_filter,
with_payload=with_payload,
with_vectors=with_vectors
)
# 转换为字典格式
output = []
for r in results:
item = {
"id": r.id,
"score": r.score,
"payload": r.payload if with_payload else None
}
if with_vectors:
item["vector"] = r.vector
output.append(item)
logger.info(f"🔍 搜索完成，返回 {len(output)} 条结果")
return output
except Exception as e:
logger.error(f"❌ 搜索失败: {e}")
return []
def delete_points(
self,
collection_name: str,
point_ids: List[int]
) -> bool:
"""根据 ID 删除点"""
try:
if not self.collection_exists(collection_name):
logger.error(f"❌ Collection '{collection_name}' 不存在")
return False
self.client.delete(
collection_name=collection_name,
points_selector=models.PointIdsList(
points=point_ids
)
)
logger.info(f"🗑️ 成功删除 {len(point_ids)} 条数据")
return True
except Exception as e:
logger.error(f"❌ 删除数据失败: {e}")
return False
def delete_by_filter(
self,
collection_name: str,
filter_conditions: Dict
) -> bool:
"""根据条件删除点"""
try:
if not self.collection_exists(collection_name):
logger.error(f"❌ Collection '{collection_name}' 不存在")
return False
q_filter = self._build_filter(filter_conditions)
self.client.delete(
collection_name=collection_name,
points_selector=models.FilterSelector(
filter=q_filter
)
)
logger.info(f"🗑️ 根据条件删除成功")
return True
except Exception as e:
logger.error(f"❌ 条件删除失败: {e}")
return False
def scroll(
self,
collection_name: str,
limit: int = 100,
offset: Optional[int] = None,
with_payload: bool = True,
with_vectors: bool = False
) -> tuple:
"""
分页获取 Collection 中的所有点
Returns:
(points, next_offset)
"""
try:
if not self.collection_exists(collection_name):
logger.error(f"❌ Collection '{collection_name}' 不存在")
return [], None
result = self.client.scroll(
collection_name=collection_name,
limit=limit,
offset=offset,
with_payload=with_payload,
with_vectors=with_vectors
)
points = []
for p in result[0]:
item = {
"id": p.id,
"payload": p.payload if with_payload else None
}
if with_vectors:
item["vector"] = p.vector
points.append(item)
return points, result[1]
except Exception as e:
logger.error(f"❌ 分页查询失败: {e}")
return [], None
def count(self, collection_name: str) -> int:
"""获取 Collection 中的点数量"""
try:
if not self.collection_exists(collection_name):
logger.error(f"❌ Collection '{collection_name}' 不存在")
return 0
result = self.client.count(
collection_name=collection_name,
exact=True
)
return result.count
except Exception as e:
logger.error(f"❌ 统计数量失败: {e}")
return 0
# ==================== 辅助方法 ====================
def _build_filter(self, conditions: Dict) -> models.Filter:
"""
构建 Qdrant 过滤条件
支持的格式:
- {"field": "value"} # 等值匹配
- {"field": {"gt": 10}} # 大于
- {"field": {"gte": 10}} # 大于等于
- {"field": {"lt": 10}} # 小于
- {"field": {"lte": 10}} # 小于等于
- {"field": {"in": [1, 2, 3]}} # 在列表中
- {"field": {"match": "text"}} # 文本匹配
- {"$and": [cond1, cond2]} # 逻辑与
- {"$or": [cond1, cond2]} # 逻辑或
"""
must_conditions = []
should_conditions = []
for key, value in conditions.items():
if key == "$and":
for cond in value:
must_conditions.append(self._build_filter(cond))
elif key == "$or":
for cond in value:
should_conditions.append(self._build_filter(cond))
else:
# 字段条件
if isinstance(value, dict):
# 范围或特殊匹配
if "gt" in value:
must_conditions.append(
models.FieldCondition(
key=key,
range=models.Range(
gt=value["gt"]
)
)
)
elif "gte" in value:
must_conditions.append(
models.FieldCondition(
key=key,
range=models.Range(
gte=value["gte"]
)
)
)
elif "lt" in value:
must_conditions.append(
models.FieldCondition(
key=key,
range=models.Range(
lt=value["lt"]
)
)
)
elif "lte" in value:
must_conditions.append(
models.FieldCondition(
key=key,
range=models.Range(
lte=value["lte"]
)
)
)
elif "in" in value:
must_conditions.append(
models.FieldCondition(
key=key,
match=models.MatchAny(
any=value["in"]
)
)
)
elif "match" in value:
must_conditions.append(
models.FieldCondition(
key=key,
match=models.MatchText(
text=value["match"]
)
)
)
else:
# 等值匹配
must_conditions.append(
models.FieldCondition(
key=key,
match=models.MatchValue(
value=value
)
)
)
if len(must_conditions) == 1 and not should_conditions:
return models.Filter(must=must_conditions)
else:
return models.Filter(
must=must_conditions if must_conditions else None,
should=should_conditions if should_conditions else None
)
# ==================== 关闭连接 ====================
def close(self):
"""关闭客户端连接"""
try:
self.client.close()
logger.info("🔌 客户端已关闭")
except Exception as e:
logger.error(f"❌ 关闭客户端失败: {e}")
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
# ==================== 使用示例 ====================
if __name__ == "__main__":
# 1. 创建管理器实例
manager = QdrantClientManager(host="localhost", port=6333)
# 2. 检查健康状态
if not manager.is_healthy():
print("❌ Qdrant 服务未启动，请先启动容器")
exit(1)
print(f"✅ Qdrant 版本: {manager.get_version()}")
# 3. 创建 Collection
manager.create_collection(
collection_name="test_vectors",
vector_size=4,
distance=models.Distance.COSINE,
recreate=True
)
# 4. 插入数据
points = [
{"id": 1, "vector": [0.1, 0.2, 0.3, 0.4], "payload": {"name": "item_a", "category": "fruit"}},
{"id": 2, "vector": [0.5, 0.6, 0.7, 0.8], "payload": {"name": "item_b", "category": "fruit"}},
{"id": 3, "vector": [0.9, 1.0, 1.1, 1.2], "payload": {"name": "item_c", "category": "vegetable"}},
{"id": 4, "vector": [2.0, 1.8, 1.6, 1.4], "payload": {"name": "item_d", "category": "vegetable"}},
]
manager.upsert("test_vectors", points)
# 5. 执行搜索
results = manager.search(
collection_name="test_vectors",
query_vector=[0.15, 0.25, 0.35, 0.45],
limit=3,
with_payload=True
)
print("\n🔍 搜索结果:")
for r in results:
print(f" ID={r['id']}, 相似度={r['score']:.4f}, Payload={r['payload']}")
# 6. 获取统计信息
count = manager.count("test_vectors")
print(f"\n📊 Collection 中的点数量: {count}")
# 7. 关闭连接
manager.close()
'''
2026-07-19 01:21:40,419 - INFO - HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
2026-07-19 01:21:40,421 - INFO - ✅ Qdrant 客户端初始化成功: localhost:6333
2026-07-19 01:21:40,423 - INFO - HTTP Request: GET http://localhost:6333 "HTTP/1.1 200 OK"
2026-07-19 01:21:40,425 - INFO - HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
2026-07-19 01:21:40,428 - INFO - HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
✅ Qdrant 版本: unknown
2026-07-19 01:21:40,538 - INFO - HTTP Request: PUT http://localhost:6333/collections/test_vectors "HTTP/1.1 200 OK"
2026-07-19 01:21:40,542 - INFO - HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
2026-07-19 01:21:40,542 - INFO - ✅ Collection 创建成功: test_vectors (dim=4)
2026-07-19 01:21:40,565 - INFO - HTTP Request: PUT http://localhost:6333/collections/test_vectors/points?wait=true "HTTP/1.1 200 OK"
2026-07-19 01:21:40,566 - INFO - 📥 成功插入 4 条数据到 'test_vectors'
2026-07-19 01:21:40,575 - INFO - HTTP Request: POST http://localhost:6333/collections/test_vectors/points/query "HTTP/1.1 200 OK"
2026-07-19 01:21:40,576 - INFO - 🔍 搜索完成，返回 3 条结果
2026-07-19 01:21:40,581 - INFO - HTTP Request: POST http://localhost:6333/collections/test_vectors/points/count "HTTP/1.1 200 OK"
2026-07-19 01:21:40,581 - INFO - 🔌 客户端已关闭
🔍 搜索结果:
ID=1, 相似度=0.9980, Payload={'name': 'item_a', 'category': 'fruit'}
ID=2, 相似度=0.9827, Payload={'name': 'item_b', 'category': 'fruit'}
ID=3, 相似度=0.9688, Payload={'name': 'item_c', 'category': 'vegetable'}
'''
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
7
0- 0
已为社区贡献3条内容
所有评论(0)