Skip to main content

HNSW Vector Search with Filtering

SochDB provides high-performance vector search using HNSW (Hierarchical Navigable Small World) graphs, with support for session and timestamp filtering. This enables efficient semantic search over agent memory systems, chat histories, and other temporal data.

Which Python package?

The HnswIndex class shown here ships in the native PyO3 sochdb package (version 2.0.3) — a focused HNSW / BM25 / RRF / TableDatabase engine. The broad embedded + server SDK (Database, Namespace, Collection, AgentMemory, etc.) is a separate package, also imported as sochdb, at version 0.5.9. Both publish under the PyPI name sochdb but expose mostly disjoint APIs. For general database usage prefer the 0.5.9 SDK; for the standalone in-process HNSW engine used in this guide, install the native package and import HnswIndex from sochdb.

Overview​

The HNSW index provides:

  • O(log n) search complexity vs O(n) brute-force
  • ~250x speedup over linear scan for large datasets
  • Session isolation - search only within a specific session
  • Time-window filtering - retrieve recent memories only

For small datasets the index automatically falls back to an exact parallel SIMD flat scan, which is both correct and faster than graph traversal at low vector counts (see Automatic Flat-Scan Fallback).

Performance Comparison​

ObservationsBrute-Force P99HNSW P99Speedup
40143ms~30ms5x
2007,250ms~50ms145x
1,000~36,000ms~100ms360x
10,000N/A (timeout)~200ms∞

Python SDK Usage​

from sochdb import HnswIndex  # native sochdb 2.0.3 package
import numpy as np

# Create index with 1536 dimensions (text-embedding-3-small)
index = HnswIndex(
    dimension=1536,
    m=32,                    # Connections per node (default 32; quality vs memory)
    ef_construction=200,     # Construction quality (default 200; higher = better recall)
    metric="cosine",         # "cosine" | "euclidean"/"l2" | "dot"/"dot_product"
    precision="f32",         # "f32" | "f16" | "bf16"
)

# Insert vectors (zero-copy; requires C-contiguous float32)
embeddings = np.random.randn(1000, 1536).astype(np.float32)
index.insert_batch(embeddings)  # auto-generates sequential IDs, returns count

# Search for nearest neighbors
query = np.random.randn(1536).astype(np.float32)
ids, distances = index.search(query, k=10)  # ef_search defaults to adaptive

print(f"Found {len(ids)} nearest neighbors")
for id, dist in zip(ids, distances):
    print(f"  ID {id}: distance {dist:.4f}")
Native constructor defaults

The native HnswIndex constructor signature is HnswIndex(dimension, m=32, ef_construction=200, metric="cosine", precision="f32"). Internally max_connections_layer0 = m * 2 (so M0 = 64 at the default M = 32) and level_multiplier = 1 / ln(m). When you omit ef_search in search(...), the native module uses its adaptive search path rather than a fixed value.

For agent memory systems, you often need to search within a specific session and time window. This example combines the native HnswIndex with the 0.5.9 SDK's Database for metadata storage:

from sochdb import HnswIndex   # native sochdb 2.0.3
from sochdb import Database    # ctypes SDK 0.5.9 (separate package)
import numpy as np
import time
import json

