index doesn't exist

[zhenzi0322]

• milvus: 2.2.1
• pymilvus: 2.2.1

config.py

import os

# ==== Milvus Configuration ===
MILVUS_HOST = os.getenv("MILVUS_HOST", "127.0.0.1")
MILVUS_PORT = int(os.getenv("MILVUS_PORT", "37423"))
VECTOR_DIMENSION = int(os.getenv("VECTOR_DIMENSION", "2048"))
INDEX_FILE_SIZE = int(os.getenv("INDEX_FILE_SIZE", "1024"))
METRIC_TYPE = os.getenv("METRIC_TYPE", "L2")
DEFAULT_TABLE = os.getenv("DEFAULT_TABLE", "test")
TOP_K = int(os.getenv("TOP_K", "10"))

milvus_custom_helpers.py

import os.path
import sys
from config import MILVUS_HOST, MILVUS_PORT, VECTOR_DIMENSION, METRIC_TYPE
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility
from logs import LOGGER


class MilvusCustomHelper:
    def __init__(self, milvus_host=None, milvus_port=None):
        try:
            self.collection = None
            host = milvus_host or MILVUS_HOST
            port = milvus_port or MILVUS_PORT
            connections.connect(host=host, port=port)
            LOGGER.debug(f"Successfully connect to Milvus with IP:{host} and PORT:{port}")
        except Exception as e:
            LOGGER.error(f"Failed to connect Milvus: {e}")
            sys.exit(1)

    def set_collection(self, collection_name):
        try:
            if self.has_collection(collection_name):
                self.collection = Collection(name=collection_name)
            else:
                raise Exception(f"There is no collection named:{collection_name}")
        except Exception as e:
            LOGGER.error(f"Failed to load data to Milvus: {e}")
            sys.exit(1)

    def has_collection(self, collection_name):
        # Return if Milvus has the collection
        try:
            return utility.has_collection(collection_name)
        except Exception as e:
            LOGGER.error(f"Failed to load data to Milvus: {e}")
            sys.exit(1)

    def create_collection(self, collection_name):
        # Create milvus collection if not exists
        try:
            if not self.has_collection(collection_name):
                schema = CollectionSchema([
                    FieldSchema("id", dtype=DataType.INT64, is_primary=True), 
                    FieldSchema("embedding", dtype=DataType.FLOAT_VECTOR, dim=VECTOR_DIMENSION)
                ])
                self.collection = Collection(collection_name, schema)
                LOGGER.debug(f"Create Milvus collection: {collection_name}")
            else:
                self.set_collection(collection_name)
            return "OK"
        except Exception as e:
            LOGGER.error(f"Failed to load data to Milvus: {e}")
            sys.exit(1)

    def insert(self, collection_name, vectors):
        try:
            self.create_collection(collection_name)
            data = [vectors]
            self.set_collection(collection_name)
            mr = self.collection.insert(data)
            ids = mr.primary_keys
            self.collection.load()
            LOGGER.debug(
                f"Insert vectors to Milvus in collection: {collection_name} with {len(vectors)} rows")
            return ids
        except Exception as e:
            LOGGER.error(f"Failed to load data to Milvus: {e}")
            sys.exit(1)

    def insert_vectors_ids(self, collection_name, vectors, ids):
        try:
            self.create_collection(collection_name)
            self.set_collection(collection_name)
            data = [
                ids,
                [vector.tolist() for vector in vectors],
            ]
            mr = self.collection.insert(data=data)
            self.create_index(collection_name=collection_name)
            self.collection.load(_async=True) 
        except Exception as e:
            LOGGER.error(f"Failed to load data to Milvus: {e}")

    def create_index(self, collection_name):
        try:
            self.set_collection(collection_name)
            default_index = {"index_type": "FLAT", "metric_type": METRIC_TYPE, "params": {}}
            status = self.collection.create_index(field_name="embedding", index_params=default_index)
            if not status.code:
                LOGGER.debug(
                    f"Successfully create index in collection:{collection_name} with param:{default_index}")
                return status
            else:
                raise Exception(status.message)
        except Exception as e:
            LOGGER.error(f"Failed to create index: {e}")
            # sys.exit(1)

    def delete_collection(self, collection_name):
        # Delete Milvus collection
        try:
            self.set_collection(collection_name)
            self.collection.drop()
            LOGGER.debug("Successfully drop collection!")
            return "ok"
        except Exception as e:
            LOGGER.error(f"Failed to drop collection: {e}")
            sys.exit(1)

    def search_vectors(self, collection_name, vectors, top_k):
        # Search vector in milvus collection
        try:
            self.set_collection(collection_name)
            search_params = {"metric_type": METRIC_TYPE, "params": {"nprobe": 16}}
            res = self.collection.search(vectors, anns_field="embedding", param=search_params, limit=top_k)
            return res
        except Exception as e:
            LOGGER.error(f"Failed to search vectors in Milvus: {e}")
            sys.exit(1)

    def count(self, collection_name):
        # Get the number of milvus collection
        try:
            self.set_collection(collection_name)
            num = self.collection.num_entities
            LOGGER.debug(f"Successfully get the num:{num} of the collection:{collection_name}")
            return num
        except Exception as e:
            LOGGER.error(f"Failed to count vectors in Milvus: {e}")
            sys.exit(1)

通过多线程调用insert_vectors_ids方法时出现如下的错误：

milvus_custom_helpers.py ｜ insert_vectors_ids ｜ 93 ｜ Failed to load data to Milvus: <MilvusException: (code=1, message=index doesn't exist, collectionID 438220462222213125)>

[xiaofan-luan]

def insert(self, collection_name, vectors):
    try:
        self.create_collection(collection_name)
        data = [vectors]
        self.set_collection(collection_name)
        mr = self.collection.insert(data)
        ids = mr.primary_keys
        self.collection.load()
        LOGGER.debug(
            f"Insert vectors to Milvus in collection: {collection_name} with {len(vectors)} rows")
        return ids
    except Exception as e:
        LOGGER.error(f"Failed to load data to Milvus: {e}")
        sys.exit(1)

remove the load in your insert. you can not load before index creation