Milvus | Python |
DataType.INT64 | numpy.int64 |
DataType.INT32 | numpy.int32 |
DataType.INT16 | numpy.int16 |
DataType.BOOL | Boolean |
DataType.FLOAT | numpy.float32 |
DataType.DOUBLE | numpy.double |
DataType.ARRAY | list |
DataType.VARCHAR | str |
DataType.JSON | dict |
FLOAT_VECTOR(浮点数向量) | numpy.ndarray or list (元素为numpy.float) |
(1)连接向量数据库:
from pymilvus import connections,db
connections.connect("default",host="localhost",prot="19530")
(2)删除现有集合:
from pymilvus import utility
utility.drop_collection('ITEM_INFO')
(3)字段常用数据类型创建:
from pymilvus import FieldSchema, DataType
# 创建数据类型为INT64的主键,且主键ID自动递增
ID_col = FieldSchema(
name="ID",
dtype=DataType.INT64,
is_primary=True,
description='主键、序号',
auto_id=True)
# 创建一个768维的浮点向量
vec_col = FieldSchema(
name="itemVec",
dtype=DataType.FLOAT_VECTOR,
dim=768,
description = '向量')
# 创建一个长度128维的字符串
name_col = FieldSchema(
name = 'itemName',
dtype = DataType.VARCHAR,
max_length = 128,
description = '商品名称')
# 创建double类型的浮点数
price_col = FieldSchema(
name = 'itemPrice',
dtype = DataType.DOUBLE,
description = '价格')
# 创建json格式的字段
keywords_col = FieldSchema(
name = 'keywords',
dtype = DataType.JSON)
(4)创建集合
from pymilvus import CollectionSchema, Collection, connections
# 集合所包含的字段
schema = CollectionSchema(
fields=[ID_col,vec_col,name_col,price_col,keywords_col],
description="商品信息表",
enable_dynamic_field=True
)
# 集合名称
collection_name = "GOODS_INFO"
# 构建集合后断开连接
collection = Collection(
name=collection_name,
schema=schema,
using='default',
shards_num=2
)
connections.disconnect("default")
(5)构建索引
Milvus中构建索引的数据量最小为1024,满足2的幂次方,数据量不满1024时必须使用1024构建索引。
from pymilvus import Collection
# 连接集合
collection = Collection("GOODS_INFO")
# 构建索引所需的参数
index_params = {
"metric_type":"L2",
"index_type":"IVF_FLAT",
"params":{"nlist":1024}
}
# 在向量上构建索引
collection.create_index(
field_name="itemVec",
index_params=index_params)
(6)发布集合
from pymilvus import Collection, utility, connections
# 连接集合
collection = Collection("GOODS_INFO")
# 发布集合
collection.load()
# 检查集合上线状态
utility.load_state("GOODS_INFO")
# 断开与向量数据库的连接
connections.disconnect('default')
1、假设集合中有一字段A的数据类型为整型或浮点型,则可以
expr = 'A > 0 && A <= 10'
2、假设集合中有一字段A为任意数据类型,则可以
# a,b,c,d为任意数据类型,可与A相同或不同
expr = 'A in [a,b,c,d]'
expr = 'A not in [a,b,c,d]'
3、假设集合中有一字段A为字符类型,则可以?# 在条件过滤中字符串要添加引号确定
# 在条件过滤中字符串要添加引号确定
expr = "A == 'Hello World!'"
4、假设集合中有一字段A为字符类型,则可以
# 从A开头为123的向量中进行向量检索,目前milvus只支持字符前缀匹配,不支持'1%23'或'%123'
expr = "A like '123%'"
5、假设集合中有一字段A为json类型,则可以
# A = {"x": [1,2,3]}
expr = 'json_contains(A["x"], 1)' # ==> true
expr = 'json_contains(A["x"], "a")' # ==> false
# A = {"x": [[1,2,3], [4,5,6], [7,8,9]]}
expr = 'json_contains(A["x"], [1,2,3])' # ==> true
expr = 'json_contains(A["x"], [3,2,1])' # ==> false
# A = {"x": [1,2,3,4,5,7,8]}
expr = 'json_contains_all(A["x"], [1,2,8])' # ==> true
expr = 'json_contains_all(A["x"], [4,5,6])' # ==> false 6 is not exists
# A = {"x": [1,2,3,4,5,7,8]}
expr = 'json_contains_any(A["x"], [1,2,8])' # ==> true
expr = 'json_contains_any(A["x"], [4,5,6])' # ==> true
expr = 'json_contains_any(A["x"], [6,9])' # ==> false
6、假设集合中有一字段A为list类型,则可以
# 该方法为上述方法2、的逆方法
# A: [1,2,3]
expr = 'array_contains(A, 1)' # ==> true
expr = 'array_contains(A, "a")' # ==> false
# A: [1,2,3,4,5,7,8]
expr = 'array_contains_all(A, [1,2,8])' # ==> true
expr = 'array_contains_all(A, [4,5,6])' # ==> false 6 is not exists
# A: [1,2,3,4,5,7,8]
expr = 'array_contains_any(A, [1,2,8])' # ==> true
expr = 'array_contains_any(A, [4,5,6])' # ==> true
expr = 'array_contains_any(A, [6,9])' # ==> false
# A: [1,2,3,4,5,7,8]
expr = 'array_length(int_array) == 7' # ==> true
假设有一集合,集合当中包含V、A、B、C四个字段:
V:浮点向量
A:向量的字符表达形式
B:A的属性1
C:A的属性2
现有一查询向量S、过滤条件E1和过滤条件E2(E1、E2均为列表),集合需要返回S与V相似度最高,且B包含于E1、C包含于E2的字符名称A。如果采用上述1、2、3、4方法中较基础的数据类型存表会导致查询速度和代码复杂度都较高。假设E1 = [1,2,3],E2 = [4,5,6],则对于同一条向量,需要做3 x 3 = 9次过滤。若使用5、6方法中的数据类型存表则可以将过滤次数降为2次。
例如
# B和C为字符串类型
expr = '(1 in B && 4 in C) or (1 in B && 5 in C) or (1 in B && 6 in C) ...'
# B和C为keywords中的元素,keywords为json格式
expr = "json_contains_any(keywords['B'], E1) && json_contains_any(keywords['C'], E2)"
# B和C均为列表
expr = "array_contains_any(B, E1) && json_contains_any(C, E2)"
from pymilvus import Collection, connections
# 连接数据库,连接集合,将集合数据上传至内存
connections.connect("default",host="localhost",prot="19530")
knowledge = Collection('GOODS_INFO', using='default')
knowledge.load()
# 查询参数
search_params = {
"metric_type": "L2",
"offset": 0,
"ignore_growing": False,
"params": {
"nprobe": 10,
# "radius": 10, # 对于L2,返回距离在[5,10)之间的向量
# "range_filter": 5.0 # 对于IP,返回相似度在(0.8,1]之间的向量
}
}
# 进行向量查询
results = knowledge.search(
data=query, #带查询向量
anns_field='itemVec', #表中向量字段名称
param=search_params, #查询参数
limit=5, #返回数量
expr=expr, #过滤条件,expr = None或''时,默认为不使用条件过滤
output_fields=['itemName', 'itemPrice', 'keywords'] #返回的字段名称)
# 断开数据库连接
connections.remove_connection('default')