class MemoryManager:
    """
    Manages agent memory with HNSW indexing for O(log n) search.
    """

    EMBEDDING_DIM = 1536

    def __init__(self, db_path: str):
        self.db = Database.open(db_path)
        self.hnsw_index = HnswIndex(
            dimension=self.EMBEDDING_DIM,
            m=32,
            ef_construction=200,
            metric="cosine",
        )
        self._id_to_key_map = {}
        self._next_id = 0
        self._rebuild_index()

    def _rebuild_index(self):
        """Load existing embeddings into HNSW index on startup."""
        results = self.db.scan_prefix(b"session.")

        embeddings = []
        for key, value in results:
            key_str = key.decode()
            if ".embedding" in key_str:
                turn_key = key_str.replace(".embedding", "")
                embedding = np.frombuffer(value, dtype=np.float32)

                if len(embedding) == self.EMBEDDING_DIM:
                    hnsw_id = self._next_id
                    self._next_id += 1
                    self._id_to_key_map[hnsw_id] = turn_key
                    embeddings.append((hnsw_id, embedding))

        if embeddings:
            ids = np.array([e[0] for e in embeddings], dtype=np.uint64)
            vectors = np.vstack([e[1] for e in embeddings]).astype(np.float32)
            self.hnsw_index.insert_batch_with_ids(ids, vectors)

    def store(self, session_id: str, content: str, embedding: np.ndarray):
        """Store a memory with its embedding."""
        turn = int(time.time() * 1000)  # Use timestamp as turn ID

        path = f"session.{session_id}.observations.turn_{turn}"

        # Store metadata
        metadata = {
            "session_id": session_id,
            "content": content,
            "timestamp": time.time(),
        }
        self.db.put(f"{path}.metadata".encode(), json.dumps(metadata).encode())
        self.db.put(f"{path}.embedding".encode(), embedding.tobytes())

        # Add to HNSW index
        hnsw_id = self._next_id
        self._next_id += 1
        self._id_to_key_map[hnsw_id] = path

        ids = np.array([hnsw_id], dtype=np.uint64)
        vectors = embedding.reshape(1, -1).astype(np.float32)
        self.hnsw_index.insert_batch_with_ids(ids, vectors)

    def search(
        self,
        session_id: str,
        query_embedding: np.ndarray,
        top_k: int = 10,
        hours: int = 24,
    ):
        """
        Search for similar memories with session and time filtering.

        Args:
            session_id: Only return results from this session
            query_embedding: Query vector
            top_k: Number of results to return
            hours: Time window in hours (default: 24)

        Returns:
            List of (content, similarity_score) tuples
        """
        # Over-fetch to allow for filtering
        hnsw_k = min(top_k * 3, len(self._id_to_key_map))

        if hnsw_k == 0:
            return []

        query = np.ascontiguousarray(query_embedding, dtype=np.float32)
        ids, distances = self.hnsw_index.search(query, k=hnsw_k)

        cutoff_time = time.time() - (hours * 3600)
        results = []

        for hnsw_id, distance in zip(ids, distances):
            turn_key = self._id_to_key_map.get(int(hnsw_id))
            if not turn_key:
                continue

            # Session filter
            if not turn_key.startswith(f"session.{session_id}"):
                continue

            # Load and check timestamp
            metadata_bytes = self.db.get(f"{turn_key}.metadata".encode())
            if not metadata_bytes:
                continue

            metadata = json.loads(metadata_bytes.decode())

            # Time window filter
            if metadata["timestamp"] < cutoff_time:
                continue

            # Convert cosine distance to similarity
            similarity = 1.0 - float(distance)
            results.append((metadata["content"], similarity))

            if len(results) >= top_k:
                break

        return results
Native metadata filtering

The native HnswIndex can also filter inside the index without a post-scan: attach key/value metadata with set_metadata(node_id, metadata) or set_metadata_batch(...), then call search_filtered(query, k, filter=[("session", "123")], ef_search=None). Filters use AND semantics; ef_search defaults to 200 on this path.

Usage Example​

import numpy as np

# Initialize
memory = MemoryManager("./agent_memory_db")

# Generate fake embedding (in practice, use OpenAI/Azure embeddings)
def get_embedding(text: str) -> np.ndarray:
    return np.random.randn(1536).astype(np.float32)

# Store memories
memory.store("session_123", "User asked about Python", get_embedding("Python question"))
memory.store("session_123", "Explained list comprehensions", get_embedding("list comprehension"))
memory.store("session_456", "Different session topic", get_embedding("other topic"))

# Search within session
query = get_embedding("How do I use list comprehensions?")
results = memory.search("session_123", query, top_k=5, hours=24)

for content, score in results:
    print(f"[{score:.3f}] {content}")

Letting SochDB pick parameters​

If you do not want to tune m / ef_construction by hand, the native package can recommend parameters from your dimension and (optionally) dataset size:

from sochdb import recommended_hnsw_params, build_index_from_numpy
import numpy as np

embeddings = np.random.randn(50_000, 768).astype(np.float32)

# recommended_hnsw_params(dimension, n_vectors=None, target_recall=0.95)
params = recommended_hnsw_params(768, n_vectors=50_000, target_recall=0.95)
print(params)  # -> {"m": ..., "ef_construction": ..., "ef_search": ..., "note": ...}

