Sign in Subscribe
python

Busca por Similaridade com HNSW: Vetores em Microsegundos

user

• 4 min read

Você tem 500 mundos de conhecimento, cada um representado por um vetor de 384 dimensões. O usuário envia uma mensagem e você precisa encontrar qual mundo é mais relevante. A abordagem ingênua é calcular cosine similarity com todos os 500 e pegar o maior. Funciona, mas é O(n). Com 10.000 mundos, são 10.000 multiplicações de vetores por mensagem. Com HNSW, a mesma busca leva O(log n) e retorna resultados praticamente idênticos ao brute-force.

Implementei um índice HNSW persistente com suporte a inserção dinâmica, remoção via tombstones, auto-save periódico e compactação automática. O resultado é busca por similaridade em microsegundos que escala sem degradar, com persistência em disco para não perder o índice entre restarts.

Guia de tópicos:

  • O Problema da Busca Linear
  • Como HNSW Funciona (Intuição sem Matemática)
  • Implementação Persistente
  • Tombstones e Compactação
  • Threshold Dinâmico: Não é Só Encontrar o Mais Próximo
  • Exemplo Prático em Python
  • Considerações Finais

O Problema da Busca Linear

Busca linear funciona perfeitamente para poucos vetores. Com 50 mundos e vetores de 384 dimensões, são 50 dot products que levam microsegundos. Mas o custo cresce linearmente: 500 mundos = 10x mais lento, 5000 = 100x.

Para um sistema que roda por meses acumulando conhecimento, o número de mundos cresce indefinidamente. Sem índice eficiente, a latência de cada mensagem aumentaria com o tempo. O usuário não percebe a diferença entre 1ms e 5ms, mas percebe entre 5ms e 500ms.

Como HNSW Funciona (Intuição sem Matemática)

Imagina uma cidade com bairros. Em vez de visitar todas as casas para encontrar a mais parecida com a sua, você primeiro vai ao bairro certo (camada superior, poucos nós, visão global), depois à rua certa (camada intermediária), e finalmente à casa certa (camada inferior, todos os nós). Cada salto te aproxima do alvo sem visitar todos.

HNSW constrói esse grafo em múltiplas camadas durante a inserção. Na busca, você começa no topo e desce, sempre seguindo vizinhos mais próximos do alvo. O resultado é busca aproximada com recall > 99% na prática.

Implementação Persistente

O índice é salvo em disco (arquivo binário + JSON de mapeamento IDs). Na inicialização, se os arquivos existem, carrega. Se não, cria do zero. Auto-save a cada 1000 inserções para não perder dados em crash.

Capacidade cresce dinamicamente: começa com 256 slots e dobra quando necessário (resize_index). Isso é amortizado: a maioria das inserções não causa resize.

Tombstones e Compactação

HNSW não suporta remoção eficiente nativa. A solução é marcar slots como deletados (tombstones) e manter uma lista de slots livres para reutilização. Quando tombstones excedem 40% dos slots, o índice é reconstruído do zero com apenas os vetores vivos. Mantém performance ótima a longo prazo.

Threshold Dinâmico: Não é Só Encontrar o Mais Próximo

Encontrar o mundo mais similar não é suficiente. Se o mais similar tem 0.3 de cosine, provavelmente não é relevante. O sistema usa threshold dinâmico: 0.60 + maturidade × 0.10. Mundos maduros exigem mais similaridade. Se nenhum passa, cria mundo novo.

Exemplo Prático em Python

import numpy as np
import json
import os

try:
    import hnswlib
    HAS_HNSW = True
except ImportError:
    HAS_HNSW = False


