To implement KNN for large datasets without Python loops, compute pairwise squared Euclidean distances via matrix ops, use numpy.argpartition to get top-k partial indices, then optionally sort those k. Batch the queries to limit memory.python
import numpy as np
def knn_numpy_batch(X, Q, k, batch_size=1024, dtype=np.float32):
"""
X: (N, D) reference vectors
Q: (M, D) query vectors
returns: indices (M, k) of nearest neighbors in X (unsorted within k unless sorted=True)
"""
X = np.asarray(X, dtype=dtype)
Q = np.asarray(Q, dtype=dtype)
N, D = X.shape
M = Q.shape[0]
# precompute reference squared norms
X_norm2 = np.sum(X * X, axis=1) # (N,)
indices = np.empty((M, k), dtype=np.int64)
for i in range(0, M, batch_size):
q = Q[i:i+batch_size] # (b, D)
q_norm2 = np.sum(q * q, axis=1) # (b,)
# cross-term: (b, N) = q @ X.T
cross = q.dot(X.T) # (b, N)
# squared distances: ||q-x||^2 = q2 + x2 - 2*cross
dists = q_norm2[:, None] + X_norm2[None, :] - 2.0 * cross # (b, N)
# partial select k smallest distances per row
part_idx = np.argpartition(dists, kth=k-1, axis=1)[:, :k] # (b, k)
# optional: sort the k neighbors (in ascending distance)
rows = np.arange(dists.shape[0])[:, None]
sorted_k = np.argsort(dists[rows, part_idx], axis=1)
indices[i:i+batch_size] = part_idx[rows, sorted_k]
return indices
Key points:- Uses identity ||a-b||^2 = ||a||^2 + ||b||^2 - 2a·b to avoid large (M×N×D) allocations.- Argpartition is O(N) per query for k selection; sorting only k items keeps cost low.- Time: O(M*N*D / vector-efficiency) dominated by matrix multiply; Space: O(batch_size*N) for dists.Memory trade-offs and adaptation:- Full Q batched controls peak memory; reduce batch_size until dists fit (dists size = batch_size × N × 4 bytes for float32).- Convert to float32 (or float16 on GPU) to halve memory.- If N or M too large, also chunk X (split references) and accumulate top-k across chunks by merging partial top-k per query.- For production at scale consider approximate libraries (FAISS, Annoy) or GPU BLAS for much faster matrix multiplies.Edge cases:- k > N (clip k = min(k, N)); identical vectors (ties) handled by stable argpartition; empty inputs validated.