Banco vetorial na prática com pgvector e Postgres

O problema real#

Você leu que precisa de um banco vetorial e saiu comparando Pinecone, Weaviate, Milvus e Qdrant. Escolheu um, subiu infraestrutura nova, ganhou mais um serviço para monitorar, pagar e fazer backup. Algum tempo depois percebe que o Postgres que já roda em produção resolveria o seu caso com a extensão pgvector.

Um banco dedicado faz sentido em certos cenários, mas muita aplicação de RAG nunca chega neles. Neste artigo você vai montar um retriever completo dentro do Postgres, com índice HNSW, busca por similaridade, filtros e busca híbrida, e vai saber que sinais indicam a hora de migrar.

Por que o pgvector cobre tantos casos#

Um banco vetorial precisa fazer três coisas. Guardar vetores com índice, buscar por similaridade (cosseno, distância euclidiana ou produto interno) e filtrar por metadados durante a busca. O pgvector faz as três e ainda entrega o que o Postgres já tem, como transações, SQL, joins, backup e permissões. Seus embeddings ficam na mesma transação que o resto dos dados, o que simplifica muito casos como apagar os vetores de um cliente que pediu exclusão por LGPD.

O desempenho depende do tamanho do índice, das dimensões, do hardware e dos parâmetros. Em vez de confiar em um número de blog, meça no seu corpus, com a sua máquina, usando os mesmos vetores que usará em produção.

Mão na massa#

Subir um Postgres com pgvector#

A forma mais rápida é a imagem oficial do projeto, que já traz a extensão.

docker run -d --name pg-vetorial \
  -e POSTGRES_PASSWORD=senha_de_teste \
  -p 5432:5432 pgvector/pgvector:pg17

docker exec -it pg-vetorial psql -U postgres -c "CREATE EXTENSION IF NOT EXISTS vector;"

Se você usa Postgres gerenciado, confira na documentação do provedor se a extensão está disponível. Para instalar as dependências Python, use pip install "psycopg[binary]" pgvector openai numpy.

Tabela e índices#

