TL;DR — This post walks you through building a pure‑Python HNSW index that supports dynamic insertion, deletion, and k‑NN queries, giving you a concrete project that showcases algorithmic depth, performance engineering, and production‑ready design. You’ll end up with runnable code, test harnesses, and a clear roadmap to scale the prototype into a real‑world service.

Building a vector similarity search engine from scratch is a proven way to demonstrate systems thinking, algorithmic skill, and the ability to ship production‑quality code. In this guide you will implement a Hierarchical Navigable Small World (HNSW) graph entirely in Python, without relying on external C++ libraries. The implementation will handle dynamic insertion, deletion, and neighbor queries, and it will be packaged so that you can run it locally, test its correctness, and extend it toward a production system.

Why This Project Stands Out on a CV

  • Algorithmic depth – You will write the core HNSW insertion, layer selection, and pruning logic, showing you understand probabilistic skip‑list structures and graph‑based nearest‑neighbor search.
  • Data‑structure engineering – Designing mutable graph nodes, layer hierarchies, and neighbor lists in pure Python demonstrates strong command of object‑oriented design and memory layout.
  • Performance awareness – By profiling with cProfile and comparing against faiss or nmslib, you prove you can reason about constant factors, cache locality, and algorithmic complexity.
  • Production readiness – The project includes unit tests, a CLI runner, and a roadmap for persistence, scaling, and observability—skills that map directly to senior or staff engineer roles.
  • Domain relevance – Vector search underpins recommendation systems, image retrieval, and anomaly detection; showcasing a working index signals you can contribute to these high‑impact areas.

Architecture Overview

The system is composed of the following components:

  • Vector representation – Each item is a numpy.ndarray (or a plain Python list) of fixed dimensionality.
  • Node – A data class that stores the vector, a unique identifier, and a list of Layer objects.
  • Layer – A level in the HNSW hierarchy; each layer maintains a list of bidirectional edges to other nodes.
  • Graph – The top‑level container that holds all nodes, the entry point, and the current maximum layer.
  • Insertion algorithm – Randomly assigns a node to a layer based on a geometric distribution, then performs a greedy search from the top layer down, linking the node to M nearest neighbors at each level.
  • Deletion algorithm – Removes a node and rewires its neighbors to maintain connectivity, using a “repair” step that re‑links the closest remaining nodes.
  • Search – Performs a multi‑layer traversal: start at the top, descend to the bottom, and at each level keep the best ef candidates.

A simplified textual diagram:

[Entry Point] ---> Layer 3 (top)
                     |
                     v
                 Layer 2
                     |
                     v
                 Layer 1 (bottom)
                     |
                     v
               [All Nodes]

Each arrow represents a directed edge; edges are symmetric in practice.

Building It Step by Step

Below are the essential code snippets. The full script is named hnsw.py.

Step 1 – Define the Node and Layer classes

from __future__ import annotations
import math
import random
from dataclasses import dataclass, field
from typing import List, Optional, Tuple
import numpy as np

@dataclass
class Layer:
    neighbors: List[int] = field(default_factory=list)

@dataclass
class Node:
    id: int
    vector: np.ndarray
    layers: List[Layer]  # layers[0] is the bottom layer

Step 2 – Create the HNSW Graph

class HNSWGraph:
    def __init__(self, M: int = 16, ef_construction: int = 200, seed: int = 42):
        self.M = M
        self.ef_construction = ef_construction
        self.rng = random.Random(seed)
        self.nodes: List[Node] = []
        self.entry_point: Optional[int] = None
        self.max_layer: int = -1

    def _random_layer(self) -> int:
        """Geometric distribution to decide how many layers a new node spans."""
        return int(math.floor(-math.log(self.rng.random()) / math.log(self.M)))

Step 3 – Insertion

    def add(self, vector: np.ndarray) -> int:
        node_id = len(self.nodes)
        new_node = Node(id=node_id, vector=vector, layers=[Layer() for _ in range(self._random_layer() + 1)])
        self.nodes.append(new_node)

        if self.entry_point is None:
            self.entry_point = node_id
            self.max_layer = len(new_node.layers) - 1
            return node_id

        # Search from top layer down to the layer where new_node starts
        ep = self.entry_point
        for layer_idx in range(self.max_layer, len(new_node.layers) - 1, -1):
            ep = self._search_layer(vector, ep, 1, layer_idx)[0]

        # Insert at each layer the node belongs to
        for layer_idx in range(len(new_node.layers) - 1, -1, -1):
            neighbors = self._search_layer(vector, ep, self.ef_construction, layer_idx)
            self._link(new_node, layer_idx, neighbors)
            if layer_idx > 0:
                ep = neighbors[0][0]  # best neighbor becomes entry for next layer

        # Update max_layer if necessary
        if len(new_node.layers) - 1 > self.max_layer:
            self.max_layer = len(new_node.layers) - 1
            self.entry_point = node_id
        return node_id

    def _search_layer(self, query: np.ndarray, entry: int, ef: int, layer_idx: int) -> List[Tuple[int, float]]:
        """Return the `ef` closest nodes to `query` in the given layer."""
        visited = {entry}
        candidates = [(self._distance(query, self.nodes[entry].vector), entry)]
        best = []
        while candidates:
            dist, cur = candidates.pop()
            if len(best) < ef or dist < best[-1][0]:
                best.append((dist, cur))
                best.sort(key=lambda x: x[0])
                best = best[:ef]
            for nb in self.nodes[cur].layers[layer_idx].neighbors:
                if nb not in visited:
                    visited.add(nb)
                    nb_dist = self._distance(query, self.nodes[nb].vector)
                    if len(best) < ef or nb_dist < best[-1][0]:
                        candidates.append((nb_dist, nb))
        return best

    def _link(self, node: Node, layer_idx: int, neighbors: List[Tuple[int, float]]):
        """Add bidirectional edges between `node` and its selected neighbors."""
        for nb_id, _ in neighbors:
            if nb_id == node.id:
                continue
            # Add to node's neighbor list
            node.layers[layer_idx].neighbors.append(nb_id)
            # Add to neighbor's list
            self.nodes[nb_id].layers[layer_idx].neighbors.append(node.id)
            # Prune to keep at most M edges per node per layer (optional)
            if len(node.layers[layer_idx].neighbors) > self.M:
                node.layers[layer_idx].neighbors = node.layers[layer_idx].neighbors[-self.M:]

    @staticmethod
    def _distance(a: np.ndarray, b: np.ndarray) -> float:
        return float(np.linalg.norm(a - b))