class PersistentVectorIndex:
    """HNSW index with persistence, tombstones, and auto-compaction."""

    FUSION_THRESHOLD = 0.94
    COMPACTION_RATIO = 0.40

    def __init__(self, dim=384, path="./index"):
        self.dim = dim
        self._path = path
        self._id_to_slot = {}
        self._slot_to_id = []
        self._tombstones = 0
        self._index = None

        if HAS_HNSW:
            self._index = hnswlib.Index(space="ip", dim=dim)
            self._index.init_index(max_elements=256, ef_construction=200, M=16,
                                   allow_replace_deleted=True)
            self._index.set_ef(50)

    @staticmethod
    def _normalize(v):
        v = np.array(v, dtype=np.float32)
        n = np.linalg.norm(v)
        return v / n if n > 0 else v

    def add(self, item_id: str, vector):
        vec = self._normalize(vector)

        if item_id in self._id_to_slot:
            # Update existing
            slot = self._id_to_slot[item_id]
            if self._index:
                self._index.mark_deleted(slot)
                self._index.add_items(vec.reshape(1, -1), [slot], replace_deleted=True)
        else:
            # New item
            slot = len(self._slot_to_id)
            self._slot_to_id.append(item_id)
            self._id_to_slot[item_id] = slot
            if self._index:
                if slot >= self._index.get_max_elements():
                    self._index.resize_index(slot * 2)
                self._index.add_items(vec.reshape(1, -1), [slot])

    def remove(self, item_id: str):
        if item_id not in self._id_to_slot:
            return
        slot = self._id_to_slot.pop(item_id)
        self._slot_to_id[slot] = None
        self._tombstones += 1
        if self._index:
            self._index.mark_deleted(slot)

    def search(self, query_vector, top_k=5) -> list[tuple[str, float]]:
        vec = self._normalize(query_vector)

        if self._index and self._id_to_slot:
            k = min(top_k, len(self._id_to_slot))
            labels, distances = self._index.knn_query(vec.reshape(1, -1), k=k)
            results = []
            for lbl, dist in zip(labels[0], distances[0]):
                if lbl < len(self._slot_to_id) and self._slot_to_id[lbl]:
                    results.append((self._slot_to_id[lbl], float(1.0 - dist)))
            return results

        # Fallback: linear scan
        results = []
        for item_id, slot in self._id_to_slot.items():
            # Would need stored vectors for fallback
            pass
        return results

    def needs_compaction(self) -> bool:
        total = len(self._slot_to_id)
        return total > 0 and (self._tombstones / total) > self.COMPACTION_RATIO

    def find_fusion_pairs(self, vectors: dict, threshold=None) -> list[tuple]:
        """Find pairs of items similar enough to merge."""
        threshold = threshold or self.FUSION_THRESHOLD
        pairs = []
        for item_id, vec in vectors.items():
            results = self.search(vec, top_k=3)
            for other_id, sim in results:
                if other_id != item_id and sim > threshold:
                    pair = tuple(sorted([item_id, other_id]))
                    if pair not in pairs:
                        pairs.append(pair)
        return pairs


# Demo
index = PersistentVectorIndex(dim=128)

# Add some worlds
for i in range(50):
    index.add(f"world_{i}", np.random.randn(128))

# Search
query = np.random.randn(128)
results = index.search(query, top_k=3)
print("Top 3 nearest worlds:")
for item_id, similarity in results:
    print(f"  {item_id}: sim={similarity:.4f}")

# Remove and check compaction
for i in range(25):
    index.remove(f"world_{i}")
print(f"\\nTombstones: {index._tombstones}/{len(index._slot_to_id)}")
print(f"Needs compaction: {index.needs_compaction()}")

Considerações Finais

HNSW é a estrutura de dados que permite escalar busca por similaridade sem sacrificar latência. Para qualquer sistema que precisa de "encontre os N mais similares" com baixa latência (recomendação, RAG, deduplicação, classificação), HNSW é provavelmente a resposta se você tem mais de 100 vetores.

A implementação com persistência, tombstones e compactação garante que o índice se mantém eficiente a longo prazo sem intervenção manual. É infraestrutura que você configura uma vez e esquece.

Links indicativos:

Sign up for more like this.

Enter your email
Subscribe