milvus向量数据库详解

文章目录

• 一、连接管理
• 二、创建collection
• 三、删除、加载、释放collection
• 四、查看collection信息
• • 1.检查 collection 是否存在
  • 2.列出当前连接下的所有 collection
• 五、创建partition
• 六、删除、加载、释放partition
• 七、查看partition信息
• • 1.检查 partition 是否存在
  • 2.列出当前 collection 下的所有 partition
• 八、插入数据
• 九、删除数据
• 十、创建、删除索引
• 十一、向量检索
• 十二、其他检索

---

一、连接管理

1. 创建连接

from pymilvus import connections

connections.connect(
	alias="default", 		# 连接别名，可以不写，默认为 default
	host='localhost', 		# milvus ip
	port='19530'			# milvus 端口
	)
	

2. 断开连接

from pymilvus import connections

connections.disconnect("default")

---

二、创建collection

collection 类似于 mysql 的数据表

from pymilvus import CollectionSchema, FieldSchema, DataType, Collection

# 声明 collection 结构 ------------------------------
# 字段1
book_id = FieldSchema(
    name="book_id", 				# 字段名 
    dtype=DataType.INT64, 			# 数据类型，整数，主键必须使用整数
    is_primary=True, 				# 是否为主键
    auto_id=True					# 是否设置为“自增id”
    )
# 字段2
word_count = FieldSchema(
    name="word_count", 				# 字段名
    dtype=DataType.INT64,  			# 数据类型，整数
    )
# 字段3
book_intro = FieldSchema(
    name="book_intro", 				# 字段名
    dtype=DataType.FLOAT_VECTOR, 	# 数据类型，浮点向量
    dim=2							# 向量维度
    )
    
schema = CollectionSchema(
    fields=[book_id, word_count, book_intro],    # 字段加入 collection
    description="Test book search"				 # collection 描述，可以不写，默认为空字符串
    )
collection_name = "book"					     # collection 名称
# ----------------------------------------

# 创建 collection ------------------
collection = Collection(
    name=collection_name, 	        # collection 名称
    schema=schema, 					# collection 结构
    using='default', 				# 创建在哪个连接下，可以不写，默认创建在 default 下
    shards_num=2,					# 集合分片数，可以不写，默认为2
    consistency_level="Bounded"		# 一致性级别，可以不写，默认为 Bounded
    )
# ---------------------------------

ps：详细字段说明见官网
https://milvus.io/cn/docs/v2.0.x/create_collection.md

---

三、删除、加载、释放collection

from pymilvus import utility
from pymilvus import Collection

# 1.删除 book collection
utility.drop_collection("book") 

# 2.加载 book collection 到内存
collection = Collection("book")  
collection.load()

# 3.释放 book collection
collection = Collection("book")
collection.release()

---

四、查看collection信息 1.检查 collection 是否存在

from pymilvus import utility

utility.has_collection("book")

一般用于创建 collection 时做查重检查，举两个例子

from pymilvus import utility

# 1.当 book collection 不存在时就创建
collection_name = "book"		
if not utility.has_collection(collection_name):
	# 声明表结构
	book_id = FieldSchema(name="book_id", dtype=DataType.INT64, is_primary=True, auto_id=True)
	book_intro = FieldSchema(name="book_intro", dtype=DataType.FLOAT_VECTOR, dim=2)
	schema = CollectionSchema(fields=[book_id, book_intro])
	# 创建表
	collection = Collection(name=collection_name, schema=schema)

# 2.创建 book collection 之前先删除重名旧 collection 
collection_name = "book"		
if utility.has_collection(collection_name):
	utility.drop_collection("book")
# 声明表结构
book_id = FieldSchema(name="book_id", dtype=DataType.INT64, is_primary=True, auto_id=True)
book_intro = FieldSchema(name="book_intro", dtype=DataType.FLOAT_VECTOR, dim=2)
schema = CollectionSchema(fields=[book_id, book_intro])
# 创建表
collection = Collection(name=collection_name, schema=schema)

2.列出当前连接下的所有 collection

可以对 collection 进行批量处理操作，举三个例子

from pymilvus import utility
from pymilvus import Collection

# 1.清空 collection 
for collectionName in utility.list_collections():
    utility.drop_collection(collectionName)