# build_index_from_numpy uses recommended_hnsw_params when m/ef are omitted
index = build_index_from_numpy(embeddings, metric="cosine")

The recommendation heuristic is:

  • M by dimension: dimension <= 128 → M = 16; 129–512 → M = 24; 513+ → M = 32.
  • ef_construction = max(200, M * 8).
  • ef_search by target recall: recall >= 0.99 → 40·M; >= 0.95 → 20·M; >= 0.90 → 10·M; otherwise 6·M.

JavaScript/TypeScript SDK​

The Node.js SDK exposes an embedded HnswIndex backed by the same engine over FFI. Its constructor and methods are synchronous (no await):

import { HnswIndex } from '@sochdb/sochdb';

// Create HNSW index (defaults match the engine: M=32, efConstruction=256)
const index = new HnswIndex({
  dimension: 1536,
  maxConnections: 32,   // optional, default 32
  efConstruction: 256,  // optional, default 256
  efSearch: 100,        // optional
});

// Insert vectors (synchronous): string IDs + parallel array of number[] vectors
index.insertBatch(
  ['doc-1', 'doc-2'],
  [vector1, vector2],
);

// Search (synchronous) -> SearchResult[] of { id: string, distance: number }
const results = index.search(queryVector, 10);
for (const { id, distance } of results) {
  console.log(`${id}: ${distance.toFixed(4)}`);
}

// Tune query depth at runtime
index.efSearch = 200;
Node defaults differ from the native Python module

The Node HnswIndex mirrors the engine's HnswConfig::default(): maxConnections (M) defaults to 32 and efConstruction to 256, with M0 = 64 and F32 precision inherited from the engine. insertBatch returns void and search returns a SearchResult[] directly — neither returns a Promise.

Configuration Parameters​

HNSW Parameters​

ParameterDefaultDescription
m (max_connections)32Max connections per node. Higher = better recall, more memory
m0 (max_connections_layer0)64Layer-0 connections; derived as 2 * m
ef_construction256 (core) / 200 (native module)Construction-time search depth. Higher = better index quality
ef_search500 (core engine)Query-time search depth. Higher = better recall, slower
metric"cosine"Distance metric: "cosine", "euclidean"/"l2", "dot"
precision"f32"Quantization precision: "f32", "f16", "bf16"
Defaults at a glance

The core engine HnswConfig::default() is M = 32, M0 = 64, ef_construction = 256, ef_search = 500, metric = Cosine, precision = F32. The native Python HnswIndex constructor defaults ef_construction to 200 (a slightly lighter build), and when you call search(...) without an explicit ef_search it uses an adaptive search path instead of a single fixed value. There is no dimension-dependent ef_search value baked into the engine — the only dimension-aware switch is the flat-scan threshold below.

These map to recommended_hnsw_params (see above):

Use Casemef_constructionef_search
Low-dim (<= 128), recall ~0.9516200320
Mid-dim (129–512), recall ~0.9524200480
High-dim (513+), recall ~0.9532256640
Latency-first (recall ~0.90, M=16)16200160
note

These rows are computed from the recommendation heuristic (ef_construction = max(200, M*8), ef_search = 20*M at recall 0.95, 10*M at recall 0.90). Treat them as starting points and verify recall on your own data.

Automatic Flat-Scan Fallback​

For small indexes, exact brute-force is both correct and faster than HNSW traversal, so the engine automatically uses a parallel SIMD flat scan below a dimension-aware vector-count threshold:

flat_scan_threshold = if dimension <= 128  -> 10_000 vectors
                      else if dimension <= 384 -> 4_000 vectors
                      else                  -> 1_000 vectors  (768D+)

When the number of vectors is at or below the threshold for its dimension, the native HnswIndex automatically runs an exact scan over rayon-parallel SIMD chunks — giving exact results with no recall loss — without you calling a separate method. Above the threshold, search uses the HNSW graph. There is no separate ef_search ramp (e.g. a 500/1500 split) keyed on dimension in the engine; the dimension-aware behavior is this flat-scan crossover.

If you need an explicit exact-search API, the ctypes SDK's VectorIndex (the 0.5.9 package) exposes search_exact(...) / search_exact_f64(...) for ground-truth recall checks:

