一篇文章教你学会使用Milvus数据库
·
第一章:向量数据库概述
milvus数据库是用来存储,索引,管理嵌入向量的数据库.非结构化数据通过向量嵌入转换为向量,然后存储到milvus向量数据库中.milvus可以处理万亿级别的向量索引.
1.1milvus和mysql的区别
存储数据方面,mysql存储结构化数据(Excel,SQL表等),milvus存储非结构化数据(图片,音频,视频等)转换来的嵌入向量
运行速度方面,milvus支持gpu加速,为高维向量快速检索优化,在亿级别向量检索可实现毫秒级响应.MySQL不支持高维数据检索,针对结构化数据查询,不支持相似度检索,高维向量检索效果极差
使用场景方面,milvus适用于AI场景,比如说人脸识别,图像视频检索等需要计算向量相似度的场景.
MySQL适用于业务场景,比如订单管理用户管理等需要强一致性的场景.
1.2milvus和MySQL的关联
MySQL存储用户信息,milvus存储向量数据,通过id关联查询实现'检索+业务'一体化
1.3milvus的工作原理
milvus存储索引向量,可以通过计算相似距离来分析数据之间的相关性.
第二章:milvus关键字
2.1collection,field关键字
类似于MySQL的表和字段
2.2建表方法以及表字段
建表的时候要先搭建表结构
2.3表实现
搭建表以后再插入字段
第三章:milvus检索方式
大部分milvus向量使用近似最近邻搜索(ANNS),不指定检索方式,默认为暴力检索
检索方式包括:
3.1FLAT(暴力检索)
检索方式:平铺逐个检索
适用场景:适合在小型,百万级向量索引中使用.
优点:100%精确
缺点:在大数据集中检索效率低
3.2.IVF_FLAT(倒排索引)
检索方式:把索引分组,倒排索引,检索
适用场景:适合在大数据集中使用
优点:中和了检索速度和准确率
3.3.IVF_FQ8
在IVF_FLAT的基础上增加了量化的步骤
检索方式:先分组,对向量数据量化压缩(把四字节的浮点数压缩为一字节的整数表示),然后倒排索引,最后查询处理
适用场景:适合在大数据集中使用
优点:IVF_SQ8是IVF_FLAT的优化版,既保留了查询速度,又节省了空间.
缺点:因为压缩的缘故,精度进一步下降
3.4IVF_PQ
引入了倒排索引和乘积量化,目的是加快检索速度
检索方式:先分簇,每个簇由他的聚类中心表示.输入查询的时候,计算与每个聚类中心的距离,选择最近的几个簇再详细计算.每个簇内的向量被分成了众多子向量,每个子向量只用一个编码表示,这样完成了乘积量化.后续使用倒排索引完成查询.
适用场景:适合在大规模高维向量集使用
优点:IVF_PQ减少了存储空间的使用,同时保留了较高的查询速度和准确性
缺点:因为压缩的缘故,精度进一步下降
3.5HNSW
基于图索引
检索方式:搜索的时候不是逐个搜索,而是沿着线路找相似的站点.
适用场景:对搜索效率有高要求
优点:查询速度效率极高
第四章:相似度衡量指标
4.1欧几里得距离
地图上的距离指的就是欧几里得距离
4.2余弦相似度
取值范围是(-1,1),越接近于1相似度越高,越接近于-1,相似度越低
4.3内积
两个向量的的点积计算,值越大,相似度越高
第五章:代码实现milvus
创建milvus数据库,实现增删改查以及混合检索的操作
"""
演示通过python创建milvus数据库
"""
import random
from pymilvus import MilvusClient, DataType, AnnSearchRequest, RRFRanker, WeightedRanker
data = [
{"id": 0,
"vector": [0.3580376395471989, -0.6023495712049978, 0.18414012509913835, -0.26286205330961354, 0.9029438446296592],
"color": "pink_8682"},
{"id": 1,
"vector": [0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104],
"color": "red_7025"},
{"id": 2,
"vector": [0.43742130801983836, -0.5597502546264526, 0.6457887650909682, 0.7894058910881185, 0.20785793220625592],
"color": "orange_6781"},
{"id": 3, "vector": [0.3172005263489739, 0.9719044792798428, -0.36981146090600725, -0.4860894583077995,
0.95791889146345], "color": "pink_9298"},
{"id": 4, "vector": [0.4452349528804562, -0.8757026943054742, 0.8220779437047674, 0.46406290649483184,
0.30337481143159106], "color": "red_4794"},
{"id": 5, "vector": [0.985825131989184, -0.8144651566660419, 0.6299267002202009, 0.1206906911183383,
-0.1446277761879955], "color": "yellow_4222"},
{"id": 6, "vector": [0.8371977790571115, -0.015764369584852833, -0.31062937026679327, -0.562666951622192,
-0.8984947637863987], "color": "red_9392"},
{"id": 7, "vector": [-0.33445148015177995, -0.2567135004164067, 0.8987539745369246, 0.9402995886420709,
0.5378064918413052], "color": "grey_8510"},
{"id": 8, "vector": [0.39524717779832685, 0.4000257286739164, -0.5890507376891594, -0.8650502298996872,
-0.6140360785406336], "color": "white_9381"},
{"id": 9, "vector": [0.5718280481994695, 0.24070317428066512, -0.3737913482606834, -0.06726932177492717,
-0.6980531615588608], "color": "purple_4976"}
]
# todo 1 创建数据库,返回客户端
def create_db():
client = MilvusClient()
databases = client.list_databases()
if 'milvus_demo' not in databases:
client.create_database('milvus_demo')
print('milvus_demo创建成功')
else:
client.use_database('milvus_demo')
return client
# todo 2 创建表
def create_table():
client = MilvusClient()
client.use_database('milvus_demo')
# 1.定义表结构(集合的schema)
# auto_id:主键字段需要手动赋值(不自动生成)
# enable_dynamic_field :允许插入未在schema中定义额外字段
schema = client.create_schema(auto_id=False, enable_dynamic_filed=True)
# 2.向schame中添加具体的字段(表中插列)
# id 是 主键 类型是64整型
schema.add_field(field_name='id', datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name='vector', datatype=DataType.FLOAT_VECTOR, dim=5)
schema.add_field(field_name='scalar', datatype=DataType.VARCHAR, max_length=256, description='标量字段')
# 3.创建表名(集合)
client.create_collection(collection_name='demo_v1', schema=schema)
print('创建表成功')
# todo 3 定义客户端
client = create_db()
# todo 4 构建索引&添加&删除
def get_indexs():
# 1.初始化索引参数配置对象,用于设置索引相关参数
prepare_index = client.prepare_index_params()
# # 2. 配置向量字段索引下的信息
prepare_index.add_index(field_name='vector', index_type='IVF_FLAT', metric_type='COSINE', index_name='vector_index',
params={'nlist': 128})
client.create_index(collection_name='demo_v1', index_params=prepare_index)
# --------场景2: 查看索引信息
# #1.列出集合中所有索引
result = client.list_indexes(collection_name='demo_v1')
print(result)
# 2.查看指定索引的配置信息
result2 = client.describe_index(collection_name='demo_v1', index_name='vector_index')
print(result2)
# -------- 场景3: 集合的加载与释放
# 1.查看一下当前集合状态
print(f'集合的加载状态:{client.get_load_state(collection_name="demo_v1")}')
# 2.将集合加载到内存,加载后,更新集合的状态
client.load_collection(collection_name='demo_v1')
print(f'结合加载状态:{client.get_load_state(collection_name="demo_v1")}')
# 3.释放集合 ,释放以后,就变成未加载状态
client.release_collection(collection_name='demo_v1')
print(f'结合加载状态:{client.get_load_state(collection_name="demo_v1")}')
# ------场景4:删除索引------------------------------------------
# 1.删除向量索引
client.drop_index(collection_name='demo_v1', index_name='vector_index')
# ------------场景5: 给标量字段添加索引------------------------
# 1.配置标量字段的参数
prepare_index = client.prepare_index_params()
prepare_index.add_index(field_name='scalar', index_name='scalar_index')
# 执行索引操作
client.create_index(collection_name='demo_v1', index_params=prepare_index)
# 3.查看索引信息
print(f'查看索引信息:{client.list_indexes(collection_name="demo_v1")}')
result3 = client.describe_index(collection_name='demo_v1', index_name='scalar_index')
print(f'索引的详细信息{result3}')
# todo 5 表数据操作(实体对象) 的增删改
def get_add_del_update(data):
# ----------------场景一: 创建collection集合(milvus表)-----------------
# 1.直接创建集合 ,无需提前定义schema(结构) 简化方式
# 参1: 集合名称 参2:向量的维度 参3:向量相似度计算方式
client.create_collection(collection_name='demo_v2', dimension=5, metric_type='IP')
# 2.执行upsert(insert+update)操作(存在就更新,不存在就添加)
client.upsert(collection_name="demo_v2", data=data)
# 场景2: 给集合创建分区
# 1. 创建分组 (分组用于对集合数据分类存储,方便管理)
client.create_partition(collection_name='demo_v2', partition_name='partitionA')
# 2.向分区内执行upsert操作
client.upsert(collection_name='demo_v2', data=data, partition_name='partitionA')
client.flush(collection_name='demo_v2')
def get_del(data):
client.create_collection(collection_name='demo_v3', dimension=5, metric_type='IP')
client.create_partition(collection_name='demo_v3', partition_name='partitionA')
client.upsert(collection_name='demo_v3', data=data)
client.upsert(collection_name='demo_v3', data=data, partition_name='partitionA')
print('数据添加成功')
# #--------场景1: 从集合删除数据
res = client.delete(collection_name='demo_v3', filter='id in [5,6]')
print(res)
res2 = client.delete(collection_name='demo_v3', partition_name='partitionA', ids=[2])
print(res2)
def get_query():
# -------场景1: 单一向量查询(根据单个向量找相似结果)
# #search() 执行单一向量查询
# #参1: 集合名 (表名) 参2:查询向量列表(用户的问题转换为向量) 参3:查询最相似的数量(top-k) 参4:查询参数 参5:返回字段列表
result = client.search(
collection_name='demo_v3',
data=[[0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104]],
limit=2,
output_fields=['id', 'vector', 'color'],
metric_type='IP'
)
print(result)
print('* ' * 50)
# -------场景2:批量向量搜索(根据多个向量同时去找相似结果)---------------------
result2 = client.search(
collection_name='demo_v3',
data=[
[0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104],
[0.3172005263489739, 0.9719044792798428, -0.36981146090600725, -0.4860894583077995, 0.95791889146345]
],
limit=2,
output_fields=['id', 'vector', 'color'],
metric_type="IP"
)
print(result2)
# ------场景3: 分区搜索 仅在指定的分区去找相似的结果
result3 = client.search(
collection_name="demo_v2",
data=[[0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104]],
limit=2,
search_params={"metric_type": "IP"},
partition_names=["partitionA"]
)
print(f"result3:{result3}")
# ------------场景4:指定输出字段搜索(仅返回需要的字段,减少数据量)
res3 = client.search(
collection_name="demo_v3",
data=[[0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104]],
limit=2,
search_params={"metric_type": "IP"},
output_fields=["id"]
)
print(res3)
# -------场景5: 带过滤条件的搜索(先过滤,在找相似的结果)-------
res4 = client.search(
collection_name="demo_v3",
data=[[0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104]],
limit=2,
search_params={"metric_type": "IP"},
output_fields=["id", "color"],
filter='color like "red%"'
)
print(f"res4:{res4}")
# --------场景6:范围搜索(仅返回相似度在指定范围的结果)
dict_params = {
'metric_type': 'IP',
"params": {
'radius': 0.8,
'range_dilter': 1
}
}
res5 = client.search(
collection_name="demo_v3",
limit=6,
search_params=dict_params,
output_fields=["id", "color"],
filter='color like "red%"',
data=[[0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104]],
)
print(f"res5:{res5}")
# todo 8 实体对象(表数据)的混合查询 ->准备数据
def complex_query_add_data():
# ----------场景1:定义schema(包含多个向量字段的表结构)
# 1.创建集合 schema ,禁止动态字段(仅允许使用预定义字段) 只能使用你在schema中预定义的字段
schema = client.create_schema(enable_dynamic_filed=False)
# 2.向schema 添加字段
# film_id 电影(id) 64位整数 is_primary 标注为主键
schema.add_field(field_name='film_id', datatype=DataType.INT64, is_primary=True)
# filmVector 电影特征向量 浮点型向量 维度 5
schema.add_field(field_name='filmVector', datatype=DataType.FLOAT_VECTOR, dim=5)
# posterVector 电影海报特征向量 浮点型向量 维度 5
schema.add_field(field_name='posterVector', datatype=DataType.FLOAT_VECTOR, dim=5)
# ---- 场景2: 定义多个向量字段的索引
# 1.初始化索引参数配置对象
prepare_index = client.prepare_index_params()
prepare_index.add_index(field_name='film_id', index_type='INT64')
# 2.为filmVector
# 参1:要创建索引字段名 参2: 索引类型(适合大规模向量检索) 参3:相似度计算方式(欧式距离) 参4:聚类数量
prepare_index.add_index(field_name='filmVector', index_type='IVF_FLAT', metric_type='L2', params={'nlist': 128})
# 3.为posterVector
prepare_index.add_index(field_name='posterVector', index_type='IVF_FLAT', metric_type='COSINE')
# ---- 场景3:创建包含多个向量字段的集合
client.create_collection(collection_name='demo_v4', schema=schema, index_params=prepare_index)
# ----------场景4:生成并插入实体数据(批量添加多条记录)
# 1.初始化实体列表(存储多条数据)
entities = []
for _ in range(1000):
film_id = random.randint(1, 10000)
filmVector = [random.random() for _ in range(5)]
posterVector = [random.random() for _ in range(5)]
entity = {'film_id': film_id, 'filmVector': filmVector, 'posterVector': posterVector}
entities.append(entity)
client.upsert(collection_name='demo_v4', entities=entities)
# todo 9 实体对象(表数据)的混合查询->代码实现
def complex_query():
# --------------场景1:多向量字段混合检索(合并多个搜索结果)
# 1.配置第1个向量字段的搜索要求
query_filmVector = [
[0.8896863042430693, 0.370613100114602, 0.23779315077113428, 0.38227915951132996, 0.5997064603128835]]
# 1.2 配置搜索参数
search_param1 = {
"data": query_filmVector, # 搜索用的向量数据
"anns_field": "filmVector", # 指定搜索向量字段filmVector
"param": {"metric_type": "L2", "nprobe": 10}, # 使用欧式距离 IVF索引参数(查看前10个最相似的簇)
"limit": 2 # 该字段搜索最多返回2条结果
}
# 1.3创建第1个搜索请求对象 ** 把字典值当参数进行传递
request_1 = AnnSearchRequest(**search_param1)
query_posterVector = [
[0.02550758562349764, 0.006085637357292062, 0.5325251250159071, 0.7676432650114147, 0.5521074424751443]]
# 2.配置搜索参数
search_param2 = {
"data": query_posterVector,
"anns_field": "posterVector",
"param": {"metric_type": "COSINE"},
"limit": 2
}
request_2 = AnnSearchRequest(**search_param2)
# 3.配置混合搜索(即: 合并两个字段的搜索结果)
# 3.1组合多个搜索请求对象
res = [request_1, request_2]
# 3.2定义结果排序器
# ranker=RRFRanker(100) (倒数排序融合)
ranker = WeightedRanker(0.7, 0.3) # (加权排序)
# 4.执行混合查询
outputs = client.hybrid_search(
collection_name="demo_v4", # 目标集合名称
reqs=res, # 执行多个搜索请求列表
ranker=ranker, # 用于合并结果的排序器
output_fields=["filmVector", "posterVector"],
limit=2
)
print(f"混合检索结果:{outputs}")
# -----------场景2:遍历抖音检索结果详细
for hits in outputs:
print("Top K结果")
for hit in hits:
print(hit)
if __name__ == '__main__':
# 1.1测试建库
# create_db()
# 1.2测试建表
# create_table()
# get_indexs()
# get_add_del_update(data)
# get_del(data)
# get_query()
# complex_query_add_data()
# complex_query()
client.drop_collection(collection_name="demo_v4")
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
36
0- 0
已为社区贡献4条内容
所有评论(0)