向量数据库Milvus字符串查询

        由于项目需要，用到了向量数据库Milvus，刚开始都没有遇到问题，直到一个表的主键是字符串（VARCHAR），在查询时刚好要以该主键作为查询条件，此时会出现异常，特此记录一下。
        记取，字符串查询，构建表达式时要加上单引号,比如下面的'{face_id}'，实在face_id本来就是一个字符串范例了，假如不加会出现如下的异常：
        # pymilvus.exceptions.MilvusException: <MilvusException: (code=65535, message=cannot parse expression: face_id == 2_0, error: invalid expression: face_id == 2_0)>
  详细看下面的代码（milvus_demo.py），其中exists()函数中构建查询表达式时做了特别处理：

1. from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility, Partition
2. import time
3. from datetime import datetime
4. from typing import List
7. #用于测试字符串查询的demo
10. # MILVUS向量数据库地址
11. MILVUS_HOST_ONLINE = '127.0.0.1'
12. MILVUS_PORT = 19530
14. # 检索时返回的匹配内容条数
15. VECTOR_SEARCH_TOP_K = 100
18. class MilvusAvatar:
19.     # table_name 表名
20.     # partition_names  分区名,使用默认即可
21.     def __init__(self, mode, table_name, *, partition_names=["default"], threshold=1.1, client_timeout=3):
22.         self.table_name = table_name
23.         self.partition_names = partition_names
24.         
25.         self.host = MILVUS_HOST_ONLINE
26.         self.port = MILVUS_PORT
27.         self.client_timeout = client_timeout
28.         self.threshold = threshold
29.         self.sess: Collection = None
30.         self.partitions: List[Partition] = []
31.         self.top_k = VECTOR_SEARCH_TOP_K
32.         self.search_params = {"metric_type": "L2", "params": {"nprobe": 256}}
33.         self.create_params = {"metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 2048}}
34.                
35.         self.init()
37.     @property
38.     def fields(self):
39.         fields = [
40.             FieldSchema(name='face_id', dtype=DataType.VARCHAR, max_length=640, is_primary=True, auto_id = False),
41.             FieldSchema(name='media_id', dtype=DataType.INT64),
42.             FieldSchema(name='file_path', dtype=DataType.VARCHAR, max_length=640),  #原图片保存路径
43.             FieldSchema(name='name', dtype=DataType.VARCHAR, max_length=640),  #姓名
44.             FieldSchema(name='count', dtype=DataType.INT64),  #数量
45.             FieldSchema(name='save_path', dtype=DataType.VARCHAR, max_length=640),  #现保存的绝对路径，包含文件名
46.             FieldSchema(name='embedding', dtype=DataType.FLOAT_VECTOR, dim=512)
47.         ]
48.         return fields
50.     @property
51.     def output_fields(self):
52.         return ['face_id','media_id', 'file_path', 'name', 'count', 'save_path','embedding']
54.     def init(self):
55.         try:
56.             connections.connect(host=self.host, port=self.port)  # timeout=3 [cannot set]
57.             if utility.has_collection(self.table_name):
58.                 self.sess = Collection(self.table_name)
59.                 print(f'collection {self.table_name} exists')
60.             else:
61.                 schema = CollectionSchema(self.fields)
62.                 print(f'create collection {self.table_name} {schema}')
63.                 self.sess = Collection(self.table_name, schema)
64.                 self.sess.create_index(field_name="embedding", index_params=self.create_params)
65.             for index in self.partition_names:
66.                 if not self.sess.has_partition(index):
67.                     self.sess.create_partition(index)
68.             self.partitions = [Partition(self.sess, index) for index in self.partition_names]
69.             print('partitions: %s', self.partition_names)
70.             self.sess.load()
71.         except Exception as e:
72.             print(e)
74.         
75.     def query_expr_sync(self, expr, output_fields=None, client_timeout=None):
76.         if client_timeout is None:
77.             client_timeout = self.client_timeout
78.         if not output_fields:
79.             output_fields = self.output_fields
81.         print(f"MilvusAvatar query_expr_sync:{expr},output_fields:{output_fields}")
82.         print(f"MilvusAvatar num_entities:{self.sess.num_entities}")
83.         if self.sess.num_entities == 0:
84.             return []
85.             
86.         return  self.sess.query(partition_names=self.partition_names,
87.                                 output_fields=output_fields,
88.                                 expr=expr,
89.                                 _async= False,
90.                                 offset=0,
91.                                 limit=1000)
92.         
93.     # emb 为一个人脸特征向量
94.     def insert_avatar_sync(self, face_id, media_id, file_path, name, save_path, embedding):
95.         print(f'now insert_avatar {file_path}')
96.         print(f'now insert_avatar {file_path}')
97.                        
98.         data = [[] for _ in range(len(self.sess.schema))]
99.         data[0].append(face_id)
100.         data[1].append(media_id)
101.         data[2].append(file_path)
102.         data[3].append(name)
103.         data[4].append(1)
104.         data[5].append(save_path)
105.         data[6].append(embedding)
107.         # 执行插入操作
108.         try:
109.             print('Inserting into Milvus...')
110.             self.partitions[0].insert(data=data)
111.             print(f'{file_path}')
112.             
113.             print(f"MilvusAvatar insert_avatar num_entities:{self.sess.num_entities}")
114.         except Exception as e:
115.             print(f'Milvus insert media_id:{media_id}, file_path:{file_path} failed: {e}')
116.             print(f'Milvus insert media_id:{media_id}, file_path:{file_path} failed: {e}')
117.             return False
119.         return True   
120.         
121.   
122.     # embs是一个数组
123.     def search_emb_sync(self, embs, expr='', top_k=None, client_timeout=None):
124.         if self.sess is None:
125.             return None
126.    
127.         if not top_k:
128.             top_k = self.top_k
129.         milvus_records = self.sess.search(data=embs, partition_names=self.kb_ids, anns_field="embedding",
130.                                           param=self.search_params, limit=top_k,
131.                                           output_fields=self.output_fields, expr=expr, timeout=client_timeout)
132.         print(f"milvus_records:{milvus_records}")
133.         return milvus_records   
134.         
135.       
136.     def exists(self,face_id):
137.         print(f"exists:{face_id},{type(face_id)}")
138.         # 记住，字符串查询，构建表达式时要加上单引号,比如下面的'{face_id}'，其实face_id本来就是一个字符串类型了，如果不加会出现如下的异常：
139.         # pymilvus.exceptions.MilvusException: <MilvusException: (code=65535, message=cannot parse expression: face_id == 2_0, error: invalid expression: face_id == 2_0)>
140.         res = self.query_expr_sync(expr=f"face_id == '{face_id}'", output_fields=self.output_fields)
141.         #print(f"exists:{res},{len(res)}")
142.         if len(res) > 0:
143.             return True
144.         
145.         return False
146.    
147.    
148.     # 修改照片数   
149.     def add_count(self, face_id):
150.         res = self.query_expr_sync(expr=f"face_id == '{face_id}'", output_fields=self.output_fields)
151.         self.sess.delete(expr=f"face_id == '{face_id}'")
152.         for result in res:
153.             media_id = result['media_id']
154.             file_path = result['file_path']
155.             name = result['name']
156.             count = int(result['count'])
157.             save_path = result['save_path']
158.             embedding = result['embedding']
159.             
160.             data = [[] for _ in range(len(self.sess.schema))]
161.             data[0].append(face_id)
162.             data[1].append(media_id)
163.             data[2].append(file_path)
164.             data[3].append(name)
165.             data[4].append(count + 1)
166.             data[5].append(save_path)
167.             data[6].append(embedding)   
168.             print(f"add_count face_id:{face_id},file_path:{file_path}, count:{count}")
169.             
170.             # 执行插入操作
171.             try:
172.                 print('Inserting into Milvus...')
173.                 self.partitions[0].insert(data=data)
174.             except Exception as e:
175.                 print(f'Milvus insert media_id:{media_id}, file_path:{file_path} failed: {e}')
176.                 return False               
177.         
179.     def delete_collection(self):
180.         print("delete_collection")
181.         self.sess.release()
182.         utility.drop_collection(self.table_name)
184.     def delete_partition(self, partition_name):
185.         print("delete_partition")
186.         part = Partition(self.sess, partition_name)
187.         part.release()
188.         self.sess.drop_partition(partition_name)
189.         
190.     def query_all(self,limit=None):
191.         res = self.sess.query(partition_names = self.partition_names,
192.                                 output_fields = ["face_id","media_id", "name", "count", "save_path"],
193.                                 expr= f"face_id != ''",
194.                                 _async = False,
195.                                 offset = 0,
196.                                 limit = None)
197.                                 
198.         print(res)
199.         return res
201. if __name__ == "__main__":
202.     milvus_avatar= MilvusAvatar("local", "avatar", partition_names=["avatar"])
203.     media_id = 2
204.     index = 0
205.     face_id = f"{media_id}_{index}"
206.     file_path = "/home/data/bbh.jpg"
207.     save_path = "/home/data/bbh_avatar.jpg"
208.     embedding = [i/1000 for i in range(512)]
209.     milvus_avatar.insert_avatar_sync(face_id, media_id, file_path, "bbh", save_path, embedding)
210.     #result = milvus_avatar.query_all()
211.     #print(result)
212.     print(milvus_avatar.exists(face_id))
213.    

复制代码

实行：python milvus_demo.py
假如是针对非字符串字段进行查询，则无需做上面的特别处理。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。