# ctypes SDK VectorIndex (sochdb 0.5.9): explicit exact brute-force k-NN
from sochdb import VectorIndex

vindex = VectorIndex(dimension=1536)
# ... insert vectors ...
exact = vindex.search_exact(query, k=10)   # returns list[(id, distance)]

At the engine level, the smart chooser search_smart(query, k, exact_threshold) (default exact_threshold = 1000) calls exact search below the threshold and adaptive HNSW above it; the native HnswIndex applies the equivalent flat-scan crossover automatically.

Index Maintenance & Diagnostics​

The native HnswIndex exposes graph-maintenance and health methods you can run after large ingests or to recover recall:

# Additive neighbor refinement: fills empty connection slots,
# never removes existing edges. Returns the number of nodes improved.
improved = index.refine_graph_additive()

# Full refinement pass (re-evaluates neighbors).
index.refine_graph()

# Rebuild layer-0 with exact brute-force k-NN (small/medium indexes only;
# ~0.3s per 10K vectors). Best recall boost when the index is not huge.
index.optimize()

# Reconnect orphaned nodes via BFS.
index.repair()

# Connectivity / degree diagnostics.
report = index.diagnose()
# -> {reachable, total, orphan_count, avg_degree, zero_degree_nodes, target_degree}
print(report)
optimize() is for small/medium indexes

optimize() performs an exact O(N²·D) layer-0 rebuild. It is a strong recall booster for small and medium indexes, but it does not scale to very large indexes (a multi-million-vector exact rebuild is prohibitively expensive). For large builds, rely on the default graph quality plus refine_graph_additive() rather than optimize().

Best Practices​

1. Batch Insertions​

# Good: Batch insert
index.insert_batch(embeddings)  # ~15,000 vec/s

# Avoid: One-by-one insert
for emb in embeddings:
    index.insert_batch(emb.reshape(1, -1))  # much slower

2. Contiguous Arrays​

# Good: Contiguous float32 array (insert_batch is zero-copy on these)
embeddings = np.ascontiguousarray(data, dtype=np.float32)
index.insert_batch(embeddings)

# Avoid: Non-contiguous or wrong dtype (raises ValueError)
embeddings = some_list  # Requires copy / conversion

3. Warm-up Searches​

# First search may be slower due to memory allocation
# Warm up the index with a dummy query
_ = index.search(np.zeros(1536, dtype=np.float32), k=1)

# Now benchmark real queries

4. Session-Based Sharding​

For multi-tenant systems, consider one index per tenant:

class MultiTenantMemory:
    def __init__(self):
        self.indices = {}  # tenant_id -> HnswIndex

    def get_index(self, tenant_id: str) -> HnswIndex:
        if tenant_id not in self.indices:
            self.indices[tenant_id] = HnswIndex(dimension=1536)
        return self.indices[tenant_id]
Very large fan-out (100M–1B vectors)

The native package also ships a MultiShardHnswIndex for extreme scale. It is a pure-Python threaded scatter-gather wrapper that routes vectors to per-shard HnswIndex instances by id % n_shards and merges results — it is not a core-engine type. Reach for it only when a single index no longer fits; for most workloads one HnswIndex (or one per tenant, above) is simpler and faster.

Troubleshooting​

High Latency at Scale​

If P99 latency exceeds 100ms for 1000+ vectors:

  1. Check index type: Ensure you are using HnswIndex, not an exact path
  2. Reduce ef_search: Lower values = faster but less accurate (pass ef_search to search or set efSearch in Node)
  3. Use batched queries: search_batch() for multiple queries at once

Low Recall​

If relevant results are missing:

  1. Increase ef_search: Higher values improve recall
  2. Run refine_graph_additive() (or optimize() on small indexes) after a large ingest
  3. Check embedding quality: Ensure embeddings are normalized
  4. Verify metric: Use cosine for text embeddings

Memory Usage​

For 1M 1536-dim vectors:

  • Full precision (f32): ~6GB
  • Half precision (f16): ~3GB
  • BF16 quantization: ~3GB
# Enable quantization for memory savings
index = HnswIndex(dimension=1536, precision="f16")  # or "bf16"

See Also​