# 2.批量加载 collection 到内存
for collectionName in utility.list_collections():
	collection = Collection(collectionName)
	collection.load()
	
# 3.批量释放 collection
for collectionName in utility.list_collections():
	collection = Collection(collectionName)
	collection.release()	

---

五、创建partition

Milvus 允许将大量的向量数据划分成一定数量的 Partition ，可以将搜索和其他操作限制在特定的 Partition 上来提高性能。

from pymilvus import Collection

collection = Collection("book")  # book collection 必须是已存在的
collection.create_partition("novel")

---

六、删除、加载、释放partition

from pymilvus import Collection

# 删除 book collection 下的 novel partition ---------------------
collection = Collection("book")  # book collection 必须是已存在的
collection.drop_partition("novel")
# --------------------------------------------------------------

# 加载 book collection 下的 novel partition ---------------------
# 第一种方式
collection = Collection("book")  # book collection 必须是已存在的
collection.load(["novel"])  # novel partition 必须是已存在的

# 第二种方式
partition = Partition("novel")  # novel partition 必须是已存在的
partition.load()
# --------------------------------------------------------------

# 释放 book collection 下的 novel partition ---------------------
partition = Partition("novel")  # novel partition 必须是已存在的
partition.release()
# --------------------------------------------------------------

---

七、查看partition信息 1.检查 partition 是否存在

from pymilvus import Collection

collection = Collection("book")   # book collection 必须是已存在的
collection.has_partition("novel")

一般用于创建 partition 时做查重检查，举两个例子

from pymilvus import Collection


collection = Collection("book")   # book collection 必须是已存在的

# 1.当 novel partition 不存在时就创建
partition_name = "novel"
if not collection.has_partition(partition_name):
	collection.create_partition(partition_name)

# 2.创建 novel partition 之前先删除重名旧 partition 
partition_name = "novel"
if collection.has_partition(partition_name):
	collection.drop_partition(partition_name)

2.列出当前 collection 下的所有 partition

from pymilvus import Collection

collection = Collection("book")   # book collection 必须是已存在的
collection.partitions

---

八、插入数据

milvus插入数据时接收的格式有些特殊，我把完整的 建表、声明数据、插入数据 流程都写在下面了，你细品（旺柴）

from pymilvus import Collection

# 声明 collection ------------------------------------------------------------
# 字段1【书的主键】
book_id = FieldSchema(
    name="book_id", 				# 字段名 
    dtype=DataType.INT64, 			# 数据类型，整数，主键必须使用整数
    is_primary=True, 				# 是否为主键
    auto_id=True					# 是否设置为“自增id”
    )
# 字段2【字数】
word_count = FieldSchema(
    name="word_count", 				# 字段名
    dtype=DataType.INT64,  			# 数据类型，整数
    )
# 字段3【摘要】
book_intro = FieldSchema(
    name="book_intro", 				# 字段名
    dtype=DataType.FLOAT_VECTOR, 	# 数据类型，浮点向量
    dim=2							# 向量维度
    )
# 字段加入 collection
schema = CollectionSchema(
    fields=[book_id, word_count, book_intro],    
    description="Test book search"				 # collection 描述，可以不写，默认为空字符串
    )

# 创建 collection
collection = Collection(
    name="book", 	        		# collection 名称
    schema=schema, 					# collection 结构
    using='default', 				# 创建在哪个连接下，可以不写，默认创建在 default 下
    shards_num=2,					# 集合分片数，可以不写，默认为2
    consistency_level="Bounded"		# 一致性级别，可以不写，默认为 Bounded
    )
# ------------------------------------------------------------------------

# 声明数据 ----------------------------------------
data = [
	[
		38461,   	# 第一本书的字数
		9476, 		# 第二本书的字数
		66381		# 第三本书的字数
	],
	[
		"《应得的权利》是康奈尔大学哲学系副教授凯特·曼恩的振聋发聩之作",					# 第一本书的摘要
		"讲述母女相伴相守的日常——《日日杂记》中文版首度亮相",							# 第二本书的摘要
		"《5%的改变》这是心理学家李松蔚最近两年来，通过网络进行心理干预的精彩案例合辑",		# 第三本书的摘要
	],
 	[
		[0.86557, 0.27613],			# 第一本书的向量
		[-0.16557, -0.74434],		# 第二本书的向量
		[0.95557, -0.17413],		# 第三本书的向量
	],
]
# -------------------------------------------------

