To evaluate the quality (recall) of an HNSW index, you can compare its results against a BFIndex (Brute Force Index). The BFIndex stores vectors as-is and performs an exhaustive search, providing the ground truth for the actual k nearest neighbors.
Initialization:
Use hnswlib.BFIndex(space, dim) to create a non-initialized index. space can be 'l2', 'cosine', or 'ip'.
Key Methods:
init_index(max_elements): Initializes the index. max_elements defines the capacity; exceeding this during insertion will throw an exception.add_items(data, ids=None): Inserts data (numpy array of shape N*dim). ids is an optional numpy array of integer labels.delete_vector(label): Removes the element associated with the given label from search results.knn_query(data, k=1): Performs a batch query for the k closest elements for each vector in data (shape N*dim). Returns a numpy array of shape (N, k) containing labels.save_index(path_to_index): Saves the index to disk.load_index(path_to_index, max_elements=0): Loads an index from disk into an uninitialized index.
import hnswlib
import numpy as np
dim = 32
num_elements = 100000
k = 10
nun_queries = 10
# Generating sample data
data = np.float32(np.random.random((num_elements, dim)))
# Declaring index
hnsw_index = hnswlib.Index(space='l2', dim=dim)
bf_index = hnswlib.BFIndex(space='l2', dim=dim)
# Initing both hnsw and brute force indices
hnsw_index.init_index(max_elements=num_elements, ef_construction=200, M=16)
bf_index.init_index(max_elements=num_elements)
# Controlling the recall for hnsw by setting ef:
hnsw_index.set_ef(200)
# Set number of threads used during batch search/construction
hnsw_index.set_num_threads(1)
hnsw_index.add_items(data)
bf_index.add_items(data)
# Generating query data
query_data = np.float32(np.random.random((nun_queries, dim)))
# Query the elements and measure recall:
labels_hnsw, distances_hnsw = hnsw_index.knn_query(query_data, k)
labels_bf, distances_bf = bf_index.knn_query(query_data, k)
# Measure recall
correct = 0
for i in range(nun_queries):
for label in labels_hnsw[i]:
for correct_label in labels_bf[i]:
if label == correct_label:
correct += 1
break
print("recall is :", float(correct)/(k*nun_queries))