in datastore/providers/milvus_datastore.py [0:0]
def _create_index(self):
# TODO: verify index/search params passed by os.environ
self.index_params = MILVUS_INDEX_PARAMS or None
self.search_params = MILVUS_SEARCH_PARAMS or None
try:
# If no index on the collection, create one
if len(self.col.indexes) == 0:
if self.index_params is not None:
# Convert the string format to JSON format parameters passed by MILVUS_INDEX_PARAMS
self.index_params = json.loads(self.index_params)
logger.info("Create Milvus index: {}".format(self.index_params))
# Create an index on the 'embedding' field with the index params found in init
self.col.create_index(
EMBEDDING_FIELD, index_params=self.index_params
)
else:
# If no index param supplied, to first create an HNSW index for Milvus
try:
i_p = {
"metric_type": "IP",
"index_type": "HNSW",
"params": {"M": 8, "efConstruction": 64},
}
logger.info(
"Attempting creation of Milvus '{}' index".format(
i_p["index_type"]
)
)
self.col.create_index(EMBEDDING_FIELD, index_params=i_p)
self.index_params = i_p
logger.info(
"Creation of Milvus '{}' index successful".format(
i_p["index_type"]
)
)
# If create fails, most likely due to being Zilliz Cloud instance, try to create an AutoIndex
except MilvusException:
logger.info("Attempting creation of Milvus default index")
i_p = {
"metric_type": "IP",
"index_type": "AUTOINDEX",
"params": {},
}
self.col.create_index(EMBEDDING_FIELD, index_params=i_p)
self.index_params = i_p
logger.info("Creation of Milvus default index successful")
# If an index already exists, grab its params
else:
# How about if the first index is not vector index?
for index in self.col.indexes:
idx = index.to_dict()
if idx["field"] == EMBEDDING_FIELD:
logger.info("Index already exists: {}".format(idx))
self.index_params = idx["index_param"]
break
self.col.load()
if self.search_params is not None:
# Convert the string format to JSON format parameters passed by MILVUS_SEARCH_PARAMS
self.search_params = json.loads(self.search_params)
else:
# The default search params
metric_type = "IP"
if "metric_type" in self.index_params:
metric_type = self.index_params["metric_type"]
default_search_params = {
"IVF_FLAT": {"metric_type": metric_type, "params": {"nprobe": 10}},
"IVF_SQ8": {"metric_type": metric_type, "params": {"nprobe": 10}},
"IVF_PQ": {"metric_type": metric_type, "params": {"nprobe": 10}},
"HNSW": {"metric_type": metric_type, "params": {"ef": 10}},
"RHNSW_FLAT": {"metric_type": metric_type, "params": {"ef": 10}},
"RHNSW_SQ": {"metric_type": metric_type, "params": {"ef": 10}},
"RHNSW_PQ": {"metric_type": metric_type, "params": {"ef": 10}},
"IVF_HNSW": {
"metric_type": metric_type,
"params": {"nprobe": 10, "ef": 10},
},
"ANNOY": {"metric_type": metric_type, "params": {"search_k": 10}},
"AUTOINDEX": {"metric_type": metric_type, "params": {}},
}
# Set the search params
self.search_params = default_search_params[
self.index_params["index_type"]
]
logger.info("Milvus search parameters: {}".format(self.search_params))
except Exception as e:
logger.error("Failed to create index, error: {}".format(e))