Step 4 – Deletion

    def remove(self, node_id: int):
        """Delete a node and repair its neighbors."""
        if node_id not in {n.id for n in self.nodes}:
            raise KeyError(f"Node {node_id} not found")
        node = self.nodes[node_id]
        # For each layer, remove node from neighbors' lists and rewire
        for layer_idx, layer in enumerate(node.layers):
            for nb_id in layer.neighbors:
                nb = self.nodes[nb_id]
                # Remove node from neighbor's list
                nb.layers[layer_idx].neighbors = [x for x in nb.layers[layer_idx].neighbors if x != node_id]
                # Optional: rewire to maintain connectivity
                # Here we simply leave the graph; more sophisticated repair can be added.
        # Remove node from graph
        self.nodes[node_id] = None  # mark as deleted
        # Update entry point if needed
        if self.entry_point == node_id:
            # Find a new entry point among remaining nodes
            remaining = [n for n in self.nodes if n is not None]
            if remaining:
                self.entry_point = remaining[0].id
            else:
                self.entry_point = None
                self.max_layer = -1

Step 5 – Neighbor Query (k‑NN)

    def knn(self, query: np.ndarray, k: int, ef: int = 100) -> List[Tuple[int, float]]:
        """Return the `k` nearest neighbors to `query`."""
        if self.entry_point is None:
            return []
        ep = self.entry_point
        for layer_idx in range(self.max_layer, 0, -1):
            ep = self._search_layer(query, ep, 1, layer_idx)[0][0]
        # Search bottom layer
        candidates = self._search_layer(query, ep, max(ef, k), 0)
        candidates.sort(key=lambda x: x[0])
        return candidates[:k]

Step 6 – Putting It Together

if __name__ == "__main__":
    # Example usage
    hnsw = HNSWGraph(M=16, ef_construction=200)
    # Generate random 128‑dimensional vectors
    for i in range(1000):
        vec = np.random.rand(128).astype(np.float32)
        hnsw.add(vec)

    # Query
    query_vec = np.random.rand(128).astype(np.float32)
    result = hnsw.knn(query_vec, k=5)
    print("Nearest neighbors:", result)

Running and Testing It

  1. Install dependencies

    pip install numpy

  2. Run the script

    python hnsw.py

  3. Validate correctness

    • Compare the returned neighbors against a brute‑force scipy.spatial.distance.cdist on a small dataset.
    • Use pytest to create unit tests for insertion, deletion, and query.

    Example test snippet:

    def test_basic_insert_and_query():
    g = HNSWGraph()
    v1 = np.array([1.0, 0.0])
    v2 = np.array([0.0, 1.0])
    g.add(v1)
    g.add(v2)
    res = g.knn(v1, k=1)
    assert res[0][0] == 0  # nearest is itself

  4. Performance benchmark

    • Time insertion of 10 k vectors and compare with faiss.IndexFlatL2.
    • Use timeit or hyperfine for repeatable measurements.

Extending It: Your Roadmap to Senior‑Level

  1. Persistence – Serialize the graph to disk using pickle or a custom binary format; on startup, deserialize to restore state. Why it matters: enables recovery after crashes and offline index building.

  2. Horizontal scaling – Shard the vector space across multiple processes or machines, each owning a partition of the HNSW graph; implement a coordinator that routes queries to relevant shards. Why it matters: allows handling billions of vectors beyond single‑node memory limits.

  3. Observability – Expose metrics (insertion latency, query QPS, layer distribution) via Prometheus; add structured logging with structlog. Why it matters: production systems require visibility to detect degradation and capacity planning.

  4. Fault tolerance – Replicate each node to a secondary replica; on primary failure, promote the replica automatically using a consensus protocol like Raft. Why it matters: guarantees high availability for search services.

  5. Benchmarking harness – Integrate with standard datasets (e.g., ANN‑Benchmarks) and generate latency‑throughput curves; compare against nmslib, faiss, and ScaNN. Why it matters: provides evidence of performance gains and guides tuning of M and ef_construction.

  6. Adaptive layer selection – Replace the static geometric distribution with a learned model that predicts optimal layer count based on vector density, reducing search latency. Why it matters: shows ability to apply ML insights to core algorithm design.

Key Takeaways

  • Implementing HNSW from scratch demonstrates deep understanding of graph‑based nearest‑neighbor search and probabilistic skip‑list concepts.
  • The project covers the full lifecycle: design, implementation, testing, and a clear path to production.
  • By adding persistence, scaling, observability, fault tolerance, and benchmarking, you transform a prototype into a showcase of senior‑level systems engineering.
  • Real‑world relevance: vector search powers recommendation, retrieval, and anomaly detection systems used at companies like Meta, Spotify, and Datadog.

Further Reading