CREATE TABLE document_chunks (
    id          BIGSERIAL PRIMARY KEY,
    document_id INT NOT NULL,
    chunk_index INT NOT NULL,
    content     TEXT NOT NULL,
    embedding   vector(1536) NOT NULL,   -- precisa bater com o modelo de embedding
    metadata    JSONB NOT NULL DEFAULT '{}',
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX ON document_chunks
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

CREATE INDEX ON document_chunks USING GIN (metadata);
CREATE INDEX ON document_chunks USING GIN (to_tsvector('portuguese', content));

O número de dimensões da coluna tem de ser o mesmo do modelo (1.536 no text-embedding-3-small, por padrão). O operador do índice precisa combinar com o operador da consulta, por isso vector_cosine_ops casa com <=>. O terceiro índice acelera a busca textual que usaremos na versão híbrida.

Inserir chunks com embeddings#

Usamos o driver psycopg (versão 3) com o pacote pgvector, que registra o tipo vetorial e converte arrays do NumPy.

import numpy as np
import psycopg
from psycopg.types.json import Jsonb
from pgvector.psycopg import register_vector
from openai import OpenAI

DSN = "postgresql://postgres:senha_de_teste@localhost:5432/postgres"
ai = OpenAI()

def embed(textos: list[str]) -> list[np.ndarray]:
    resp = ai.embeddings.create(model="text-embedding-3-small", input=textos)
    return [np.array(d.embedding) for d in resp.data]

def inserir(document_id: int, chunks: list[str], extra: dict | None = None) -> None:
    vetores = embed(chunks)
    with psycopg.connect(DSN) as conn:
        register_vector(conn)
        with conn.cursor() as cur:
            cur.executemany(
                """INSERT INTO document_chunks
                   (document_id, chunk_index, content, embedding, metadata)
                   VALUES (%s, %s, %s, %s, %s)""",
                [(document_id, i, c, v, Jsonb(extra or {}))
                 for i, (c, v) in enumerate(zip(chunks, vetores))],
            )

inserir(1, [
    "A Constituição Federal foi promulgada em 1988.",
    "O Presidente da República é eleito por voto direto.",
    "O Congresso Nacional é composto pela Câmara dos Deputados e pelo Senado Federal.",
], {"ano": "1988"})

Em produção, leia a senha de uma variável de ambiente e envie os textos em lotes para a API de embeddings, respeitando o limite de itens por requisição do seu provedor.

Busca por similaridade#

def buscar(pergunta: str, k: int = 3) -> list[dict]:
    q = embed([pergunta])[0]
    with psycopg.connect(DSN) as conn:
        register_vector(conn)
        rows = conn.execute(
            """SELECT id, content, metadata, 1 - (embedding <=> %s) AS similaridade
               FROM document_chunks
               ORDER BY embedding <=> %s
               LIMIT %s""",
            (q, q, k),
        ).fetchall()
    return [{"id": r[0], "content": r[1], "metadata": r[2], "score": r[3]} for r in rows]

for r in buscar("Qual é o sistema de governo?"):
    print(f"{r['score']:.3f} | {r['content'][:70]}")

O índice HNSW é aproximado, então o recall depende do parâmetro hnsw.ef_search, que vale 40 por padrão. Se perceber resultados faltando, aumente na sessão com SET hnsw.ef_search = 100; e compare a latência.

Filtros de metadados#

Filtrar e buscar ao mesmo tempo é onde índices aproximados costumam dar problema, porque o filtro pode eliminar candidatos depois que o índice já escolheu. As versões recentes do pgvector têm varredura iterativa de índice para reduzir isso (SET hnsw.iterative_scan = relaxed_order;). Confira na documentação a versão que você está usando.

CHAVES_PERMITIDAS = {"ano", "idioma", "tenant"}

def buscar_filtrado(pergunta: str, filtros: dict[str, str], k: int = 3) -> list[tuple]:
    if not set(filtros) <= CHAVES_PERMITIDAS:
        raise ValueError("filtro não permitido")
    q = embed([pergunta])[0]
    where = " AND ".join("metadata->>%s::text = %s" for _ in filtros) or "TRUE"
    params = [x for kv in filtros.items() for x in kv]
    with psycopg.connect(DSN) as conn:
        register_vector(conn)
        return conn.execute(
            f"""SELECT id, content FROM document_chunks
                WHERE {where}
                ORDER BY embedding <=> %s LIMIT %s""",
            (*params, q, k),
        ).fetchall()

Repare que o nome da chave também vai como parâmetro e que há uma lista de chaves permitidas. Isso evita injeção de SQL quando o filtro vem de entrada externa.

Busca híbrida#

Busca vetorial acerta o sentido, mas escorrega em siglas, nomes e códigos exatos. A busca textual faz o contrário. Combinar as duas costuma melhorar o recall. Uma forma robusta é a fusão por posição (Reciprocal Rank Fusion), que soma 1/(60 + posição) de cada lista e não depende de calibrar notas de escalas diferentes.

WITH vetorial AS (
    SELECT id, ROW_NUMBER() OVER (ORDER BY embedding <=> %(q)s) AS pos
    FROM document_chunks
    ORDER BY embedding <=> %(q)s
    LIMIT 50
),
textual AS (
    SELECT id, ROW_NUMBER() OVER (ORDER BY rank DESC) AS pos
    FROM (
        SELECT id, ts_rank(to_tsvector('portuguese', content),
                           plainto_tsquery('portuguese', %(texto)s)) AS rank
        FROM document_chunks
        WHERE to_tsvector('portuguese', content) @@ plainto_tsquery('portuguese', %(texto)s)
        ORDER BY rank DESC
        LIMIT 50
    ) t
)
SELECT c.id, c.content,
       COALESCE(1.0 / (60 + v.pos), 0) + COALESCE(1.0 / (60 + t.pos), 0) AS score
FROM document_chunks c
LEFT JOIN vetorial v ON v.id = c.id
LEFT JOIN textual  t ON t.id = c.id
WHERE v.id IS NOT NULL OR t.id IS NOT NULL
ORDER BY score DESC
LIMIT 5;

Execute com conn.execute(sql, {"q": vetor, "texto": pergunta}). Dá para dar mais peso a uma das listas multiplicando os termos, mas comece sem pesos e meça.

Armadilhas comuns#

HNSW ou IVFFlat#

O HNSW costuma dar melhor equilíbrio entre velocidade e recall, mas o índice consome mais memória e demora mais para ser construído. O IVFFlat constrói mais rápido e ocupa menos, porém exige dados já carregados antes de criar o índice e ajuste de lists e probes. Comece com HNSW e só considere o outro se memória ou tempo de construção virarem problema.

Trocar de modelo de embedding sem reindexar#

Vetores de modelos diferentes vivem em espaços diferentes, e até a dimensão pode mudar. Ao trocar de modelo, recrie a coluna e reembede tudo. Guardar o nome do modelo no metadata ajuda a migrar aos poucos.

Esquecer da manutenção#

Deletes e updates geram linhas mortas. O autovacuum cuida disso na maior parte do tempo, e depois de uma carga ou exclusão em massa vale rodar VACUUM ANALYZE document_chunks;. Índices HNSW também podem ser reconstruídos com REINDEX se a qualidade cair.

Quando considerar um banco dedicado#

• Volume muito acima do que cabe confortavelmente na memória do seu Postgres, e você já mediu isso.
• Carga de busca vetorial pesada o bastante para competir com o banco transacional. Nesse caso, uma réplica de leitura dedicada às buscas pode resolver antes de migrar.
• Necessidade de recursos específicos, como multi-tenancy com isolamento forte e muita escala, ou filtros complexos sobre grandes volumes.

Se a decisão for por um serviço dedicado, compare latência, recall e custo com os seus próprios dados, e consulte a página de preços de cada fornecedor no momento da decisão.

Próximos passos#

Troque o Chroma do artigo RAG na prática pelo pgvector e acompanhe o custo da infraestrutura em FinOps para IA. Para medir se a busca realmente melhorou, use evals, como em Avaliação de LLM com evals.

Quer aplicar isso na sua empresa? Marque uma conversa de 45 minutos em https://iauaicloud.com.br/consultoria