| name | vector-index-tuning |
| description | Optimize vector index performance for latency, recall, and memory. Use when tuning HNSW parameters, selecting quantization strategies, or scaling vector search infrastructure. |
Vector Index Tuning
Guide to optimizing vector indexes for production performance.
When to Use This Skill
- Tuning HNSW parameters
- Implementing quantization
- Optimizing memory usage
- Reducing search latency
- Balancing recall vs speed
- Scaling to billions of vectors
Core Concepts
1. Index Type Selection
Data Size Recommended Index
────────────────────────────────────────
< 10K vectors → Flat (exact search)
10K - 1M → HNSW
1M - 100M → HNSW + Quantization
> 100M → IVF + PQ or DiskANN
2. HNSW Parameters
| Parameter | Default | Effect |
|---|
| M | 16 | Connections per node, ↑ = better recall, more memory |
| efConstruction | 100 | Build quality, ↑ = better index, slower build |
| efSearch | 50 | Search quality, ↑ = better recall, slower search |
3. Quantization Types
Full Precision (FP32): 4 bytes × dimensions
Half Precision (FP16): 2 bytes × dimensions
INT8 Scalar: 1 byte × dimensions
Product Quantization: ~32-64 bytes total
Binary: dimensions/8 bytes
Templates
Template 1: HNSW Parameter Tuning
import numpy as np
from typing import List, Tuple
import time
def benchmark_hnsw_parameters(
vectors: np.ndarray,
queries: np.ndarray,
ground_truth: np.ndarray,
m_values: List[int] = [8, 16, 32, 64],
ef_construction_values: List[int] = [64, 128, 256],
ef_search_values: List[int] = [32, 64, 128, 256]
) -> List[dict]:
"""Benchmark different HNSW configurations."""
import hnswlib
results = []
dim = vectors.shape[1]
n = vectors.shape[0]
for m in m_values:
for ef_construction in ef_construction_values:
index = hnswlib.Index(space='cosine', dim=dim)
index.init_index(max_elements=n, M=m, ef_construction=ef_construction)
build_start = time.time()
index.add_items(vectors)
build_time = time.time() - build_start
memory_bytes = index.element_count * (
dim * 4 +
m * 2 * 4
)
for ef_search in ef_search_values:
index.set_ef(ef_search)
search_start = time.time()
labels, distances = index.knn_query(queries, k=10)
search_time = time.time() - search_start
recall = calculate_recall(labels, ground_truth, k=10)
results.append({
"M": m,
"ef_construction": ef_construction,
"ef_search": ef_search,
"build_time_s": build_time,
"search_time_ms": search_time * 1000 / len(queries),
"recall@10": recall,
"memory_mb": memory_bytes / 1024 / 1024
})
return results
def calculate_recall(predictions: np.ndarray, ground_truth: np.ndarray, k: int) -> float:
"""Calculate recall@k."""
correct = 0
for pred, truth in zip(predictions, ground_truth):
correct += len(set(pred[:k]) & set(truth[:k]))
return correct / (len(predictions) * k)
def recommend_hnsw_params(
num_vectors: int,
target_recall: float = 0.95,
max_latency_ms: float = 10,
available_memory_gb: float = 8
) -> dict:
"""Recommend HNSW parameters based on requirements."""
if num_vectors < 100_000:
m = 16
ef_construction = 100
elif num_vectors < 1_000_000:
m = 32
ef_construction = 200
else:
m = 48
ef_construction = 256
if target_recall >= 0.99:
ef_search = 256
elif target_recall >= 0.95:
ef_search = 128
else:
ef_search = 64
return {
"M": m,
"ef_construction": ef_construction,
"ef_search": ef_search,
"notes": f"Estimated for {num_vectors:,} vectors, {target_recall:.0%} recall"
}
Template 2: Quantization Strategies
import numpy as np
from typing import Optional
class VectorQuantizer:
"""Quantization strategies for vector compression."""
@staticmethod
def scalar_quantize_int8(
vectors: np.ndarray,
min_val: Optional[float] = None,
max_val: Optional[float] = None
) -> Tuple[np.ndarray, dict]:
"""Scalar quantization to INT8."""
if min_val is None:
min_val = vectors.min()
if max_val is None:
max_val = vectors.max()
scale = 255.0 / (max_val - min_val)
quantized = np.clip(
np.round((vectors - min_val) * scale),
0, 255
).astype(np.uint8)
params = {"min_val": min_val, "max_val": max_val, "scale": scale}
return quantized, params
@staticmethod
def dequantize_int8(
quantized: np.ndarray,
params: dict
) -> np.ndarray:
"""Dequantize INT8 vectors."""
return quantized.astype(np.float32) / params[] + params[]
() -> [np.ndarray, ]:
sklearn.cluster KMeans
n, dim = vectors.shape
dim % n_subvectors ==
subvector_dim = dim // n_subvectors
codebooks = []
codes = np.zeros((n, n_subvectors), dtype=np.uint8)
i (n_subvectors):
start = i * subvector_dim
end = (i + ) * subvector_dim
subvectors = vectors[:, start:end]
kmeans = KMeans(n_clusters=n_centroids, random_state=)
codes[:, i] = kmeans.fit_predict(subvectors)
codebooks.append(kmeans.cluster_centers_)
params = {
: codebooks,
: n_subvectors,
: subvector_dim
}
codes, params
() -> np.ndarray:
binary = (vectors > ).astype(np.uint8)
n, dim = vectors.shape
packed_dim = (dim + ) //
packed = np.zeros((n, packed_dim), dtype=np.uint8)
i (dim):
byte_idx = i //
bit_idx = i %
packed[:, byte_idx] |= (binary[:, i] << bit_idx)
packed
() -> :
bytes_per_dimension = {
: ,
: ,
: ,
: ,
:
}
vector_bytes = num_vectors * dimensions * bytes_per_dimension[quantization]
index_type == :
index_bytes = num_vectors * hnsw_m * *
index_type == :
index_bytes = num_vectors * + * dimensions *
:
index_bytes =
total_bytes = vector_bytes + index_bytes
{
: vector_bytes / / ,
: index_bytes / / ,
: total_bytes / / ,
: total_bytes / / /
}
Template 3: Qdrant Index Configuration
from qdrant_client import QdrantClient
from qdrant_client.http import models
def create_optimized_collection(
client: QdrantClient,
collection_name: str,
vector_size: int,
num_vectors: int,
optimize_for: str = "balanced"
) -> None:
"""Create collection with optimized settings."""
hnsw_configs = {
"recall": models.HnswConfigDiff(m=32, ef_construct=256),
"speed": models.HnswConfigDiff(m=16, ef_construct=64),
"balanced": models.HnswConfigDiff(m=16, ef_construct=128),
"memory": models.HnswConfigDiff(m=8, ef_construct=64)
}
quantization_configs = {
"recall": None,
"speed": models.ScalarQuantization(
scalar=models.ScalarQuantizationConfig(
type=models.ScalarType.INT8,
quantile=0.99,
always_ram=True
)
),
"balanced": models.ScalarQuantization(
scalar=models.ScalarQuantizationConfig(
type=models.ScalarType.INT8,
quantile=0.99,
always_ram=False
)
),
: models.ProductQuantization(
product=models.ProductQuantizationConfig(
compression=models.CompressionRatio.X16,
always_ram=
)
)
}
optimizer_configs = {
: models.OptimizersConfigDiff(
indexing_threshold=,
memmap_threshold=
),
: models.OptimizersConfigDiff(
indexing_threshold=,
memmap_threshold=
),
: models.OptimizersConfigDiff(
indexing_threshold=,
memmap_threshold=
),
: models.OptimizersConfigDiff(
indexing_threshold=,
memmap_threshold=
)
}
client.create_collection(
collection_name=collection_name,
vectors_config=models.VectorParams(
size=vector_size,
distance=models.Distance.COSINE
),
hnsw_config=hnsw_configs[optimize_for],
quantization_config=quantization_configs[optimize_for],
optimizers_config=optimizer_configs[optimize_for]
)
() -> :
target_recall >= :
search_params = models.SearchParams(
hnsw_ef=,
exact=,
quantization=models.QuantizationSearchParams(
ignore=,
rescore=
)
)
target_recall >= :
search_params = models.SearchParams(
hnsw_ef=,
exact=,
quantization=models.QuantizationSearchParams(
ignore=,
rescore=,
oversampling=
)
)
:
search_params = models.SearchParams(
hnsw_ef=,
exact=,
quantization=models.QuantizationSearchParams(
ignore=,
rescore=
)
)
search_params
Template 4: Performance Monitoring
import time
from dataclasses import dataclass
from typing import List
import numpy as np
@dataclass
class SearchMetrics:
latency_p50_ms: float
latency_p95_ms: float
latency_p99_ms: float
recall: float
qps: float
class VectorSearchMonitor:
"""Monitor vector search performance."""
def __init__(self, ground_truth_fn=None):
self.latencies = []
self.recalls = []
self.ground_truth_fn = ground_truth_fn
def measure_search(
self,
search_fn,
query_vectors: np.ndarray,
k: int = 10,
num_iterations: int = 100
) -> SearchMetrics:
"""Benchmark search performance."""
latencies = []
for _ in range(num_iterations):
for query in query_vectors:
start = time.perf_counter()
results = search_fn(query, k=k)
latency = (time.perf_counter() - start) * 1000
latencies.append(latency)
latencies = np.array(latencies)
total_queries = num_iterations * len(query_vectors)
total_time = sum(latencies) /
SearchMetrics(
latency_p50_ms=np.percentile(latencies, ),
latency_p95_ms=np.percentile(latencies, ),
latency_p99_ms=np.percentile(latencies, ),
recall=._calculate_recall(search_fn, query_vectors, k) .ground_truth_fn ,
qps=total_queries / total_time
)
() -> :
.ground_truth_fn:
correct =
total =
query queries:
predicted = (search_fn(query, k=k))
actual = (.ground_truth_fn(query, k=k))
correct += (predicted & actual)
total += k
correct / total
() -> :
results = {}
batch_size batch_sizes:
times = []
i (, (vectors), batch_size):
batch = vectors[i:i + batch_size]
start = time.perf_counter()
build_fn(batch)
times.append(time.perf_counter() - start)
results[batch_size] = {
: np.mean(times),
: batch_size / np.mean(times)
}
results
Best Practices
Do's
- Benchmark with real queries - Synthetic may not represent production
- Monitor recall continuously - Can degrade with data drift
- Start with defaults - Tune only when needed
- Use quantization - Significant memory savings
- Consider tiered storage - Hot/cold data separation
Don'ts
- Don't over-optimize early - Profile first
- Don't ignore build time - Index updates have cost
- Don't forget reindexing - Plan for maintenance
- Don't skip warming - Cold indexes are slow
Resources
Parent Hub
