Qdrant合集
Qdrant介绍
- 官网地址 https://github.com/qdrant/qdrant
- web-ui地址 https://github.com/qdrant/qdrant-web-ui
- 官方文档 https://qdrant.tech/documentation/
- Qdrant 是一个开源的向量数据库，专门为高效的向量相似性搜索和存储而设计。
- 集合（Collection）:类似关系型数据库中的表，每个集合定义自己的向量维度和距离度量方式
- 点/向量（Points/Vectors）：基本存储单元，包含：ID + 向量 + Payload（元数据）
- Payload：附加的 JSON 格式数据，用于存储向量相关的元信息，支持过滤和查询
- 距离度量（Distance Metrics）：Cosine（余弦相似度） - 最常用，Euclidean（欧几里得距离），Dot（点积），Manhattan（曼哈顿距离）
- 分片（Shards）：数据水平分割，提高并发和性能
Qdrant安装
windows
下载qdrant-x86_64-pc-windows-msvc.zip,解压，有一个文件qdrant.exe
创建配置文件config.yaml
log_level: INFO
service:
host: 0.0.0.0
http_port: 6333
grpc_port: 6334
api_key: "xxx"
storage:
storage_path: "./storage"
# 如果你想使用内存存储（重启后数据丢失），可以设置为true，但这里我们使用持久化存储
# in_memory: false
# 性能相关配置
performance:
max_search_threads: 4
optimizers:
# 索引阈值，即达到多少点后才创建索引
indexing_threshold: 10000
下载web-ui，解压，有一个打包后的dist文件夹
将dist文件夹中的所有文件放到static目录下
完整目录:
Qdrant\
-qdrant.exe
-storage\
-config\
--config.yaml
-static\运行qdrant.exe启动Qdrant
web-ui地址 http://localhost:6333/dashboard
Qdrant测试使用
# python集成 http连接
from qdrant_client import QdrantClient
from qdrant_client.http.models import VectorParams, Distance, PointStruct, Filter, FieldCondition, MatchValue
# 初始化客户端
client = QdrantClient(
host="localhost",
https=False,
api_key="passxxx", # 如启用
)
# 创建 collection
client.create_collection(
collection_name="gundams",
vectors_config=VectorParams(size=4, distance=Distance.COSINE),
# 向量索引HNSW配置，和payload index是两个东西
hnsw_config={
"m": 8,
"ef_construct": 50,
"full_scan_threshold": 1000 # 小数据集可全扫描
},
# 优化器配置也简化
optimizers_config={
"indexing_threshold": 1000, # 早点建索引
"memmap_threshold": 5000 # 内存映射阈值
}
)
# 插入向量
points = [
PointStruct(
id=1,
vector=[0.1, 0.9, 0.6, 0.4], # 4 维
payload={"ms": "高达F91", "role": "西布克"}
),
PointStruct(
id=2,
vector=[0.3, 0.3, 0.1, 0.9], # 4 维
payload={"ms": "古兰森", "role": "白河愁"}
)
]
client.upsert(collection_name="gundams", points=points)
# 搜索
search_result = client.query_points(
collection_name="gundams",
query=[0.2, 0.8, 0.5, 0.9],
query_filter=Filter(must=[ FieldCondition(key="ms",match=MatchValue(value="高达F91")) ]), # http
with_payload=True,
limit=3
)
for point in search_result.points:
print(point)# python集成 grpc连接
from qdrant_client import QdrantClient
from qdrant_client.grpc import VectorParams, Distance, Filter, FieldCondition, Match, VectorsConfig, \
PointId, PointStruct, Vectors, Vector, Value, Condition
# 初始化客户端
client = QdrantClient(
host="localhost",
api_key="passxxx", # 如启用
port=6333,
grpc_port=6334, # gRPC专用端口
https=False,
timeout=10.0,
prefer_grpc=True # 更高性能（需开放 6334）
)
# 创建 collection
client.create_collection(
collection_name="mobilesuit",
vectors_config=VectorsConfig(params=VectorParams(size=4, distance=Distance.Cosine)),
# 向量索引HNSW配置，和payload index是两个东西
hnsw_config={
"m": 8,
"ef_construct": 50,
"full_scan_threshold": 1000 # 小数据集可全扫描
},
# 优化器配置也简化
optimizers_config={
"indexing_threshold": 1000, # 早点建索引
"memmap_threshold": 5000 # 内存映射阈值
}
)
# 插入向量
points = [
PointStruct(
id=PointId(num=1),
vectors=Vectors(vector=Vector(data=[0.1, 0.9, 0.6, 0.4])), # 4 维
payload={"ms": Value(string_value="高达F91"), "role": Value(string_value="西布克")}
),
PointStruct(
id=PointId(num=2),
vectors=Vectors(vector=Vector(data=[0.3, 0.3, 0.1, 0.9])), # 4 维
payload={"ms": Value(string_value="古兰森"), "role": Value(string_value="白河愁")}
)
]
client.upsert(collection_name="mobilesuit", points=points)
# 搜索
search_result = client.query_points(
collection_name="mobilesuit",
query=[0.3, 0.3, 0.1, 0.9],
# query_filter=Filter(must=[ Condition(field=FieldCondition(key="ms",match=Match(keyword="高达F91"))) ]), # grpc
with_payload=True,
limit=3
)
for point in search_result.points:
print(point)嵌入模型
- 嵌入模型：将文字转换成机器能理解的连续的数值向量
- BGE-large-zh：对中文进行了深度优化的开源文本嵌入模型；基于Transformer架构，输出一个1024维的浮点数向量。通常需要L2归一化
# bge-large-zh向量化脚本demo from transformers import AutoModel, AutoTokenizer import torch model_name = "../bge-large-zh" tokenizer = AutoTokenizer.from_pretrained(model_name) device = "cuda:0" # 加载量化模型 model = AutoModel.from_pretrained( model_name, dtype=torch.float16, # 使用半精度 low_cpu_mem_usage=True, device_map=device ) # 处理文本 def handle_text(text): # 生成张量，张量指将文本编码，告诉程序存在哪些字符，最终生成的向量指文本的语义转化，告诉程序文本所表达的意思 inputs = tokenizer(text, return_tensors="pt", max_length=512, truncation=True).to(device) # 获取向量 with torch.no_grad(): # 禁用梯度 outputs = model(**inputs) embeddings = outputs.last_hidden_state[:, 0] # 取CLS token 提取每个序列的CLS token的隐藏状态作为句子表示 embeddings = torch.nn.functional.normalize(embeddings, p=2, dim=1) # 使用L2范数（欧几里得范数） 在第1维度（隐藏维度）上进行归一化 return embeddings # 处理文本 def handle_texts(texts): result = [] for text in texts: # 生成张量，张量指将文本编码，告诉程序存在哪些字符，最终生成的向量指文本的语义转化，告诉程序文本所表达的意思 inputs = tokenizer(text, return_tensors="pt", max_length=512, truncation=True).to(device) # 获取向量 with torch.no_grad(): # 禁用梯度 outputs = model(**inputs) embeddings = outputs.last_hidden_state[:, 0] # 取CLS token 提取每个序列的CLS token的隐藏状态作为句子表示 embeddings = torch.nn.functional.normalize(embeddings, p=2, dim=1) # 使用L2范数（欧几里得范数） 在第1维度（隐藏维度）上进行归一化 result.append(embeddings) print(embeddings) return result
Qdrant配合嵌入模型使用
from qdrant_client import QdrantClient
from qdrant_client.http.models import VectorParams, Distance, PointStruct
import bge_handle as bge
# 初始化客户端
client = QdrantClient(
host="localhost",
api_key="passxxx", # 如启用
port=6333,
https=False,
timeout=10.0
)
def create_collection():
# 创建 collection
client.create_collection(
collection_name="gundam_event",
vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
# 向量索引HNSW配置，和payload index是两个东西
hnsw_config={
"m": 16,
"ef_construct": 100,
"full_scan_threshold": 1000, # 小数据集可全扫描
"max_indexing_threads": 0 # 0表示自动选择
},
# 优化器配置也简化
optimizers_config={
"indexing_threshold": 10000, # 早点建索引
"memmap_threshold": 10000 # 内存映射阈值
}
)
def insert(orders):
points = []
for order in orders:
order_vector = bge.handle_text(order['历史事件'])
print(order)
point = PointStruct(
id=order['编号'],
vector=order_vector[0].tolist(),
payload=order
)
points.append(point)
client.upsert(collection_name="gundam_event", points=points)
def search(text):
vector = bge.handle_text(text)
search_result = client.query_points(
collection_name="gundam_event",
query=vector[0].tolist(),
limit=2
)
return search_result
order_list = [{
'编号': 1,
'类型': 'UC',
'历史事件': '格里普斯战役（UC 0087-0088），反提坦斯组织“奥古”（AEUG）与吉恩残党“阿克西斯”联合对抗提坦斯。最后提坦斯瓦解，奥古损失惨重，阿克西斯成为新威胁。',
'人物': ['卡缪','西洛克', '科瓦特罗', '哈曼']
}, {
'编号': 2,
'类型': 'SEED',
'历史事件': '血染情人节CE 70，地球联合军秘密部队使用核弹袭击PLANT的农业卫星“尤尼乌斯7号”，造成大量调整者平民死亡。此事件成为全面战争的导火索。扎夫特对地球联合宣战。因《禁止核武条约》及N干扰器存在，战争进入以机动战士和光束武器为主的“MS战时代”。',
'人物': ['基拉','阿斯兰·萨拉','卡嘉莉','乌兹米·纳拉·阿斯哈']
},{
'编号': 3,
'类型': 'UC',
'历史事件': '拉普拉斯之乱（UC 0096），新吉恩残党“带袖的”、联邦激进派、毕斯特财团，围绕“拉普拉斯之盒”（揭露联邦政府腐败的宪章原案）的争夺战。最终宪章真相公开，动摇联邦统治合法性，宇宙住民独立运动获道义支持。',
'人物': ['奥黛丽','巴纳吉·林克斯', '弗尔·弗朗托', '利迪·马瑟纳斯']
}]
create_collection()
insert(order_list)
print(search("“拉普拉斯之盒” 的内容被公开——它并非武器，而是宇宙世纪宪章的原版第15条，承诺将来给予宇宙移民平等的自治权。这一真相动摇了联邦统治的合法性，为宇宙居民争取权利提供了法理依据"))# excel_util
from openpyxl import load_workbook
from typing import List
def read_excel_rows(file_path, start_row, end_row, sheet_name, has_header: bool = True, header_names: List[str] = None):
# 加载工作簿
wb = load_workbook(filename=file_path, data_only=True)
# 获取工作表
if sheet_name:
ws = wb[sheet_name]
else:
ws = wb.active
# 读取表头
headers = header_names
if has_header:
header_row = ws[1]
for cell in header_row:
headers.append(cell.value if cell.value else f"Column_{cell.row}")
# 确保行号在有效范围内
if start_row < 1:
start_row = 1
if end_row > ws.max_row:
end_row = ws.max_row
dict_list = []
# 读取指定行范围
for row_idx in range(start_row, end_row + 1):
row_dict = {}
row = ws[row_idx]
for i, cell in enumerate(row):
if i < len(headers):
row_dict[headers[i]] = cell.value
dict_list.append(row_dict)
return dict_list
def demo():
# 使用示例
header_names = ['编号','类型','历史事件','人物']
data_list = read_excel_rows('../file/qdrant_demo.xlsx', start_row=3, end_row=4, sheet_name='Sheet1', has_header=False, header_names=header_names)
for data in data_list:
print(data)# excel导入qdrant
from qdrant_client import QdrantClient
from qdrant_client.http.models import VectorParams, Distance, PointStruct
import excel_util
import bge_handle as bge
# 初始化客户端
client = QdrantClient(
host="localhost",
api_key="passxxx", # 如启用
port=6333,
https=False,
timeout=10.0,
)
def insert(orders):
points = []
for order in orders:
order_vector = bge.handle_text(order['历史事件'])
print(order)
point = PointStruct(
id=order['编号'],
vector=order_vector[0].tolist(),
payload=order
)
points.append(point)
client.upsert(collection_name="gundam_event", points=points)
def insert_by_excel():
header_names = ['编号','类型','历史事件','人物']
data_list = excel_util.read_excel_rows('../file/qdrant_demo.xlsx', start_row=3, end_row=12, sheet_name='Sheet1', has_header=False, header_names=header_names)
insert(data_list)
# create_collection()
# insert(order_list)
insert_by_excel()