# 插入数据 ------------------------------------------------------------
# 插入到 collection 中
collection = Collection("book")   # book collection 必须是已存在的
mr = collection.insert(data)
# --------------------------------------------------------------------

# 查看入库条数 --------------------------------------------------
# 必须执行一下 collection.num_entities ，否则插入的数据不会被更新出来
# 【不知道为啥，版本bug嘛，还是我的问题】
print("当前入库数量： {}".format(collection.num_entities))
# --------------------------------------------------------------

当然也可以插入数据到 partition 中

# 插入到指定的 partition 中
collection = Collection("book")        # book collection 必须是已存在的
mr = collection.insert(data, "novel")  # novel partition 必须是已存在的，data 格式根据需求改变

---

九、删除数据

milvus只能布尔表达式根据主键id删除数据
ps：关于布尔表达式的解释可以参考这里 https://milvus.io/cn/docs/v2.0.x/boolean.md

from pymilvus import Collection

collection = Collection("book")     # book collection 必须是已存在的
expr = "book_id in [0,1]"           # 删除主键 book_id 为 0 和 1 的数据
collection.delete(expr)

---

十、创建、删除索引

from pymilvus import Collection

# 创建索引
# 为 book 下的 book_intro 字段创建索引 --------------------------------------------------
index_params = {
        "metric_type":"IP",			  # 测量相似度的算法类型，L2（欧几里得距离）或 IP（内积）
        "index_type":"IVF_SQ8",	  	  # 索引的类型，用于加速向量搜索的索引类型
        "params":{"nlist":16384}		  # 该索引类型的参数，构建特定于索引的参数，IVF_SQ8 的 nlist 表示将数据总共分为多少个分块【聚类单元】，取值范围是[1, 65536]
    }

collection = Collection("book")       # book collection 必须是已存在的
collection.create_index(
    field_name="book_intro", 		  # 字段名		  
    index_params=index_params
    )
# ------------------------------------------------------------------------------------

# 删除索引
# 删除 book collection 的索引 ----------------------------------------
collection = Collection("book")      # book collection 必须是已存在的
collection.drop_index()
# ------------------------------------------------------------------

ps：有关创建索引时使用参数的解释可以参考这里 https://milvus.io/cn/docs/v2.0.x/index.md

---

十一、向量检索

检索之前先将数据加载到内存

一般顺序是：

1. 创建 collection
2. 插入数据
3. 创建索引
4. 加载到内存
5. 向量检索
6. 释放内存

from pymilvus import Collection

# 加载 collection
collection = Collection("book")      # Get an existing collection.
collection.load()

# 检索参数
search_params = {
	"metric_type": "IP", 				# 索引的类型，测量相似度的算法类型，L2（欧几里得距离）或 IP（内积）
	"params": {"nprobe": 16}			# 该索引类型的参数，IVF_SQ8 的 nprobe 表示检索时的分块数，取值范围[1, nlist]
}

# 检索
results = collection.search(
	data=[[0.1, 0.2]],                  # 用于搜索的向量
	anns_field="book_intro",            # 在哪个字段下进行相似度检索
	param=search_params, 				# 该索引特有的搜索参数
	limit=10, 							# 输出向量结果数
	expr=None,							# 用于过滤属性的布尔表达式（可选）
	consistency_level="Strong"			# 搜索的一致性级别（可选）
)

# 释放内存
collection.release()

# 查看最相似向量的 primary key 及其距离值
results[0].ids
results[0].distances

ps：有关检索时索参数的解释可以参考这里 https://milvus.io/cn/docs/v2.0.x/index.md

---

十二、其他检索

其他检索方式可以参考这里

1. 混合搜索 https://milvus.io/cn/docs/v2.0.x/hybridsearch.md
2. 结构化匹配 https://milvus.io/cn/docs/v2.0.x/query.md
3. 使用 Time Travel 搜索 https://milvus.io/cn/docs/v2.0.x/timetravel.md