表 索引 创建

app/api/routes.py

@@ -18,16 +18,15 @@ def hello_name(name):
     })
 
 @api_bp.route('/upload', methods=['PUT'])
-def upload(name):
+def upload():
     # 获取 JSON 格式的请求体数据
     data = request.get_json()
     # 访问具体字段
     bucket = data.get('bucket')
-    image = data.get('image')
-    key = data.get('key')
-
-
+    image2keys = data.get('image2keys')
+    print(bucket)
+    print(image2keys)
     return jsonify({
-        'message': f'Hello, {name}!',
+        'message': f'Hello!',
         'status': 'success'
     })

app/services/image_search.py

@@ -7,16 +7,11 @@ class ImageSearch:
         self.feature_extractor = feature_extractor
         self.milvus = milvus
 
-    # id = product_id
     def image_to_image_search(self, bucket, image, top_k = 100):
         try:
             # 提取查询图像的特征
             vector = self.feature_extractor.extract_features(image)
 
-            # anns_field = embedding
-            # output_fields=["image", "product_id"]
-            # search_params: Dict[str, Any] = {"metric_type": "IP", "params": {"ef": 100}}
-
             results = self.milvus.search(bucket, vector, top_k)
 
             # 处理结果

app/services/milvus.py

@@ -19,13 +19,10 @@ class MilvusClient:
     #     FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=512)  # CLIP ViT-B/32的特征维度为512
     # ]
     @staticmethod
-    def create_new_collection(collection_name, fields, description) -> Collection:
+    def create_collection(collection_name, fields, description):
         schema: CollectionSchema = CollectionSchema(fields, description)
-
         # 创建集合
-        collection = Collection(name=collection_name, schema=schema)
-
-        return collection
+        Collection(name=collection_name, schema=schema)
 
     @staticmethod
     def get_collection(collection_name):
@@ -38,9 +35,8 @@ class MilvusClient:
     #     "metric_type": "IP",  # 内积相似度
     #     "params": {"M": 16, "efConstruction": 200}
     # }
-    # def create_index(self, index_params) -> None:
-    #     if self.collection is not None:
-    #         self.collection.create_index(field_name="embedding", index_params=index_params)
+    def create_index(self, collection_name, field_name, index_params) -> None:
+        self.get_collection(collection_name).create_index(field_name=field_name, index_params=index_params)
 
     # anns_field = embedding
     # output_fields=["image", "product_id"]

tests/test_milvus.py

+import unittest
+from typing import List
+
+from pymilvus import FieldSchema, DataType
+
+from app.services.milvus import MilvusClient
+
+
+class TestCreateCollectionFunction(unittest.TestCase):
+    def test_create_collection(self):
+        fields: List[FieldSchema] = [
+            FieldSchema(name="image", dtype=DataType.VARCHAR, max_length=256, is_primary=True, auto_id=False),
+            FieldSchema(name="key", dtype=DataType.VARCHAR, max_length=256),
+            FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=512)  # CLIP ViT-B/32的特征维度为512
+        ]
+        MilvusClient().connect().create_collection("pc3", fields, "PC3 图片向量存储")
+
+    def test_create_index(self):
+        index_params = {
+            "index_type": "HNSW",
+            "metric_type": "IP",  # 内积相似度
+            "params": {"M": 16, "efConstruction": 200}
+        }
+        MilvusClient().connect().create_index("pc3", "vector", index_params)
+
+
+
+if __name__ == '__main__':
+    unittest.main()